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

orchestrator standartisiert

parent d8327473
Hide whitespace changes
Inline Side-by-side
math-tutor/backend/app/api/canvas.py
View file @ e5dc46a3
......@@ -13,7 +13,6 @@ from mpxpy.mathpix_client import MathpixClient
router = APIRouter()
settings = config.get_mathpix_settings()
mathpix_client = None
......
math-tutor/backend/app/api/chat.py
View file @ e5dc46a3
......@@ -14,7 +14,7 @@ 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
from app.deterministic_services.orchestrators import orchestrator_qa as orchestrator
router = APIRouter()
......
math-tutor/backend/app/deterministic_services/orchestrators/orchestrator_QA.py
View file @ e5dc46a3
from typing import List
from __future__ import annotations
from app.deterministic_services import context_store, tool_logging, vector_store, Source, referenz_decoder
from app.LLM_services import qa_LLM
from app.deterministic_services.embeddings import EmbeddingFactory
from app.deterministic_services import context_store
from app.deterministic_services.orchestrators import orchestrator_base as base
import app.config as config
def _is_new_chat(messages: list[dict]) -> bool:
return not any(m.get("role") == "assistant" for m in messages)
def _on_bootstrap(state: base.ChatState, query_text: str) -> None:
base.bootstrap_retrieval(state.sheet, query_text, state.tool_log)
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 _on_turn_logic(state: base.ChatState) -> None:
return None
def _bootstrap_context(sheet: dict, query_text: str, tool_log: list[dict]) -> None:
# Erstelle einen neuen Retrieval-Block oder aktualisiere den bestehenden
sources = _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
def _retrieve_context(query_text: str, pg_url: str | None = None) -> List[Source]:
embedder = EmbeddingFactory.create(config.get_embedding_settings())
url = pg_url or config.get_postgres_url()
sources = vector_store.retrieve(
pg_url=url,
embedder=embedder,
query=query_text,
k=8,
expand_links=True,
)
return sources
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.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
# 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)
reply = qa_LLM.answer_question(**args)
base.append_tool_log(state.tool_log, "answer_question", args, reply)
return reply
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 @ e5dc46a3
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable, List
import app.config as config
from app.deterministic_services import (
Source,
context_store,
referenz_decoder,
tool_logging,
vector_store,
)
from app.deterministic_services.embeddings import EmbeddingFactory
@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 append_tool_log(tool_log: list[dict], name: str, args: dict, response: object) -> None:
tool_log.append({"name": name, "arguments": args, "response": 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) -> List[Source]:
embedder = EmbeddingFactory.create(config.get_embedding_settings())
url = pg_url or config.get_postgres_url()
return vector_store.retrieve(
pg_url=url,
embedder=embedder,
query=query_text,
k=8,
expand_links=True,
)
def bootstrap_retrieval(sheet: dict, query_text: str, tool_log: list[dict]) -> None:
sources = retrieve_context(query_text=query_text)
retrievals = sheet.get("retrieval_contexts", [])
source_dump = {"sources": [source.to_string() for source in sources]}
if retrievals:
context_store.update_retrieval_context(sheet, query_text, sources)
append_tool_log(tool_log, "update_retrieve_context", {"query": query_text}, source_dump)
else:
context_store.add_retrieval_context(sheet, query_text, sources)
append_tool_log(tool_log, "retrieve_context", {"query": query_text}, source_dump)
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:
state = init_chat_state(messages, draft)
if state.new_chat or not state.sheet.get("initialized"):
on_bootstrap(state, state.last_user)
state.sheet["initialized"] = True
on_turn_logic(state)
reply = on_build_reply(state)
return finalize_response(state, reply)
math-tutor/backend/app/deterministic_services/orchestrators/orchestrator_tutor.py
View file @ e5dc46a3
from typing import List
from __future__ import annotations
from app.deterministic_services import context_store, tool_logging, vector_store, Source, referenz_decoder
from app.LLM_services import hint_LLM, decision_LLM, math_intent_LLM, solver_LLM
from app.deterministic_services.embeddings import EmbeddingFactory
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
import app.config as config
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 = _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 _on_bootstrap(state: base.ChatState, query_text: str) -> None:
base.bootstrap_retrieval(state.sheet, query_text, state.tool_log)
math_solution = 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 _retrieve_context(query_text: str, pg_url: str | None = None) -> List[Source]:
embedder = EmbeddingFactory.create(config.get_embedding_settings())
url = pg_url or config.get_postgres_url()
sources = vector_store.retrieve(
pg_url=url,
embedder=embedder,
query=query_text,
k=8,
expand_links=True,
)
return sources
context_store.add_math_solution(state.sheet, math_solution)
base.append_tool_log(state.tool_log, "math_intent_LLM", {"query": query_text}, math_solution)
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.update_history(sheet, messages)
tool_log: list[dict] = []
def _on_turn_logic(state: base.ChatState) -> None:
history_turns = context_store.history_turns(state.messages)
sheet_text = context_store.format_sheet(state.sheet)
if new_chat or not sheet.get("initialized"):
_bootstrap_context(sheet, last_user, tool_log)
sheet["initialized"] = True
if state.new_chat:
llm_solution = solver_LLM.solve_question(state.last_user, sheet_text)
base.append_tool_log(
state.tool_log,
"LLM_Solution",
{"question": state.last_user, "sheet": sheet_text},
llm_solution,
)
context_store.add_LLM_solution(state.sheet, llm_solution)
return
history_turns = context_store.history_turns(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_turns, sheet_text)
_append_tool_log(tool_log, "decision", {"sheet": sheet_text}, decision)
context_store.add_decision(sheet, decision)
decision = decision_LLM.needs_more_context(history_turns, sheet_text)
base.append_tool_log(state.tool_log, "decision", {"sheet": sheet_text}, decision)
context_store.add_decision(state.sheet, decision)
if decision.get("needs_more_context"):
full_query = ("\n").join(_extract_user_messages(messages))
_bootstrap_context(sheet, full_query, tool_log)
if decision.get("needs_more_context"):
full_query = "\n".join(base.extract_user_messages(state.messages))
_on_bootstrap(state, full_query)
# 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),
def _on_build_reply(state: base.ChatState) -> str | None:
history_turns = context_store.history_turns(state.messages)
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(sheet)]),
"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)
reply = hint_LLM.generate_hint(**args)
base.append_tool_log(state.tool_log, "generate_hint", args, reply)
return reply
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