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

aufräumen des Context_stores

parent dea78dc1
Hide whitespace changes
Inline Side-by-side
math-tutor/backend/app/deterministic_services/context_store.py
View file @ 04286406
......@@ -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")
......@@ -29,23 +33,6 @@ def get_chat_id(messages: list[dict], draft: str | None = None) -> str:
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 history_turns(messages: list[dict]) -> list[dict]:
turns: list[dict] = []
for msg in messages:
role = msg.get("role")
content = msg.get("content", "")
if role and content:
turns.append({"role": role, "content": content})
return turns
def init_sheet(chat_id: str, messages: list[dict]) -> dict[str, Any]:
timestamp = _utc_now()
return {
......@@ -75,92 +62,19 @@ def load_sheet(chat_id: str) -> dict[str, Any] | None:
return sheet
return None
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 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()
def add_math_solution(
sheet: dict[str, Any],
solution: str,
) -> None:
sheet["math_solutions"].append(
{
"solution": solution,
}
)
sheet["updated_at"] = _utc_now()
def add_LLM_solution(
sheet: dict[str, Any],
solution: str,
) -> None:
sheet["LLM_solutions"].append(
{
"solution": 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)
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")
# 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 get_history_turns(sheet: dict[str, Any]) -> list[dict]:
return history_turns(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", "")
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 = []
......@@ -204,6 +118,51 @@ def format_sheet(sheet: dict[str, Any]) -> str:
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 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()
def get_retrieval(sheet: dict[str, Any]) -> list[Source]:
retrievals = sheet.get("retrieval_contexts", [])
if not retrievals:
......@@ -212,16 +171,52 @@ def get_retrieval(sheet: dict[str, Any]) -> list[Source]:
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)
# ---------------------------------------------------------------------------------------------------
# Math-solution related
# ---------------------------------------------------------------------------------------------------
def add_math_solution(
sheet: dict[str, Any],
solution: str,
) -> None:
sheet["math_solutions"].append(
{
"solution": solution,
}
)
sheet["updated_at"] = _utc_now()
chat_id = sheet.get("chat_id", "unknown")
latest_path = os.path.join(_LOG_DIR, f"{chat_id}_latest.json")
def first_math_solution(sheet: dict[str, Any]) -> str:
if not sheet["math_solutions"]:
return ""
return sheet["math_solutions"][0].get("solution", "")
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
# ---------------------------------------------------------------------------------------------------
# LLM-solution related
# ---------------------------------------------------------------------------------------------------
def add_LLM_solution(
sheet: dict[str, Any],
solution: str,
) -> None:
sheet["LLM_solutions"].append(
{
"solution": solution,
}
)
sheet["updated_at"] = _utc_now()
def last_LLM_solution(sheet: dict[str, Any]) -> str:
if not sheet["LLM_solutions"]:
return ""
return sheet["LLM_solutions"][-1].get("solution", "")
# ---------------------------------------------------------------------------------------------------
# decision related
# ---------------------------------------------------------------------------------------------------
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()
\ No newline at end of file
math-tutor/backend/app/deterministic_services/orchestrators/orchestrator_base.py
View file @ 04286406
......@@ -60,13 +60,8 @@ def bootstrap_retrieval(sheet: dict, query_text: str, tool_log: list[dict]) -> N
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)
context_store.update_retrieval_context(sheet, query_text, sources)
append_tool_log(tool_log, "update_retrieve_context", {"query": query_text}, source_dump)
def init_chat_state(messages: list[dict], draft: str | None = None) -> ChatState:
if not messages:
......
math-tutor/backend/app/deterministic_services/vector_store.py
View file @ 04286406
......@@ -16,9 +16,9 @@ import app.config
embedding_dim = app.config.get_embedding_settings().target_dim
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
# Einlesen der Dokumente
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
@dataclass
class DocRecord:
......@@ -114,9 +114,9 @@ def load_docs(base_dir: Path) -> List[DocRecord]:
return docs
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
# Init der Databse
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
DATABASE_CREATION_SQL = f"""
CREATE EXTENSION IF NOT EXISTS vector;
......@@ -155,9 +155,9 @@ def init_db(pg_url: str) -> None:
register_vector(conn)
conn.commit()
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
# Einfügen der Dokumente und Embeddings
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
UPSERT_SQL = """
INSERT INTO docs (
......@@ -241,9 +241,9 @@ def embed_documents(embedder: EmbeddingLike, texts: List[str]) -> List[List[floa
def embed_query(embedder: EmbeddingLike, text: str) -> List[float]:
return embedder.embed_query(text)
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
# Retrival der Dokumente
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
@dataclass
class Retrieved:
......@@ -460,9 +460,9 @@ def retrieve(
sources = _retrivla_to_sources(retrivla_dict)
return sources
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
# Retrival in Sources umwandeln
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
class Source(BaseModel):
source_id: SourceID
......@@ -534,9 +534,9 @@ def _retrivla_to_sources(retrievd : Dict[str, List[Retrieved]]) -> List[Source]:
))
return sources
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
# Listen für Filterung
# -----------------------------
# --------------------------------------------------------------------------------------------------------------------
def list_sections(pg_url: str) -> List[Dict[str, Any]]:
sql = """
......
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