Skip to content
GitLab
Commit e31efbb9 authored by Kantz's avatar Kantz
Browse files

Merge branch 'dev' into 'main' 

Dev

See merge request kantz/tutor_react!4
parents 43e305bb 5e93567b
Hide whitespace changes
Inline Side-by-side
math-tutor/README.md
View file @ e31efbb9
......@@ -15,7 +15,11 @@ cd math-tutor/frontend
npm install
npm run dev
```
Um es im Netzwerk verfügbar zu machen mit: npm run dev -- --host 0.0.0.0
To make it accessible over the network.
Add the frontend- and backend-adress in the .env file in the frontend- and backend-folder.
python -m uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
npm run dev -- --host 0.0.0.0
## Database (pgvector) setup
Ensure `POSTGRES_URL` and embedding env vars are in `backend/.env`:
......@@ -51,7 +55,7 @@ python -m scripts.retrieval_cli query --q "Was ist eine Teilmenge?" --k 8 --expa
## Configuration
System prompt (LLM):
Differend Orchestrators:
- Edit `math-tutor/backend/app/api/chat.py` and update `SYSTEM_PROMPT`.
Retrieval settings:
......@@ -63,7 +67,7 @@ Retrieval settings:
- Core retrieval logic is in `math-tutor/backend/app/services/vector_store.py` (`retrieve`).
## Testing
python -m test.hint_test ...
python -m test.vector_store_test --query "Was ist eine Teilmenge?" --k 8 --expand
python -m test.hint_test --chat-id draft_session_mlgmxxzc_avmjfb
python -m test.retrieval_store_test --query "Was ist eine Teilmenge?" --k 8 --expand
python -m test.math_intent_test --input "Integrate x^2" --input "Was ist 2+2?"
python -m test.decision_test --chat-id draft_session_mlgeyn5z_unpxyh
\ No newline at end of file
python -m test.decision_test --chat-id draft_session_mlgmxxzc_avmjfb
math-tutor/backend/.env-Example
View file @ e31efbb9
MATHPIX_APP_ID=""
MATHPIX_APP_KEY=""
OPENAI_BASE_URL=""
OPENAI_BASE_URL="https://chat-ai.academiccloud.de/v1/"
OPENAI_API_KEY=""
OPENAI_CHAT_MODEL=""
OPENAI_CHAT_TEMPERATURE=""
OPENAI_EMBED_MODEL=""
OPENAI_CHAT_MODEL="mistral-large-3-675b-instruct-2512"
OPENAI_CHAT_TEMPERATURE="0.2"
OPENAI_EMBED_MODEL="e5-mistral-7b-instruct"
EMBEDDING_TYPE="" # "openai-like" or "sentence-transformers"
EMBEDDING_TYPE="sentence-transformer" # "openai-like" or "sentence-transformer"
EMBEDDING_DIM="512"
SENTENCE_TRANSFORMER_MODEL="jinaai/jina-embeddings-v4"
ORCHESTRATOR="tutor" # "tutor" or "qa"
RETRIEVAL_IMPL="child" # "child" or "subsection"
POSTGRES_URL=""
OLLAMA_URL=""
OLLAMA_MODEL= ""
OLLAMA_TEMPERATURE=""
OLLAMA_URL="http://localhost:11434"
OLLAMA_MODEL= "ministral-3"
OLLAMA_TEMPERATURE="0.2"
FRONTEND_URL=""
\ No newline at end of file
FRONTEND_URL="http://localhost:5173"
\ No newline at end of file
math-tutor/backend/app/LLM_services/decision_LLM.py
View file @ e31efbb9
from ollama import chat
from app.deterministic_services import llm_client
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.
......@@ -27,37 +27,27 @@ WICHTIG:
- Löse nicht die Aufgabe, sondern bewerte nur die Situation bzgl. der nächsten didaktischen Schritte.
"""
def needs_more_context(history: str, context_sheet: str, model: str = "ministral-3") -> dict:
user_prompt = (
"Kontextblatt:\n"
f"{context_sheet}\n\n"
"Historie:\n"
f"{history}\n\n"
"Treffe eine Entscheidung."
def needs_more_context(context_sheet: str) -> dict:
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."
}
)
messages = [
{"role": "system", "content": CLASSIFIER_SYSTEM},
{"role": "user", "content": user_prompt},
]
resp = chat(
model=model,
resp, tool_outputs = llm_client.chat_with_tools(
messages=messages,
tools=[context_decision],
stream=False,
options={"temperature": 0},
use_ollama=True,
return_after_tools=True,
)
# 1) Ideal: strukturiertes Toolcall-Objekt
if resp.message.tool_calls:
call = resp.message.tool_calls[0]
if call.function.name != "context_decision":
return {"needs_more_context": True, "reason": f"Unexpected tool: {call.function.name}"}
args = call.function.arguments
# Tool ausführen (oder alternativ direkt args zurückgeben)
return context_decision(**args)
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)}
# 2) Fallback: falls Modell doch keinen Toolcall gemacht hat
return {"needs_more_context": True, "reason": resp.message.content or "No tool_call returned"}
return {"needs_more_context": False, "reason": "No tool_call returned"+str(resp)}
math-tutor/backend/app/LLM_services/hint_LLM.py
View file @ e31efbb9
from app.deterministic_services import llm_client
HINT_SYSTEM_PROMPT = """
Du bist ein didaktischer Mathe-Tutor."
1) Antworte NUR mit einem kurzen Tipp (1-2 Sätze), keine Beispiele, keine Herleitung, keine komplette Lösung.
2) Beziehe dich PRIMÄR auf die 'Aktuelle Frage'. Ignoriere ältere Nebenfragen, außer sie sind nötig.
3) Wenn der Nutzer etwas Falsches sagt weiße ihn darauf hin gib aber keine Lösung an.
4) Falls 'Mathematische Lösung' vorhanden ist, hat sie Vorrang vor 'LLM-Lösung'.
verwende die $ für mathematische Formeln, z.B. $a^2 + b^2 = c^2$
Antworte mit folgender Struktur
Das war richtig/das ist noch nicht richtig/gute Frage
[Hier soll der nächster Tipp um weiter zu machen/um den Fehler zu korrigieren stehen ohne Klammern]
"""
def generate_hint(
query: str,
task: str,
LLM_solution: str,
math_solution: str | None = None,
history: str | None = None,
history: list[dict] | None = None,
sources: str | None = None,
) -> str:
system_prompt = (
"Du bist ein didaktischer Tutor. "
"Gebe dem Nutzer einen hilfreichen Tipp um mathematische Aufgaben zu lösen."
"Gebe keine Beispiele oder Erklärungen, sondern nur den Tipp."
"Gebe keine komplette Lösung der Aufgabe."
"Die Mathematische Lösung hat immer Vorrang vor der LLM Lösung."
"Halte dich kurz und prägnant."
)
prompt = (
"Aufgabe:\n"
+ task
+ "\n\n"
"Loesung (von einem LLM):\n"
+ LLM_solution
+ "\n"
)
context_parts = [
f"Aufgabe:\n{task}",
f"LLM-Lösung:\n{LLM_solution}",
]
if math_solution:
prompt += "Mathematische Loesung:\n" + math_solution + "\n"
if history:
prompt = "\nHistorie:\n" + history + "\n" + prompt
context_parts.append(f"Mathematische Lösung (maßgeblich):\n{math_solution}")
if sources:
prompt = "\nKontext:\n" + sources + "\n" + prompt
if query:
prompt += "\n Aktuelle Frage:" + query
prompt += "\nGebe einen hilfreichen Tipp zum nächsten Schritt der Aufgabe der mir bei meiner aktuellen Frage hilft. Halte dich kurz und prägnant."
result = llm_client.chat(
messages=[{"role": "system", "content": system_prompt}, {"role": "user", "content": prompt}],
)
context_parts.append(f"Kontext/Sources:\n{sources}")
messages = [{"role": "system", "content": HINT_SYSTEM_PROMPT}]
# Kompakter Kontext als eine Nachricht (kein langer Fließtext mit History mischen)
messages.append({"role": "user", "content": "\n\n".join(context_parts)})
# History als echte Turns (und ggf. begrenzen, siehe Punkt 2)
if history:
messages.extend(history)
# Aktuelle Frage als letzte Nachricht, fett hervorgehoben durch Struktur/Delimiters
messages.append({
"role": "user",
"content": f"AKTUELLE FRAGE (höchste Priorität):\n{query}\n\nGib einen kurzen Tipp zum nächsten Schritt, der genau diese Frage adressiert."
})
result = llm_client.chat(messages=messages)
return llm_client.get_message_content(result)
\ No newline at end of file
math-tutor/backend/app/LLM_services/math_intent_LLM.py
View file @ e31efbb9
import sympy as sp
from ollama import chat
from app.deterministic_services import llm_client
def sympy_solve(task: str, input: str, symbols: list[str] | None = None) -> str:
"""
......@@ -31,38 +32,25 @@ def sympy_solve(task: str, input: str, symbols: list[str] | None = None) -> str:
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 final mit dem vereinfachten Ergebnis (kein Doppelbruch).
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, model: str = "ministral-3") -> str:
def solve_with_tools(user_text: str) -> str:
messages = [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": user_text},
{"role": "user", "content": user_text + "\nAntworte nur mit 'Keine Lösung' falls es in dieser Nachricht KEINE Berechung gibt."},
]
# 1) Modell darf Toolcalls erzeugen
resp = chat(model=model, messages=messages, tools=[sympy_solve], stream=False)
messages.append(resp.message)
# 2) Toolcalls ausführen und Ergebnisse zurückgeben
if resp.message.tool_calls:
for call in resp.message.tool_calls:
if call.function.name == "sympy_solve":
result = sympy_solve(**call.function.arguments)
else:
result = f"Unknown tool: {call.function.name}"
messages.append({
"role": "tool",
"tool_name": call.function.name,
"content": str(result),
})
# 3) Finalen Antwort-Call (Modell integriert Tool-Ergebnis)
final = chat(model=model, messages=messages, tools=[sympy_solve], stream=False)
return final.message.content
# Falls kein Toolcall: direkt zurück
return resp.message.content
result, tool_outputs = llm_client.chat_with_tools(
messages=messages,
tools=[sympy_solve],
use_ollama=True,
)
return llm_client.get_message_content(result)
math-tutor/backend/app/LLM_services/qa_LLM.py
View file @ e31efbb9
from app.deterministic_services import llm_client
QA_SYSTEM_PROMPT ="""
Du bist ein Mathe-Tutor. Antworte auf Deutsch, klar, korrekt und sehr kurz.
Nutze ausschließlich den bereitgestellten Kontext.
Wenn die Frage dort nicht beantwortet wird, antworte nur mit:
„Dazu steht nichts im Material.“
verwende die $ für mathematische Formeln, z.B. $a^2 + b^2 = c^2$
Liefer wenn möglich ein Zahlenbeispiel mit.
Zitiere immer die wichtigste Quelle direkt inline im Format der eckigen Klammern aus dem Kontext, z.B.:
[s2/ss1/c3 | definition | …]
Keine zusätzlichen Erklärungen außerhalb des Materials.
"""
def answer_question(
question: str,
history: str | None = None,
sources: str | None = None,
) -> str:
system_prompt = (
"""
Du bist ein Mathe-Tutor. Antworte auf Deutsch, klar, korrekt und sehr kurz.
Nutze ausschließlich den bereitgestellten Kontext.
Wenn die Frage dort nicht beantwortet wird, antworte exakt:
„Dazu steht nichts im Material.“
verwende die $$ für mathematische Formeln, z.B. $$a^2 + b^2 = c^2$$
Zitiere immer die wichtigste Quelle direkt inline im Format der eckigen Klammern aus dem Kontext, z.B.:
[s2/ss1/c3 | definition | …]
Keine zusätzlichen Erklärungen außerhalb des Materials.
"""
)
prompt = (
"Frage:\n"
+ question
......@@ -31,7 +32,9 @@ def answer_question(
if sources:
prompt = "\nKontext:\n" + sources + "\n" + prompt
prompt += "\nGebe eine didaktisch wertvolle Antwort. Halte dich sehr kurz und prägnant."
result = llm_client.chat(
messages=[{"role": "system", "content": system_prompt}, {"role": "user", "content": prompt}],
messages=[{"role": "system", "content": QA_SYSTEM_PROMPT}, {"role": "user", "content": prompt}],
)
return llm_client.get_message_content(result)
\ No newline at end of file
math-tutor/backend/app/LLM_services/solver_LLM.py
View file @ e31efbb9
from app.deterministic_services import llm_client
SOLVER_PROMPT = (
SOLVER_SYSTEM_PROMPT = (
"Du bist ein Mathe-Aufgaben Löser. Schreibe ein ausführliche genaue Lösung für die gestellt Fragen. "
"In der du alle Zwischenschritte aufführst und möglich Fehlerquellen auflistet."
"Deine Lösung soll korrekt sein und didaktisch gut formuliert."
......@@ -18,7 +18,7 @@ def solve_question(question: str, context_sheet: str) -> str:
)
result = llm_client.chat(
messages=[
{"role": "system", "content": SOLVER_PROMPT},
{"role": "system", "content": SOLVER_SYSTEM_PROMPT},
{"role": "user", "content": prompt},
],
)
......
math-tutor/backend/app/api/canvas.py
View file @ e31efbb9
......@@ -13,12 +13,12 @@ from mpxpy.mathpix_client import MathpixClient
router = APIRouter()
settings = config.get_mathpix_settings()
mathpix_client = None
if settings.app_id and settings.app_key:
mathpix_client = MathpixClient(app_id=settings.app_id, app_key=settings.app_key)
mathpix_client = MathpixClient(
app_id=settings.app_id, app_key=settings.app_key)
class CanvasSaveRequest(BaseModel):
......@@ -35,7 +35,8 @@ class CanvasSaveResponse(BaseModel):
@router.post("/api/canvas/save", response_model=CanvasSaveResponse)
def save_canvas(request: CanvasSaveRequest) -> CanvasSaveResponse:
if not request.data_url.startswith("data:image"):
raise HTTPException(status_code=400, detail="data_url must be an image")
raise HTTPException(
status_code=400, detail="data_url must be an image")
header, encoded = request.data_url.split(",", 1)
if "image/png" not in header:
......@@ -44,14 +45,16 @@ def save_canvas(request: CanvasSaveRequest) -> CanvasSaveResponse:
try:
raw = base64.b64decode(encoded)
except base64.binascii.Error as exc:
raise HTTPException(status_code=400, detail="invalid base64 payload") from exc
raise HTTPException(
status_code=400, detail="invalid base64 payload") from exc
drawings_dir = Path(__file__).resolve().parents[1] / "storage" / "drawings"
drawings_dir.mkdir(parents=True, exist_ok=True)
timestamp = int(time.time())
safe_hint = request.filename_hint or "drawing"
safe_hint = "".join(ch for ch in safe_hint if ch.isalnum() or ch in ("-", "_"))
safe_hint = "".join(
ch for ch in safe_hint if ch.isalnum() or ch in ("-", "_"))
if not safe_hint:
safe_hint = "drawing"
......@@ -63,7 +66,8 @@ def save_canvas(request: CanvasSaveRequest) -> CanvasSaveResponse:
handle.write(raw)
try:
image = mathpix_client.image_new(file_path=str(img_path), include_line_data=True)
image = mathpix_client.image_new(
file_path=str(img_path), include_line_data=True)
lines = image.lines_json()
line_type = lines[0]["type"] if lines else None
if line_type != "math":
......
math-tutor/backend/app/api/chat.py
View file @ e31efbb9
from __future__ import annotations
from logging import config
from typing import List, Optional
import logging
......@@ -8,9 +9,12 @@ from fastapi import APIRouter, HTTPException, Path, Query
from pydantic import BaseModel, Field
from app.deterministic_services import session_store
from app.deterministic_services.orchestrators import orchestrator_tutor as orchestrator
from app.deterministic_services.orchestrators import orchestrator_QA
import app.config as config
if config.get_orchestrator() == "tutor":
from app.deterministic_services.orchestrators import orchestrator_tutor as orchestrator
else:
from app.deterministic_services.orchestrators import orchestrator_qa as orchestrator
router = APIRouter()
......@@ -29,7 +33,7 @@ class ChatRequest(BaseModel):
class ChatResponse(BaseModel):
reply: str
sources: List[str] = []
sources: List[dict] = []
class ChatArchiveResponse(BaseModel):
......@@ -66,7 +70,8 @@ def chat(request: ChatRequest) -> ChatResponse:
raise HTTPException(status_code=500, detail=str(exc)) from exc
except Exception as exc:
logger.exception("Chat request failed")
raise HTTPException(status_code=502, detail="chat provider failed") from exc
raise HTTPException(
status_code=502, detail="chat provider failed") from exc
return ChatResponse(reply=reply, sources=sources)
......@@ -77,7 +82,8 @@ def list_archives(limit: int = Query(20, ge=1, le=200)) -> List[ChatArchiveSumma
return session_store.list_archives(limit=limit)
except Exception as exc:
logger.exception("Chat archive list failed")
raise HTTPException(status_code=502, detail="chat archive list failed") from exc
raise HTTPException(
status_code=502, detail="chat archive list failed") from exc
@router.get("/api/chat/archive/{chat_id}", response_model=ChatArchiveDetail)
......@@ -86,7 +92,8 @@ def get_archive(chat_id: str = Path(..., min_length=1)) -> ChatArchiveDetail:
record = session_store.load_archive(chat_id)
except Exception as exc:
logger.exception("Chat archive load failed")
raise HTTPException(status_code=502, detail="chat archive load failed") from exc
raise HTTPException(
status_code=502, detail="chat archive load failed") from exc
if not record:
raise HTTPException(status_code=404, detail="chat archive not found")
......@@ -94,7 +101,8 @@ def get_archive(chat_id: str = Path(..., min_length=1)) -> ChatArchiveDetail:
return ChatArchiveDetail(
chat_id=record["chat_id"],
saved_at=record.get("saved_at", ""),
history=[ChatMessage(role=item["role"], text=item["text"]) for item in record["history"]],
history=[ChatMessage(role=item["role"], text=item["text"])
for item in record["history"]],
)
......@@ -110,6 +118,7 @@ def archive_chat(request: ChatRequest) -> ChatArchiveResponse:
)
except Exception as exc:
logger.exception("Chat archive failed")
raise HTTPException(status_code=502, detail="chat archive failed") from exc
raise HTTPException(
status_code=502, detail="chat archive failed") from exc
return ChatArchiveResponse(status="ok", chat_id=chat_id)
math-tutor/backend/app/api/context.py
View file @ e31efbb9
......@@ -6,8 +6,6 @@ from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from app.deterministic_services import context_store
from app.deterministic_services import Source
router = APIRouter()
......@@ -21,6 +19,7 @@ class ContextRequest(BaseModel):
messages: List[ChatMessage]
draft: Optional[str] = None
@router.post("/api/context/retrieval")
def get_retrieval_context(request: ContextRequest) -> List[dict]:
if not request or not request.messages:
......
math-tutor/backend/app/api/health.py
View file @ e31efbb9
from __future__ import annotations
import logging
import os
from typing import Any, Dict
import httpx
import psycopg
from fastapi import APIRouter
import app.config as config
router = APIRouter()
logger = logging.getLogger(__name__)
def _check_ollama() -> dict:
base_url = os.getenv("OLLAMA_URL", "http://localhost:11434").rstrip("/")
base_url = config.get_ollama_settings().base_url.rstrip("/")
url = f"{base_url}/api/tags"
try:
with httpx.Client(timeout=5.0) as client:
......@@ -34,14 +35,13 @@ def _normalize_openai_models_url(base_url: str) -> str:
def _check_openai() -> dict:
base_url = os.getenv("OPENAI_BASE_URL")
api_key = os.getenv("OPENAI_API_KEY")
if not base_url or not api_key:
settings = config.get_openai_base_settings()
if not settings:
return {"status": "missing_config"}
url = _normalize_openai_models_url(base_url)
url = _normalize_openai_models_url(settings.base_url)
try:
headers = {"Authorization": f"Bearer {api_key}"}
headers = {"Authorization": f"Bearer {settings.api_key}"}
with httpx.Client(timeout=5.0) as client:
response = client.get(url, headers=headers)
if response.status_code in (401, 403):
......@@ -53,8 +53,9 @@ def _check_openai() -> dict:
def _check_postgres() -> dict:
pg_url = os.getenv("POSTGRES_URL")
if not pg_url:
try:
pg_url = config.get_postgres_url()
except ValueError:
return {"status": "missing_config"}
try:
with psycopg.connect(pg_url, connect_timeout=5) as conn:
......
math-tutor/backend/app/api/retrieval.py deleted 100644 → 0
View file @ 43e305bb
from __future__ import annotations
from __future__ import annotations
import os
from pathlib import Path
from typing import List, Optional
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from app import config
from app.deterministic_services.embeddings import OpenAILikeEmbeddings
from app.deterministic_services import vector_store
router = APIRouter()
def _get_embedder() -> OpenAILikeEmbeddings:
settings = config.get_embedding_settings()
return OpenAILikeEmbeddings(
base_url=settings.base_url,
api_key=settings.api_key,
model=settings.model,
target_dim=settings.target_dim,
)
class InitDbRequest(BaseModel):
pg_url: Optional[str] = None
class IngestRequest(BaseModel):
pg_url: Optional[str] = None
base_dir: str = Field(default="markdown")
class QueryRequest(BaseModel):
pg_url: Optional[str] = None
query: str = Field(..., min_length=1)
k: int = 4
expand_links: bool = True
section_index: Optional[int] = None
subsection_index: Optional[int] = None
source_type_filter: Optional[List[str]] = None
neighbor_expand: int = 0
@router.post("/api/retrieval/init-db")
def init_db(request: InitDbRequest) -> dict:
pg_url = request.pg_url or os.getenv("POSTGRES_URL")
if not pg_url:
raise HTTPException(status_code=500, detail="Missing POSTGRES_URL")
vector_store.init_db(pg_url)
return {"status": "ok"}
@router.post("/api/retrieval/ingest")
def ingest(request: IngestRequest) -> dict:
pg_url = request.pg_url or os.getenv("POSTGRES_URL")
if not pg_url:
raise HTTPException(status_code=500, detail="Missing POSTGRES_URL")
base_dir = Path(request.base_dir)
if not base_dir.exists():
raise HTTPException(status_code=400, detail="base_dir does not exist")
embedder = _get_embedder()
docs = vector_store.load_docs(base_dir)
embeddings = vector_store.embed_passages(embedder, [doc.markdown for doc in docs])
upserted = vector_store.upsert_docs(pg_url, docs, embeddings)
return {"status": "ok", "upserted": upserted}
@router.post("/api/retrieval/query")
def query(request: QueryRequest) -> dict:
pg_url = request.pg_url or os.getenv("POSTGRES_URL")
if not pg_url:
raise HTTPException(status_code=500, detail="Missing POSTGRES_URL")
embedder = _get_embedder()
result = vector_store.retrieve(
pg_url=pg_url,
embedder=embedder,
query=request.query,
k=request.k,
section_index=request.section_index,
subsection_index=request.subsection_index,
source_type_filter=request.source_type_filter,
expand_links=request.expand_links,
neighbor_expand=request.neighbor_expand,
)
return result
@router.get("/api/retrieval/sections")
def sections(pg_url: Optional[str] = None) -> list[dict]:
url = pg_url or os.getenv("POSTGRES_URL")
if not url:
raise HTTPException(status_code=500, detail="Missing POSTGRES_URL")
return vector_store.list_sections(url)
@router.get("/api/retrieval/subsections")
def subsections(pg_url: Optional[str] = None, section_index: Optional[int] = None) -> list[dict]:
url = pg_url or os.getenv("POSTGRES_URL")
if not url:
raise HTTPException(status_code=500, detail="Missing POSTGRES_URL")
return vector_store.list_subsections(url, section_index)
math-tutor/backend/app/config.py
View file @ e31efbb9
......@@ -7,6 +7,7 @@ from typing import Optional
load_dotenv()
class EmbeddingSettings(BaseModel):
embedding_type: str # "openai-like" oder "sentence-transformer"
base_url: Optional[str] = None
......@@ -14,19 +15,32 @@ class EmbeddingSettings(BaseModel):
model: str
target_dim: int = 1024
def get_orchestrator() -> str:
return os.getenv("ORCHESTRATOR", "qa").lower()
def get_retrieval_impl() -> str:
value = os.getenv("RETRIEVAL_IMPL", "child").strip().lower()
if value in {"child", "subsection"}:
return value
return "child"
def get_embedding_settings() -> EmbeddingSettings:
embedding_type = os.getenv("EMBEDDING_TYPE", "openai-like")
if embedding_type == "sentence-transformer":
return EmbeddingSettings(
embedding_type=embedding_type,
model=os.getenv("SENTENCE_TRANSFORMER_MODEL", "all-MiniLM-L6-v2"),
target_dim=int(os.getenv("SENTENCE_TRANSFORMER_TARGET_DIM", "1024")),
model=os.getenv("SENTENCE_TRANSFORMER_MODEL",
"jinaai/jina-embeddings-v4"),
target_dim=int(os.getenv("EMBEDDING_DIM", "512")),
)
if embedding_type == "openai-like":
base_url = os.getenv("OPENAI_BASE_URL")
api_key = os.getenv("OPENAI_API_KEY")
model = os.getenv("OPENAI_EMBED_MODEL", "text-embedding-3-large")
model_target_dim = int(os.getenv("OPENAI_EMBED_TARGET_DIM", "1024"))
model = os.getenv("OPENAI_EMBED_MODEL", "e5-mistral-7b-instruct")
model_target_dim = int(os.getenv("EMBEDDING_DIM", "1024"))
if not base_url or not api_key:
raise ValueError("Missing OPENAI_BASE_URL or OPENAI_API_KEY")
return EmbeddingSettings(
......@@ -48,6 +62,7 @@ class OllamaSettings:
keepalive: str | None
temperature: float | None
@dataclass(frozen=True)
class OpenAIChatSettings:
base_url: str
......@@ -56,6 +71,13 @@ class OpenAIChatSettings:
timeout: float | None
temperature: float | None
@dataclass(frozen=True)
class OpenAIBaseSettings:
base_url: str
api_key: str
@dataclass(frozen=True)
class MathpixSettings:
app_id: str
......@@ -70,18 +92,21 @@ def _read_float(value: str | None) -> float | None:
except ValueError:
return None
def get_mathpix_settings() -> MathpixSettings:
return MathpixSettings(
app_id=os.getenv("MATHPIX_APP_ID"),
app_key=os.getenv("MATHPIX_APP_KEY")
app_key=os.getenv("MATHPIX_APP_KEY")
)
def get_frontend_url() -> str:
url = os.getenv("FRONTEND_URL")
if not url:
raise ValueError("Missing FRONTEND_URL")
return url
def get_ollama_settings() -> OllamaSettings:
return OllamaSettings(
base_url=os.getenv("OLLAMA_URL", "http://localhost:11434"),
......@@ -91,17 +116,28 @@ def get_ollama_settings() -> OllamaSettings:
temperature=_read_float(os.getenv("OLLAMA_TEMPERATURE")),
)
def get_openai_base_settings() -> OpenAIBaseSettings | None:
base_url = os.getenv("OPENAI_BASE_URL")
api_key = os.getenv("OPENAI_API_KEY")
if not base_url or not api_key:
return None
return OpenAIBaseSettings(
base_url=base_url,
api_key=api_key,
)
def get_openai_chat_settings() -> OpenAIChatSettings | None:
model = os.getenv("OPENAI_CHAT_MODEL")
if not model:
return None
base_url = os.getenv("OPENAI_BASE_URL")
api_key = os.getenv("OPENAI_API_KEY")
if not base_url or not api_key:
base_settings = get_openai_base_settings()
if not base_settings:
raise ValueError("Missing OPENAI_BASE_URL or OPENAI_API_KEY for chat")
return OpenAIChatSettings(
base_url=base_url,
api_key=api_key,
base_url=base_settings.base_url,
api_key=base_settings.api_key,
model=model,
timeout=_read_float(os.getenv("OPENAI_CHAT_TIMEOUT")),
temperature=_read_float(os.getenv("OPENAI_CHAT_TEMPERATURE")),
......
math-tutor/backend/app/deterministic_services/context_store.py
View file @ e31efbb9
......@@ -10,6 +10,10 @@ from typing import Any
from app.deterministic_services import Source
# ---------------------------------------------------------------------------------------------------
# Basisfunktionen des Contex-Sheets
# ---------------------------------------------------------------------------------------------------
_CACHE: dict[str, dict[str, Any]] = {}
_LOCK = Lock()
_LOG_DIR = os.path.join("logs", "context_sheets")
......@@ -22,22 +26,14 @@ def _utc_now() -> str:
def get_chat_id(messages: list[dict], draft: str | None = None) -> str:
if draft:
return f"draft_{draft}"
first_user = next((m for m in messages if m.get("role") == "user" and m.get("content")), None)
first_user = next((m for m in messages if m.get("role")
== "user" and m.get("content")), None)
if not first_user:
return "unknown"
digest = hashlib.sha1(first_user["content"].encode("utf-8")).hexdigest()
return digest[:12]
def format_history(messages: list[dict]) -> str:
lines = []
for msg in messages:
role = msg.get("role", "unknown")
content = msg.get("content", "")
lines.append(f"{role}: {content}")
return "\n".join(lines)
def init_sheet(chat_id: str, messages: list[dict]) -> dict[str, Any]:
timestamp = _utc_now()
return {
......@@ -45,9 +41,9 @@ def init_sheet(chat_id: str, messages: list[dict]) -> dict[str, Any]:
"created_at": timestamp,
"updated_at": timestamp,
"history": messages[:],
"retrieval_contexts": [],
"sources": [],
"math_solutions": [],
"LLM_solutions":[],
"LLM_solutions": [],
"decisions": [],
"initialized": False,
}
......@@ -68,32 +64,117 @@ def load_sheet(chat_id: str) -> dict[str, Any] | None:
return None
def save_sheet(sheet: dict[str, Any]) -> None:
os.makedirs(_LOG_DIR, exist_ok=True)
sheet["updated_at"] = _utc_now()
chat_id = sheet.get("chat_id", "unknown")
latest_path = os.path.join(_LOG_DIR, f"{chat_id}_latest.json")
payload = json.dumps(sheet, ensure_ascii=True, indent=2)
with open(latest_path, "w", encoding="utf-8") as f:
f.write(payload)
with _LOCK:
_CACHE[chat_id] = sheet
def format_sheet(sheet: dict[str, Any]) -> str:
parts = []
history = format_history(sheet.get("history", []))
parts.append("HISTORY:\n" + (history or "(leer)"))
sources = sheet.get("sources", [])
if sources:
# Erstelle eine formatierte Liste der Quellen mit ihren Scores
source_blocks = []
for source in sources:
source_blocks.append(Source.model_validate(source).to_string())
source_info = "\n".join(source_blocks) if source_blocks else ""
parts.append(f"SOURCES:\n{source_info}")
else:
parts.append("SOURCES:\n(leer)")
math_solutions = sheet.get("math_solutions", [])
if math_solutions:
blocks = []
for item in math_solutions:
blocks.append(
"TASK: {task}\nINPUT: {input}\nSYMBOLS: {symbols}\nSOLUTION: {solution}".format(
task=item.get("task", ""),
input=item.get("input", ""),
symbols=",".join(item.get("symbols", [])),
solution=item.get("solution", ""),
)
)
parts.append("MATH_SOLUTIONS:\n" + "\n\n".join(blocks))
else:
parts.append("MATH_SOLUTIONS:\n(leer)")
return "\n\n".join(parts)
# ---------------------------------------------------------------------------------------------------
# History related
# ---------------------------------------------------------------------------------------------------
def update_history(sheet: dict[str, Any], messages: list[dict]) -> None:
sheet["history"] = messages[:]
sheet["updated_at"] = _utc_now()
def add_retrieval_context(
sheet: dict[str, Any],
query: str,
sources: list[Source],
) -> None:
retrieval_entry = {"query": query, "sources": [source.model_dump() for source in sources]}
sheet["retrieval_contexts"].append(retrieval_entry)
sheet["updated_at"] = _utc_now()
def format_history(messages: list[dict]) -> str:
lines = []
for msg in messages:
role = msg.get("role", "unknown")
content = msg.get("content", "")
lines.append(f"{role}: {content}")
return "\n".join(lines)
def get_history(sheet: dict[str, Any]) -> str:
return format_history(sheet.get("history", []))
def get_history_turns(sheet: dict[str, Any]) -> list[dict]:
return sheet.get("history", [])
def get_task(sheet: dict[str, Any]) -> str:
history = sheet.get("history", [])
if history:
return history[-1].get("content", "")
return ""
# ---------------------------------------------------------------------------------------------------
# Retrieval related
# ---------------------------------------------------------------------------------------------------
def update_retrieval_context(
sheet: dict[str, Any],
query: str,
sources: list[Source],
) -> None:
retrievals = sheet.get("retrieval_contexts", [])
if retrievals:
latest_retrieval = retrievals[-1]
latest_retrieval["sources"] = [source.model_dump() for source in sources]
latest_retrieval["query"] = query
sheet["updated_at"] = _utc_now()
temp_sources = get_retrieval(sheet)
for source in sources:
temp_sources.append(source)
temp_sources = sorted(sources, key=lambda x: x.score, reverse=True)
temp_sources = temp_sources[:8]
for source in temp_sources:
sheet["sources"].append(source.model_dump())
sheet["updated_at"] = _utc_now()
def get_retrieval(sheet: dict[str, Any]) -> list[Source]:
sources = sheet.get("sources", [])
if not sources:
return []
return [Source.model_validate(source) for source in sources]
# ---------------------------------------------------------------------------------------------------
# Math-solution related
# ---------------------------------------------------------------------------------------------------
def add_math_solution(
......@@ -107,6 +188,17 @@ def add_math_solution(
)
sheet["updated_at"] = _utc_now()
def first_math_solution(sheet: dict[str, Any]) -> str:
if not sheet["math_solutions"]:
return ""
return sheet["math_solutions"][0].get("solution", "")
# ---------------------------------------------------------------------------------------------------
# LLM-solution related
# ---------------------------------------------------------------------------------------------------
def add_LLM_solution(
sheet: dict[str, Any],
solution: str,
......@@ -119,102 +211,17 @@ def add_LLM_solution(
sheet["updated_at"] = _utc_now()
def add_decision(sheet: dict[str, Any], decision: dict[str, Any]) -> None:
entry = {"timestamp": _utc_now(), **decision}
sheet["decisions"].append(entry)
sheet["updated_at"] = _utc_now()
# wird aktuell nicht benutzt
def latest_math_solution(sheet: dict[str, Any]) -> str:
if not sheet["math_solutions"]:
return ""
return sheet["math_solutions"][-1].get("solution", "")
def last_LLM_solution(sheet: dict[str, Any]) -> str:
if not sheet["LLM_solutions"]:
return ""
return sheet["LLM_solutions"][-1].get("solution", "")
def get_task(sheet: dict[str, Any]) -> str:
history = sheet.get("history", [])
if history:
return history[-1].get("content", "")
return ""
def get_history(sheet: dict[str, Any]) -> str:
return format_history(sheet.get("history", []))
def first_math_solution(sheet: dict[str, Any]) -> str:
if not sheet["math_solutions"]:
return ""
return sheet["math_solutions"][0].get("solution", "")
# ---------------------------------------------------------------------------------------------------
# decision related
# ---------------------------------------------------------------------------------------------------
def format_sheet(sheet: dict[str, Any]) -> str:
parts = []
history = format_history(sheet.get("history", []))
parts.append("HISTORY:\n" + (history or "(leer)"))
retrievals = sheet.get("retrieval_contexts", [])
if retrievals:
blocks = []
for item in retrievals:
query = item.get('query', '')
sources = item.get('sources', [])
# Erstelle eine formatierte Liste der Quellen mit ihren Scores
source_blocks = []
for source in sources:
source_blocks.append(Source.model_validate(source).to_string())
source_info = "\n".join(source_blocks) if source_blocks else ""
blocks.append(f"QUERY: {query}\nSOURCES:\n{source_info}")
parts.append("RETRIEVAL_CONTEXT:\n" + "\n\n".join(blocks))
else:
parts.append("RETRIEVAL_CONTEXT:\n(leer)")
math_solutions = sheet.get("math_solutions", [])
if math_solutions:
blocks = []
for item in math_solutions:
blocks.append(
"TASK: {task}\nINPUT: {input}\nSYMBOLS: {symbols}\nSOLUTION: {solution}".format(
task=item.get("task", ""),
input=item.get("input", ""),
symbols=",".join(item.get("symbols", [])),
solution=item.get("solution", ""),
)
)
parts.append("MATH_SOLUTIONS:\n" + "\n\n".join(blocks))
else:
parts.append("MATH_SOLUTIONS:\n(leer)")
return "\n\n".join(parts)
def get_retrieval(sheet: dict[str, Any]) -> list[Source]:
retrievals = sheet.get("retrieval_contexts", [])
if not retrievals:
return []
latest_retrieval = retrievals[-1]
sources= latest_retrieval.get("sources", [])
return [Source.model_validate(source) for source in sources]
def save_sheet(sheet: dict[str, Any]) -> None:
os.makedirs(_LOG_DIR, exist_ok=True)
def add_decision(sheet: dict[str, Any], decision: dict[str, Any]) -> None:
entry = {"timestamp": _utc_now(), **decision}
sheet["decisions"].append(entry)
sheet["updated_at"] = _utc_now()
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
chat_id = sheet.get("chat_id", "unknown")
snapshot_path = os.path.join(_LOG_DIR, f"{chat_id}_{timestamp}.json")
latest_path = os.path.join(_LOG_DIR, f"{chat_id}_latest.json")
payload = json.dumps(sheet, ensure_ascii=True, indent=2)
with open(snapshot_path, "w", encoding="utf-8") as f:
f.write(payload)
with open(latest_path, "w", encoding="utf-8") as f:
f.write(payload)
with _LOCK:
_CACHE[chat_id] = sheet
math-tutor/backend/app/deterministic_services/embedding_provider.py 0 → 100644
View file @ e31efbb9
from __future__ import annotations
import time
from threading import Lock
from typing import Any, Dict, Tuple
import app.config as config
from app.deterministic_services.embeddings import BaseEmbeddings, EmbeddingFactory
_EMBEDDER_LOCK = Lock()
_CACHED_EMBEDDER: BaseEmbeddings | None = None
_CACHED_KEY: Tuple[Any, ...] | None = None
def _settings_cache_key(settings: config.EmbeddingSettings) -> Tuple[Any, ...]:
return (
settings.embedding_type,
settings.base_url,
settings.model,
settings.target_dim,
)
def get_embedder() -> tuple[BaseEmbeddings, bool]:
settings = config.get_embedding_settings()
key = _settings_cache_key(settings)
global _CACHED_EMBEDDER
global _CACHED_KEY
with _EMBEDDER_LOCK:
if _CACHED_EMBEDDER is not None and _CACHED_KEY == key:
return _CACHED_EMBEDDER, True
_CACHED_EMBEDDER = EmbeddingFactory.create(settings)
_CACHED_KEY = key
return _CACHED_EMBEDDER, False
def warmup_embedder() -> Dict[str, Any]:
started = time.perf_counter()
embedder, cache_hit = get_embedder()
init_ms = round((time.perf_counter() - started) * 1000, 2)
warm_started = time.perf_counter()
embedder.embed_query("warmup")
warmup_ms = round((time.perf_counter() - warm_started) * 1000, 2)
total_ms = round((time.perf_counter() - started) * 1000, 2)
return {
"cache_hit": cache_hit,
"embedder_init_ms": init_ms,
"embed_query_warmup_ms": warmup_ms,
"total_warmup_ms": total_ms,
}
math-tutor/backend/app/deterministic_services/embeddings.py
View file @ e31efbb9
......@@ -20,33 +20,40 @@ class EmbeddingType(str, Enum):
class OpenAILikeConfig(BaseModel):
"""Konfiguration für OpenAI-ähnliche APIs."""
base_url: str = Field(..., description="Base URL der API (z. B. http://localhost:11434/v1)")
api_key: str = Field(..., description="API-Key (z. B. 'ollama' für Ollama)")
model: str = Field(..., description="Modellname (z. B. 'nomic-embed-text')")
base_url: str = Field(...,
description="Base URL der API (z. B. http://localhost:11434/v1)")
api_key: str = Field(...,
description="API-Key (z. B. 'ollama' für Ollama)")
model: str = Field(...,
description="Modellname (z. B. 'nomic-embed-text')")
target_dim: int = Field(1024, description="Ziel-Dimension der Embeddings")
class SentenceTransformerConfig(BaseModel):
"""Konfiguration für lokale SentenceTransformer-Modelle."""
model: str = Field(..., description="Name des SentenceTransformer-Modells (z. B. 'all-MiniLM-L6-v2')")
target_dim: int = Field(1024, description="Ziel-Dimension der Embeddings")
model: str = Field(...,
description="Name des SentenceTransformer-Modells (z. B. 'all-MiniLM-L6-v2')")
target_dim: int = Field(384, description="Ziel-Dimension der Embeddings")
class EmbeddingConfig(BaseModel):
"""Gemeinsame Konfiguration für die Factory."""
embedding_type: EmbeddingType = Field(..., description="Typ der Embeddings")
config: Union[OpenAILikeConfig, SentenceTransformerConfig] = Field(..., description="Spezifische Konfiguration")
embedding_type: EmbeddingType = Field(...,
description="Typ der Embeddings")
config: Union[OpenAILikeConfig, SentenceTransformerConfig] = Field(
..., description="Spezifische Konfiguration")
# -----------------------------
# Basisklasse für Embeddings
# -----------------------------
class BaseEmbeddings:
"""
Basisklasse für Embedding-Generierung mit gemeinsamen Methoden.
"""
def __init__(self, target_dim: int = 1024) -> None:
def __init__(self, target_dim: int = 384) -> None:
self.target_dim = target_dim
def _normalize(self, vec: List[float]) -> List[float]:
......@@ -59,7 +66,8 @@ class BaseEmbeddings:
def _truncate(self, vec: List[float]) -> List[float]:
"""Trunziert oder füllt den Vektor auf die Ziel-Dimension."""
if len(vec) < self.target_dim:
raise ValueError(f"Embedding dimension {len(vec)} < target {self.target_dim}")
raise ValueError(
f"Embedding dimension {len(vec)} < target {self.target_dim}")
if len(vec) > self.target_dim:
vec = vec[: self.target_dim]
return self._normalize(vec)
......@@ -114,7 +122,8 @@ class OpenAILikeEmbeddings(BaseEmbeddings):
}
with httpx.Client(timeout=60.0) as client:
response = client.post(self.endpoint, headers=headers, json=payload)
response = client.post(
self.endpoint, headers=headers, json=payload)
response.raise_for_status()
data = response.json().get("data")
......@@ -148,18 +157,28 @@ class SentenceTransformerEmbeddings(BaseEmbeddings):
def model(self) -> SentenceTransformer:
"""Liefert das SentenceTransformer-Modell (lazy load)."""
if self._model is None:
self._model = SentenceTransformer(self.model_name)
self._model = SentenceTransformer(
self.model_name, trust_remote_code=True)
self._model.max_seq_length = 512
return self._model
def _embed(self, inputs: List[str] | str) -> List[List[float]]:
"""Generiert Embeddings mit dem lokalen SentenceTransformer-Modell."""
embeddings = self.model.encode(inputs)
# Konvertiere in Liste von Listen (falls nötig)
if isinstance(embeddings, list) and all(isinstance(x, (int, float)) for x in embeddings[0]):
# Einzelner Vektor
return [self._truncate(embeddings)]
# Mehrere Vektoren
return [self._truncate(vec) for vec in embeddings]
def embed_documents(self, texts: List[str]) -> List[List[float]]:
"""Generiert Embeddings für eine Liste von Texten."""
passage_embeddings = self.model.encode(
sentences=texts,
task="retrieval",
prompt_name="passage",
)
return [self._truncate([float(x) for x in emb]) for emb in passage_embeddings]
def embed_query(self, text: str) -> List[float]:
"""Generiert ein Embedding für einen einzelnen Text."""
query_embeddings = self.model.encode(
sentences=[text],
task="retrieval",
prompt_name="query",
)
return self._truncate([float(x) for x in query_embeddings[0]])
# -----------------------------
......@@ -190,4 +209,5 @@ class EmbeddingFactory:
elif config.embedding_type == EmbeddingType.SENTENCE_TRANSFORMER:
return SentenceTransformerEmbeddings(config=config)
else:
raise ValueError(f"Unsupported embedding type: {config.embedding_type}")
\ No newline at end of file
raise ValueError(
f"Unsupported embedding type: {config.embedding_type}")
math-tutor/backend/app/deterministic_services/llm_client.py
View file @ e31efbb9
import inspect
import json
from typing import Any, Callable
import ollama
from openai import OpenAI
......@@ -13,6 +15,12 @@ def _filter_kwargs(func, kwargs: dict) -> dict:
return kwargs
return {key: value for key, value in kwargs.items() if key in signature.parameters}
def _extract_message(response) -> object:
if isinstance(response, dict) and "message" in response:
return response["message"]
if hasattr(response, "message"):
return response.message
return {}
def _chat_openai(messages: list[dict]) -> dict:
settings = config.get_openai_chat_settings()
......@@ -20,7 +28,8 @@ def _chat_openai(messages: list[dict]) -> dict:
return {}
timeout = settings.timeout or 60.0
client = OpenAI(api_key=settings.api_key, base_url=settings.base_url, timeout=timeout)
client = OpenAI(api_key=settings.api_key,
base_url=settings.base_url, timeout=timeout)
kwargs = {"messages": messages, "model": settings.model}
if settings.temperature is not None:
kwargs["temperature"] = settings.temperature
......@@ -29,9 +38,18 @@ def _chat_openai(messages: list[dict]) -> dict:
return {"raw": response, "message": message}
def get_message_content(result: dict | object) -> str:
message = result.get("message") if isinstance(result, dict) else result
if isinstance(message, dict):
return message.get("content", "") or ""
if hasattr(message, "content"):
return getattr(message, "content") or ""
return ""
def chat(
messages: list[dict],
tools: list[dict] | None = None,
tools: list[Callable[..., Any]] | None = None,
use_ollama: bool = False,
) -> dict:
if not tools and not use_ollama:
......@@ -50,29 +68,66 @@ def chat(
if settings.temperature is not None:
kwargs["options"] = {"temperature": settings.temperature}
response = client.chat(**_filter_kwargs(client.chat, kwargs))
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)}
def _extract_message(response) -> object:
if isinstance(response, dict) and "message" in response:
return response["message"]
if hasattr(response, "message"):
return response.message
return {}
def chat_with_tools(
messages: list[dict],
tools: list[Callable[..., Any]],
use_ollama: bool = True,
return_after_tools: bool = False,
) -> tuple[dict, list[dict[str, Any]]]:
tool_map = {tool.__name__: tool for tool in tools}
result = chat(messages=messages, tools=tools, use_ollama=use_ollama)
tool_outputs = _apply_tool_calls(result, messages, tool_map)
if not tool_outputs or return_after_tools:
return result, tool_outputs
def get_message_content(result: dict | object) -> str:
message = result.get("message") if isinstance(result, dict) else result
if isinstance(message, dict):
return message.get("content", "") or ""
if hasattr(message, "content"):
return getattr(message, "content") or ""
return ""
final_result = chat(messages=messages, tools=tools, use_ollama=use_ollama)
return final_result, tool_outputs
def get_tool_calls(result: dict | object) -> list:
def _apply_tool_calls(
result: dict | object,
messages: list[dict],
tool_map: dict[str, Callable[..., Any]],
) -> list[dict[str, Any]]:
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:
if isinstance(message, dict):
return message.get("tool_calls") or []
if hasattr(message, "tool_calls"):
......@@ -80,11 +135,42 @@ def get_tool_calls(result: dict | object) -> list:
return []
def normalize_tool_call(tool_call: object) -> dict:
if isinstance(tool_call, dict):
function = tool_call.get("function") or {}
return {"name": function.get("name"), "arguments": function.get("arguments")}
function = getattr(tool_call, "function", None)
if function:
return {"name": getattr(function, "name", None), "arguments": getattr(function, "arguments", None)}
return {"name": None, "arguments": None}
def _tool_call_name_args(call: object) -> tuple[str, dict[str, Any]]:
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]:
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
math-tutor/backend/app/deterministic_services/orchestrators/orchestrator_QA.py
View file @ e31efbb9
from app.deterministic_services import context_store, retrieval_service, tool_logging
from app.deterministic_services import referenz_decoder
from app.LLM_services import qa_LLM
from typing import List
def _is_new_chat(messages: list[dict]) -> bool:
return not any(m.get("role") == "assistant" for m in messages)
def _append_tool_log(tool_log: list[dict], name: str, args: dict, response: object) -> None:
tool_log.append({"name": name, "arguments": args, "response": response})
def _bootstrap_context(sheet: dict, query_text: str, tool_log: list[dict]) -> None:
# Erstelle einen neuen Retrieval-Block oder aktualisiere den bestehenden
sources = retrieval_service.retrieve_context(query_text=query_text)
retrievals = sheet.get("retrieval_contexts", [])
if retrievals:
context_store.update_retrieval_context(sheet, query_text, sources)
_append_tool_log(tool_log, "update_retrieve_context", {"query": query_text}, {"sources": [source.to_string() for source in sources]})
else:
context_store.add_retrieval_context(sheet, query_text, sources)
_append_tool_log(tool_log, "retrieve_context", {"query": query_text}, {"sources": [source.to_string() for source in sources]})
def _extract_user_messages(messages: list[dict]) -> list[str]:
"""
Extrahiert alle Inhalte von Messages mit der Rolle 'user' und gibt sie als Liste von Strings zurück.
Der letzte user-Content wird als letztes Element in der Liste enthalten sein.
"""
user_contents = []
for message in messages:
if message.get("role") == "user":
content = message.get("content", "")
if content: # Nur hinzufügen, wenn Inhalt vorhanden ist
user_contents.append(content)
return user_contents
from __future__ import annotations
from app.LLM_services import qa_LLM
from app.deterministic_services import context_store
from app.deterministic_services.orchestrators import orchestrator_base as base
def run_chat(messages: list[dict], draft: str | None = None) -> dict:
# Input-Fehlerbehandlung
if not messages:
raise ValueError("messages required")
last_user = _extract_user_messages(messages)[-1] if _extract_user_messages(messages) else None
if not last_user:
raise ValueError("last user message required")
def _on_bootstrap(state: base.ChatState, query_text: str) -> None:
base.bootstrap_retrieval(state.sheet, query_text, state.tool_log)
# Context Store Mangement
chat_id = context_store.get_chat_id(messages, draft=draft)
new_chat = _is_new_chat(messages)
sheet = context_store.load_sheet(chat_id)
if new_chat or not sheet:
sheet = context_store.init_sheet(chat_id, messages)
context_store.update_history(sheet, messages)
tool_log: list[dict] = []
def _on_turn_logic(state: base.ChatState) -> None:
return None
if new_chat or not sheet.get("initialized"):
_bootstrap_context(sheet, last_user, tool_log)
sheet["initialized"] = True
# Antwort generierung
hint_args = {
"question": context_store.get_task(sheet),
"history": context_store.format_history(messages),
"sources": "\n".join([source.to_string() for source in context_store.get_retrieval(sheet)]),
def _on_build_reply(state: base.ChatState) -> str | None:
args = {
"question": context_store.get_task(state.sheet),
"history": context_store.format_history(state.messages),
"sources": "\n".join([source.to_string() for source in context_store.get_retrieval(state.sheet)]),
}
reply = qa_LLM.answer_question(**hint_args)
_append_tool_log(tool_log, "answer_question", hint_args, reply)
return base.log_timed_call(
state.tool_log,
"answer_question",
args,
lambda: qa_LLM.answer_question(**args),
)
if not reply:
reply = "Dazu steht nichts im Material"
else:
decoded, _ = referenz_decoder.decode_references(
reply, context_store.get_retrieval(sheet)
)
reply = decoded
context_store.save_sheet(sheet)
tool_logging.write_tool_log(
tool_log,
created_at=sheet.get("created_at"),
chat_id=sheet.get("chat_id"),
def run_chat(messages: list[dict], draft: str | None = None) -> dict:
return base.run_chat_common(
messages=messages,
draft=draft,
on_bootstrap=_on_bootstrap,
on_turn_logic=_on_turn_logic,
on_build_reply=_on_build_reply,
)
return {"reply": reply, "sources": sheet.get("sources", []), "tool_log": tool_log}
math-tutor/backend/app/deterministic_services/orchestrators/orchestrator_base.py 0 → 100644
View file @ e31efbb9
from __future__ import annotations
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Callable, List, TypeVar
import app.config as config
from app.deterministic_services import (
Source,
context_store,
embedding_provider,
referenz_decoder,
retrieval_store,
tool_logging,
)
T = TypeVar("T")
@dataclass
class ChatState:
messages: list[dict]
draft: str | None
chat_id: str
new_chat: bool
sheet: dict
tool_log: list[dict]
last_user: str
def is_new_chat(messages: list[dict]) -> bool:
return not any(m.get("role") == "assistant" for m in messages)
def _utc_now_iso() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _start_timing() -> tuple[str, float]:
return _utc_now_iso(), time.perf_counter()
def _finish_timing(started_perf: float) -> tuple[str, float]:
finished_at = _utc_now_iso()
duration_ms = round((time.perf_counter() - started_perf) * 1000, 2)
return finished_at, duration_ms
def append_tool_log(
tool_log: list[dict],
name: str,
args: dict,
response: object,
*,
started_at: str | None = None,
finished_at: str | None = None,
duration_ms: float | None = None,
) -> None:
entry = {"name": name, "arguments": args, "response": response}
if started_at is not None:
entry["started_at"] = started_at
if finished_at is not None:
entry["finished_at"] = finished_at
if duration_ms is not None:
entry["duration_ms"] = duration_ms
tool_log.append(entry)
def log_timed_call(
tool_log: list[dict],
name: str,
args: dict,
call: Callable[[], T],
) -> T:
started_at, started_perf = _start_timing()
try:
response = call()
except Exception as exc:
finished_at, duration_ms = _finish_timing(started_perf)
append_tool_log(
tool_log,
name,
args,
{"error": str(exc)},
started_at=started_at,
finished_at=finished_at,
duration_ms=duration_ms,
)
raise
finished_at, duration_ms = _finish_timing(started_perf)
append_tool_log(
tool_log,
name,
args,
response,
started_at=started_at,
finished_at=finished_at,
duration_ms=duration_ms,
)
return response
def extract_user_messages(messages: list[dict]) -> list[str]:
user_contents: list[str] = []
for message in messages:
if message.get("role") == "user":
content = message.get("content", "")
if content:
user_contents.append(content)
return user_contents
def retrieve_context(
query_text: str, pg_url: str | None = None
) -> tuple[List[Source], dict]:
embedder_started = time.perf_counter()
embedder, cache_hit = embedding_provider.get_embedder()
embedder_ms = round((time.perf_counter() - embedder_started) * 1000, 2)
url = pg_url or config.get_postgres_url()
retrieval_started = time.perf_counter()
sources = retrieval_store.retrieve(
pg_url=url,
embedder=embedder,
query=query_text,
k=8,
expand_links=True,
)
retrieval_ms = round((time.perf_counter() - retrieval_started) * 1000, 2)
return sources, {
"embedder_get_ms": embedder_ms,
"embedder_cache_hit": cache_hit,
"retrieval_ms": retrieval_ms,
"retrieve_context_internal_ms": round(embedder_ms + retrieval_ms, 2),
}
def bootstrap_retrieval(sheet: dict, query_text: str, tool_log: list[dict]) -> None:
started_at, started_perf = _start_timing()
sources, timing = retrieve_context(query_text=query_text)
finished_at, duration_ms = _finish_timing(started_perf)
source_dump = {
"sources": [source.to_string() for source in sources],
"timing": timing,
}
append_tool_log(
tool_log,
"retrieve_context",
{"query": query_text},
source_dump,
started_at=started_at,
finished_at=finished_at,
duration_ms=duration_ms,
)
context_store.update_retrieval_context(sheet, sources)
def init_chat_state(messages: list[dict], draft: str | None = None) -> ChatState:
if not messages:
raise ValueError("messages required")
user_messages = extract_user_messages(messages)
if not user_messages:
raise ValueError("last user message required")
last_user = user_messages[-1]
chat_id = context_store.get_chat_id(messages, draft=draft)
new_chat = is_new_chat(messages)
sheet = context_store.load_sheet(chat_id)
if new_chat or not sheet:
sheet = context_store.init_sheet(chat_id, messages)
context_store.update_history(sheet, messages)
return ChatState(
messages=messages,
draft=draft,
chat_id=chat_id,
new_chat=new_chat,
sheet=sheet,
tool_log=[],
last_user=last_user,
)
def finalize_response(state: ChatState, reply: str | None) -> dict:
if not reply:
reply = "Dazu steht nichts im Material"
else:
decoded, _ = referenz_decoder.decode_references(
reply, context_store.get_retrieval(state.sheet)
)
reply = decoded
context_store.save_sheet(state.sheet)
tool_logging.write_tool_log(
state.tool_log,
created_at=state.sheet.get("created_at"),
chat_id=state.sheet.get("chat_id"),
)
return {"reply": reply, "sources": state.sheet.get("sources", []), "tool_log": state.tool_log}
def run_chat_common(
messages: list[dict],
draft: str | None,
on_bootstrap: Callable[[ChatState, str], None],
on_turn_logic: Callable[[ChatState], None],
on_build_reply: Callable[[ChatState], str | None],
) -> dict:
init_started_at, init_started_perf = _start_timing()
state = init_chat_state(messages, draft)
init_finished_at, init_duration_ms = _finish_timing(init_started_perf)
append_tool_log(
state.tool_log,
"orchestrator_init_state",
{"message_count": len(messages), "has_draft": draft is not None},
{"chat_id": state.chat_id, "new_chat": state.new_chat},
started_at=init_started_at,
finished_at=init_finished_at,
duration_ms=init_duration_ms,
)
if state.new_chat or not state.sheet.get("initialized"):
log_timed_call(
state.tool_log,
"orchestrator_bootstrap",
{"query": state.last_user},
lambda: on_bootstrap(state, state.last_user),
)
state.sheet["initialized"] = True
log_timed_call(
state.tool_log,
"orchestrator_turn_logic",
{"new_chat": state.new_chat},
lambda: on_turn_logic(state),
)
reply = log_timed_call(
state.tool_log,
"orchestrator_build_reply",
{"chat_id": state.chat_id},
lambda: on_build_reply(state),
)
return finalize_response(state, reply)
math-tutor/backend/app/deterministic_services/orchestrators/orchestrator_tutor.py
View file @ e31efbb9
from app.deterministic_services import context_store, retrieval_service, tool_logging
from app.deterministic_services import referenz_decoder
from app.tools import math_tool
from app.LLM_services import hint_LLM, decision_LLM, math_intent_LLM, solver_LLM
from typing import List
from __future__ import annotations
from app.LLM_services import decision_LLM, hint_LLM, math_intent_LLM, solver_LLM
from app.deterministic_services import context_store
from app.deterministic_services.orchestrators import orchestrator_base as base
def _is_new_chat(messages: list[dict]) -> bool:
return not any(m.get("role") == "assistant" for m in messages)
def _append_tool_log(tool_log: list[dict], name: str, args: dict, response: object) -> None:
tool_log.append({"name": name, "arguments": args, "response": response})
def _bootstrap_context(sheet: dict, query_text: str, tool_log: list[dict]) -> None:
# Erstelle einen neuen Retrieval-Block oder aktualisiere den bestehenden
sources = retrieval_service.retrieve_context(query_text=query_text)
retrievals = sheet.get("retrieval_contexts", [])
if retrievals:
context_store.update_retrieval_context(sheet, query_text, sources)
_append_tool_log(tool_log, "update_retrieve_context", {"query": query_text}, {"sources": [source.to_string() for source in sources]})
else:
context_store.add_retrieval_context(sheet, query_text, sources)
_append_tool_log(tool_log, "retrieve_context", {"query": query_text}, {"sources": [source.to_string() for source in sources]})
math_solution = math_intent_LLM.solve_with_tools(query_text)
def _on_bootstrap(state: base.ChatState, query_text: str) -> None:
base.bootstrap_retrieval(state.sheet, query_text, state.tool_log)
math_solution = base.log_timed_call(
state.tool_log,
"math_intent_LLM",
{"query": query_text},
lambda: math_intent_LLM.solve_with_tools(query_text),
)
if math_solution:
context_store.add_math_solution(sheet, math_solution)
_append_tool_log(tool_log, "math_intent_LLM", {"query": query_text}, math_solution)
def _extract_user_messages(messages: list[dict]) -> list[str]:
"""
Extrahiert alle Inhalte von Messages mit der Rolle 'user' und gibt sie als Liste von Strings zurück.
Der letzte user-Content wird als letztes Element in der Liste enthalten sein.
"""
user_contents = []
for message in messages:
if message.get("role") == "user":
content = message.get("content", "")
if content: # Nur hinzufügen, wenn Inhalt vorhanden ist
user_contents.append(content)
return user_contents
def run_chat(messages: list[dict], draft: str | None = None) -> dict:
# Input-Fehlerbehandlung
if not messages:
raise ValueError("messages required")
last_user = _extract_user_messages(messages)[-1] if _extract_user_messages(messages) else None
if not last_user:
raise ValueError("last user message required")
# Context Store Mangement
chat_id = context_store.get_chat_id(messages, draft=draft)
new_chat = _is_new_chat(messages)
sheet = context_store.load_sheet(chat_id)
if new_chat or not sheet:
sheet = context_store.init_sheet(chat_id, messages)
context_store.add_math_solution(state.sheet, math_solution)
context_store.update_history(sheet, messages)
tool_log: list[dict] = []
if new_chat or not sheet.get("initialized"):
_bootstrap_context(sheet, last_user, tool_log)
sheet["initialized"] = True
def _on_turn_logic(state: base.ChatState) -> None:
sheet_text = context_store.format_sheet(state.sheet)
history_text = context_store.format_history(messages)
sheet_text = context_store.format_sheet(sheet)
if new_chat:
llm_solution = solver_LLM.solve_question(last_user, sheet_text)
_append_tool_log(tool_log, "LLM_Solution", {"question": last_user, "sheet": sheet_text}, llm_solution)
context_store.add_LLM_solution(sheet, llm_solution)
else:
decision = decision_LLM.needs_more_context(history_text, sheet_text)
_append_tool_log(tool_log, "decision", {"sheet": sheet_text}, decision)
context_store.add_decision(sheet, decision)
if decision.get("needs_more_context"):
full_query = ("\n").join(_extract_user_messages(messages))
_bootstrap_context(sheet, full_query, tool_log)
# Hinweis und Ausgaben generierung
hint_args = {
"query": last_user if not new_chat else None,
"task": context_store.get_task(sheet),
"LLM_solution": context_store.last_LLM_solution(sheet),
"math_solution": context_store.first_math_solution(sheet),
"history": context_store.format_history(messages),
"sources": "\n".join([source.to_string() for source in context_store.get_retrieval(sheet)]),
if state.new_chat:
llm_solution = base.log_timed_call(
state.tool_log,
"LLM_Solution",
{"question": state.last_user, "sheet": sheet_text},
lambda: solver_LLM.solve_question(state.last_user, sheet_text),
)
context_store.add_LLM_solution(state.sheet, llm_solution)
return
decision = base.log_timed_call(
state.tool_log,
"decision",
{"sheet": sheet_text},
lambda: decision_LLM.needs_more_context(sheet_text),
)
context_store.add_decision(state.sheet, decision)
if decision.get("needs_more_context"):
full_query = "\n".join(base.extract_user_messages(state.messages))
_on_bootstrap(state, full_query)
def _on_build_reply(state: base.ChatState) -> str | None:
history_turns = context_store.get_history_turns(state.sheet)
args = {
"query": state.last_user if not state.new_chat else None,
"task": context_store.get_task(state.sheet),
"LLM_solution": context_store.last_LLM_solution(state.sheet),
"math_solution": context_store.first_math_solution(state.sheet),
"history": history_turns,
"sources": "\n".join([source.to_string() for source in context_store.get_retrieval(state.sheet)]),
}
reply = hint_LLM.generate_hint(**hint_args)
_append_tool_log(tool_log, "generate_hint", hint_args, reply)
return base.log_timed_call(
state.tool_log,
"generate_hint",
args,
lambda: hint_LLM.generate_hint(**args),
)
if not reply:
reply = "Dazu steht nichts im Material"
else:
decoded, _ = referenz_decoder.decode_references(
reply, context_store.get_retrieval(sheet)
)
reply = decoded
context_store.save_sheet(sheet)
tool_logging.write_tool_log(
tool_log,
created_at=sheet.get("created_at"),
chat_id=sheet.get("chat_id"),
def run_chat(messages: list[dict], draft: str | None = None) -> dict:
return base.run_chat_common(
messages=messages,
draft=draft,
on_bootstrap=_on_bootstrap,
on_turn_logic=_on_turn_logic,
on_build_reply=_on_build_reply,
)
return {"reply": reply, "sources": sheet.get("sources", []), "tool_log": tool_log}
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

Dies ist die Gitlab-Instanz des Transferportals der Hochschule für Technik Stuttgart. Hier geht es zurück zum Portal