Commit aff17010 authored by Kantz's avatar Kantz
Browse files

Artefakte der Tools entfernt

parent cacb3882
"""Deprecated legacy decision LLM module.
The active tutor orchestrator no longer calls this module. Keep it only for
manual legacy tests until it can be removed.
"""
import warnings
from app.deterministic_services import llm_client
warnings.warn(
"app.LLM_services.decision_LLM is deprecated and no longer used by the tutor orchestrator.",
DeprecationWarning,
stacklevel=2,
)
def context_decision(needs_more_context: bool, reason: str) -> dict:
# Kannst auch einfach nur return {"needs_more_context": needs_more_context, "reason": reason}
return {"needs_more_context": bool(needs_more_context), "reason": str(reason)}
CLASSIFIER_SYSTEM = """
Du bist ein Klassifikator für didaktische Tutoring-Hinweise.
Aufgabe:
Entscheide, ob Historie + Kontextblatt ausreichen, um den NÄCHSTEN didaktisch wertvollen Hinweis zu geben.
Wenn irgendetwas Wesentliches fehlt, ist needs_more_context=true.
Der Kontext ist ausreichend wenn:
1. Die aktuelle Problemstellung und Lösungsidee klar ist.
2. aktueller Lösungsstand klar ist? (was wurde schon probiert)
3. der typischer nächster Schritt ableitbar ist.
4. es keine fehlen Variablen/Definitionen/Teilaufgabe gibt.
WICHTIG:
- Antworte NICHT mit freiem Text.
- Rufe IMMER das Tool context_decision auf.
- needs_more_context ist boolean.
- reason ist eine kurze Begründung (1–3 Sätze), konkret was fehlt oder warum es reicht.
- Löse nicht die Aufgabe, sondern bewerte nur die Situation bzgl. der nächsten didaktischen Schritte.
"""
def needs_more_context(context_sheet: str) -> dict:
return {"needs_more_context": False, "reason":"Deactiviert weil nicht Funktional"}
messages = [{"role": "system", "content": CLASSIFIER_SYSTEM}]
messages.append(
{
"role": "user",
"content": "Kontextblatt:\n"
f"{context_sheet}\n\n Entscheide nur über den Kontext und nutze context_decision."
}
)
resp, tool_outputs = llm_client.chat_with_tools(
messages=messages,
tools=[context_decision],
use_ollama=True,
return_after_tools=True,
)
if tool_outputs:
result = tool_outputs[0].get("result")
if isinstance(result, dict) and "needs_more_context" in result:
return result
return {"needs_more_context": False, "reason": "needs_more_context wurde nicht gesetzt"+str(result)}
return {"needs_more_context": False, "reason": "No tool_call returned"+str(resp)}
"""Deprecated legacy math-intent LLM module.
The active tutor orchestrator no longer calls this module. Keep it only for
manual legacy tests until it can be removed.
"""
import warnings
import sympy as sp
from app.deterministic_services import llm_client
warnings.warn(
"app.LLM_services.math_intent_LLM is deprecated and no longer used by the tutor orchestrator.",
DeprecationWarning,
stacklevel=2,
)
def sympy_solve(task: str, input: str, symbols: list[str] | None = None) -> str:
"""
task: simplify|solve|diff|integrate
input: SymPy-Expression als String, z.B. '((1/a)+a)/(a+1) - (a-1)/(a+1)'
symbols: z.B. ['a']
"""
symbols = symbols or []
locals_map = {name: sp.Symbol(name) for name in symbols}
expr = sp.sympify(input, locals=locals_map)
if task == "simplify":
return str(sp.simplify(expr))
if task == "diff":
if not symbols:
raise ValueError("diff requires symbols[0]")
return str(sp.diff(expr, locals_map[symbols[0]]))
if task == "integrate":
if not symbols:
raise ValueError("integrate requires symbols[0]")
return str(sp.integrate(expr, locals_map[symbols[0]]))
if task == "solve":
if not symbols:
raise ValueError("solve requires symbols[0]")
return str(sp.solve(sp.Eq(expr, 0), locals_map[symbols[0]]))
raise ValueError(f"Unknown task: {task}")
SYSTEM = """
Du bist ein Mathe-Assistent.
Wenn es nichts zu berechnen gibt Rechung gibt gib "None" aus.
Wenn eine Rechnung nötig ist, nutze das Tool sympy_solve.
Wichtig: Übergib in input eine gültige SymPy-Expression (kein LaTeX).
Antworte nach folgendem Muster:
"Rechung: [hier soll die Rechung stehen ohne Klammern]
Lösung: [Hier soll die Lösung stehen ohne Klammern]"
ODER
"Keine Lösung"
"""
def solve_with_tools(user_text: str) -> str:
return "none"
messages = [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": user_text + "\nAntworte nur mit 'Keine Lösung' falls es in dieser Nachricht KEINE Berechung gibt."},
]
result = llm_client.chat(
messages=messages,
use_ollama=True,
)
return llm_client.get_message_content(result)
import inspect
import json
import warnings
from datetime import date
from typing import Any, Callable
......@@ -12,6 +10,10 @@ from app import config
from app.deterministic_services import llm_quota
# ---
# Response helpers
# ---
def _filter_kwargs(func, kwargs: dict) -> dict:
try:
signature = inspect.signature(func)
......@@ -53,6 +55,10 @@ def _extract_total_tokens(response: object) -> int:
return int(prompt) + int(completion)
# ---
# Quota
# ---
def _record_call(result: dict, tokens: int | None = None) -> dict:
pg_url = config.get_postgres_url()
token_count = _extract_total_tokens(result.get("raw")) if tokens is None else tokens
......@@ -60,24 +66,6 @@ def _record_call(result: dict, tokens: int | None = None) -> dict:
return result
def _warn_deprecated_provider_flags(use_ollama: bool, use_mistral: bool) -> None:
if use_ollama or use_mistral:
warnings.warn(
"use_ollama and use_mistral are deprecated and ignored. "
"Set LLM_PROVIDER in the environment instead.",
DeprecationWarning,
stacklevel=3,
)
def _warn_deprecated_tools() -> None:
warnings.warn(
"LLM toolcalling via llm_client.chat(..., tools=...) is deprecated.",
DeprecationWarning,
stacklevel=3,
)
def _ensure_within_llm_quota() -> None:
quota_settings = config.get_llm_quota_settings()
llm_quota.ensure_within_limits(
......@@ -103,6 +91,10 @@ def _quota_tracked_chat(chat_func: Callable[[], dict]) -> dict:
return result
# ---
# Provider settings
# ---
def _require_openai_chat_settings() -> config.OpenAIChatSettings:
settings = config.get_openai_chat_settings()
if not settings:
......@@ -132,6 +124,10 @@ def _require_mistral_chat_settings() -> config.MistralChatSettings:
return settings
# ---
# API calls
# ---
def _chat_openai_compatible(
messages: list[dict],
settings: config.OpenAIChatSettings | None = None,
......@@ -173,30 +169,25 @@ def _chat_mistral(
def _chat_ollama(
messages: list[dict],
tools: list[Callable[..., Any]] | None = None,
) -> dict:
settings = config.get_ollama_settings()
client = ollama.Client(host=settings.base_url, timeout=settings.timeout)
kwargs: dict = {"model": settings.model, "messages": messages}
if tools:
kwargs["tools"] = tools
if settings.keepalive:
kwargs["keep_alive"] = settings.keepalive
if settings.temperature is not None:
kwargs["options"] = {"temperature": settings.temperature}
filtered_kwargs = _filter_kwargs(client.chat, kwargs)
if tools and "tools" not in filtered_kwargs:
raise RuntimeError(
"Configured ollama Python client does not support tool calling (chat(..., tools=...)). "
"Update the 'ollama' package or switch to a backend/model path with tool support."
)
response = client.chat(**filtered_kwargs)
return {"raw": response, "message": _extract_message(response)}
# ---
# Public chat API
# ---
def get_message_content(result: dict | object) -> str:
message = result.get("message") if isinstance(result, dict) else result
if isinstance(message, dict):
......@@ -208,21 +199,8 @@ def get_message_content(result: dict | object) -> str:
def chat(
messages: list[dict],
tools: list[Callable[..., Any]] | None = None,
use_ollama: bool = False,
use_mistral: bool = False,
) -> dict:
if tools:
_warn_deprecated_tools()
_warn_deprecated_provider_flags(use_ollama, use_mistral)
provider = config.get_llm_provider()
if tools and provider != "ollama":
raise RuntimeError(
"Deprecated LLM toolcalling is only implemented for LLM_PROVIDER=ollama. "
f"Current LLM_PROVIDER={provider}."
)
if provider == "openai":
settings = _require_openai_chat_settings()
return _quota_tracked_chat(lambda: _chat_openai_compatible(messages, settings))
......@@ -233,123 +211,5 @@ def chat(
settings = _require_mistral_chat_settings()
return _quota_tracked_chat(lambda: _chat_mistral(messages, settings))
if provider == "ollama":
return _chat_ollama(messages, tools=tools)
return _chat_ollama(messages)
raise ValueError(f"Unsupported LLM_PROVIDER: {provider}")
def chat_with_tools(
messages: list[dict],
tools: list[Callable[..., Any]],
use_ollama: bool = True,
use_mistral: bool = False,
return_after_tools: bool = False,
) -> tuple[dict, list[dict[str, Any]]]:
"""Deprecated legacy wrapper around model tool calls."""
warnings.warn(
"llm_client.chat_with_tools(...) is deprecated.",
DeprecationWarning,
stacklevel=2,
)
tool_map = {tool.__name__: tool for tool in tools}
result = chat(
messages=messages,
tools=tools,
use_ollama=use_ollama,
use_mistral=use_mistral,
)
tool_outputs = _apply_tool_calls(result, messages, tool_map)
if not tool_outputs or return_after_tools:
return result, tool_outputs
final_result = chat(
messages=messages,
tools=tools,
use_ollama=use_ollama,
use_mistral=use_mistral,
)
return final_result, tool_outputs
def _apply_tool_calls(
result: dict | object,
messages: list[dict],
tool_map: dict[str, Callable[..., Any]],
) -> list[dict[str, Any]]:
"""Deprecated legacy helper for chat_with_tools."""
message = result.get("message") if isinstance(result, dict) else result
tool_calls = _extract_tool_calls(message)
outputs: list[dict[str, Any]] = []
if tool_calls:
messages.append(_message_to_dict(message))
for call in tool_calls:
name, arguments = _tool_call_name_args(call)
tool = tool_map.get(name)
if not tool:
output = f"Unknown tool: {name}"
else:
try:
output = tool(**arguments)
except Exception as exc: # pragma: no cover - defensive
output = f"Tool error: {exc}"
outputs.append(
{"name": name, "arguments": arguments, "result": output})
messages.append(
{"role": "tool", "tool_name": name, "content": str(output)})
return outputs
def _extract_tool_calls(message: object) -> list:
"""Deprecated legacy helper for chat_with_tools."""
if isinstance(message, dict):
return message.get("tool_calls") or []
if hasattr(message, "tool_calls"):
return getattr(message, "tool_calls") or []
return []
def _tool_call_name_args(call: object) -> tuple[str, dict[str, Any]]:
"""Deprecated legacy helper for chat_with_tools."""
if isinstance(call, dict):
function = call.get("function") or {}
name = function.get("name") or ""
arguments = function.get("arguments")
else:
function = getattr(call, "function", None)
name = getattr(function, "name", "") if function else ""
arguments = getattr(function, "arguments", None) if function else None
return name, _parse_tool_arguments(arguments)
def _parse_tool_arguments(arguments: object) -> dict[str, Any]:
"""Deprecated legacy helper for chat_with_tools."""
if isinstance(arguments, dict):
return arguments
if isinstance(arguments, str) and arguments.strip():
try:
parsed = json.loads(arguments)
if isinstance(parsed, dict):
return parsed
except json.JSONDecodeError:
return {}
return {}
def _message_to_dict(message: object) -> dict[str, Any]:
if isinstance(message, dict):
return message
role = getattr(message, "role", None)
content = getattr(message, "content", None)
tool_calls = getattr(message, "tool_calls", None)
payload: dict[str, Any] = {}
if role is not None:
payload["role"] = role
if content is not None:
payload["content"] = content
if tool_calls is not None:
payload["tool_calls"] = tool_calls
return payload
import argparse
import json
import os
from typing import Any
os.environ.setdefault("EMBEDDING_PROVIDER", "sentence-transformer")
os.environ.setdefault("EMBEDDING_TYPE", "sentence-transformer")
from app.LLM_services import decision_LLM
from app.deterministic_services import context_store
def _load_sheet_from_path(path: str) -> dict[str, Any]:
with open(path, "r", encoding="utf-8") as handle:
return json.load(handle)
def _resolve_sheet(args: argparse.Namespace) -> dict[str, Any]:
if args.sheet:
return _load_sheet_from_path(args.sheet)
if args.chat_id:
sheet = context_store.load_sheet(args.chat_id)
if sheet is None:
raise FileNotFoundError(
f"Kein Context-Sheet gefunden fuer chat_id={args.chat_id}")
return sheet
raise ValueError("Bitte --sheet oder --chat-id angeben.")
def main() -> None:
parser = argparse.ArgumentParser(
description="Isolierter Decision-LLM Test mit Context Sheet.")
parser.add_argument("--sheet", help="Pfad zu einem Context Sheet JSON.")
parser.add_argument(
"--chat-id", help="Chat-ID zum Laden aus logs/context_sheets.")
args = parser.parse_args()
sheet = _resolve_sheet(args)
context_text = context_store.format_sheet(sheet)
decision = decision_LLM.needs_more_context(context_text)
history_turns = context_store.get_history_turns(sheet)
last_inputs = history_turns[-1]["content"] if history_turns else ""
print("LAST_USER_INPUTS:", json.dumps(last_inputs, ensure_ascii=True))
print("OUTPUT:", json.dumps(decision, ensure_ascii=True))
if __name__ == "__main__":
main()
import argparse
import json
import os
from typing import Iterable
os.environ.setdefault("EMBEDDING_PROVIDER", "sentence-transformer")
os.environ.setdefault("EMBEDDING_TYPE", "sentence-transformer")
from app.LLM_services import math_intent_LLM
def _iter_inputs(args: argparse.Namespace) -> Iterable[str]:
if args.input_file:
with open(args.input_file, "r", encoding="utf-8") as handle:
for line in handle:
text = line.strip()
if text:
yield text
return
for text in args.input:
text = text.strip()
if text:
yield text
def main() -> None:
parser = argparse.ArgumentParser(
description="Isolierter Math-Intent Test.")
parser.add_argument(
"--input",
action="append",
default=[],
help="Eingabe fuer den Math-Intent (mehrfach angeben).",
)
parser.add_argument(
"--input-file",
help="Textdatei mit einer Eingabe pro Zeile.",
)
args = parser.parse_args()
if not args.input and not args.input_file:
raise ValueError("Bitte --input oder --input-file angeben.")
for text in _iter_inputs(args):
result = math_intent_LLM.solve_with_tools(text)
print("INPUT:", text)
print("OUTPUT:", json.dumps(result, ensure_ascii=True))
if __name__ == "__main__":
main()
......@@ -3,7 +3,6 @@ import os
import sys
import types
import unittest
import warnings
from unittest.mock import patch
os.environ.setdefault("EMBEDDING_TYPE", "sentence-transformer")
......@@ -75,10 +74,6 @@ from app.deterministic_services.orchestrators.orchestrator_base import ChatState
MESSAGES = [{"role": "user", "content": "Hallo"}]
def _dummy_tool() -> str:
return "ok"
class LLMProviderConfigTest(unittest.TestCase):
def test_get_llm_provider_accepts_supported_values(self) -> None:
for provider in ("openai", "gwdg", "mistral", "ollama"):
......@@ -238,28 +233,7 @@ class LLMClientProviderTest(unittest.TestCase):
self.assertEqual(result, expected)
openai_chat.assert_not_called()
mistral_chat.assert_not_called()
ollama_chat.assert_called_once_with(MESSAGES, tools=None)
def test_chat_deprecated_provider_flags_are_ignored(self) -> None:
settings = object()
expected = {"raw": object(), "message": {"content": "openai"}}
with patch.dict(os.environ, {"LLM_PROVIDER": "openai"}), patch.object(
llm_client, "_require_openai_chat_settings", return_value=settings
), patch.object(
llm_client, "_ensure_within_llm_quota"
), patch.object(
llm_client, "_record_call", side_effect=lambda result, tokens=None: result
), patch.object(
llm_client, "_chat_openai_compatible", return_value=expected
) as openai_chat, patch.object(
llm_client, "_chat_ollama"
) as ollama_chat:
with self.assertWarns(DeprecationWarning):
result = llm_client.chat(MESSAGES, use_ollama=True, use_mistral=True)
self.assertEqual(result, expected)
openai_chat.assert_called_once_with(MESSAGES, settings)
ollama_chat.assert_not_called()
ollama_chat.assert_called_once_with(MESSAGES)
def test_selected_gwdg_config_error_happens_before_quota(self) -> None:
with patch.dict(os.environ, {"LLM_PROVIDER": "gwdg"}, clear=True), patch.object(
......@@ -294,33 +268,6 @@ class LLMClientProviderTest(unittest.TestCase):
mistral_chat.assert_not_called()
ollama_chat.assert_not_called()
def test_chat_tools_argument_is_deprecated(self) -> None:
with patch.dict(os.environ, {"LLM_PROVIDER": "openai"}):
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
with self.assertRaisesRegex(RuntimeError, "toolcalling"):
llm_client.chat(MESSAGES, tools=[_dummy_tool])
messages = [str(warning.message) for warning in caught]
self.assertTrue(any("tools" in message for message in messages))
def test_chat_with_tools_is_deprecated(self) -> None:
response = {"raw": object(), "message": {"content": "ok"}}
with patch.dict(os.environ, {"LLM_PROVIDER": "ollama"}), patch.object(
llm_client, "_chat_ollama", return_value=response
):
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
result, tool_outputs = llm_client.chat_with_tools(
MESSAGES[:], [_dummy_tool], return_after_tools=True
)
messages = [str(warning.message) for warning in caught]
self.assertEqual(result, response)
self.assertEqual(tool_outputs, [])
self.assertTrue(any("chat_with_tools" in message for message in messages))
self.assertTrue(any("tools" in message for message in messages))
class TutorOrchestratorLegacyModuleTest(unittest.TestCase):
def test_tutor_orchestrator_does_not_import_legacy_llm_modules(self) -> None:
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment