Skip to content
GitLab
You need to sign in or sign up before continuing.
Commit 941f470b authored by Kantz's avatar Kantz
Browse files

neuer context store hinzugefügt

parent 4d6b3c71
Hide whitespace changes
Inline Side-by-side
math-tutor/backend/app/deterministic_services/context_store.py
View file @ 941f470b
from __future__ import annotations from __future__ import annotations
import hashlib # Shared/base API
import json from app.deterministic_services.context_stores.context_store_base import (
import os add_decision,
from datetime import datetime format_history,
from threading import Lock get_chat_id,
from typing import Any get_chat_id_value,
get_created_at,
from app.deterministic_services import Source get_decisions,
get_history,
get_history_turns,
# --------------------------------------------------------------------------------------------------- get_initialized,
# Basisfunktionen des Contex-Sheets get_retrieval,
# --------------------------------------------------------------------------------------------------- get_updated_at,
load_sheet,
_CACHE: dict[str, dict[str, Any]] = {} save_sheet,
_LOCK = Lock() set_chat_id,
_LOG_DIR = os.path.join("logs", "context_sheets") set_created_at,
set_decisions,
set_history,
def _utc_now() -> str: set_initialized,
return datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ") set_sources,
set_updated_at,
update_history,
def get_chat_id(messages: list[dict], draft: str | None = None) -> str: update_retrieval_context,
if draft: )
return f"draft_{draft}"
first_user = next((m for m in messages if m.get("role") # Old/open variant is the default for backwards compatibility.
== "user" and m.get("content")), None) from app.deterministic_services.context_stores.context_store_open import (
if not first_user: add_LLM_solution,
return "unknown" add_math_solution,
digest = hashlib.sha1(first_user["content"].encode("utf-8")).hexdigest() first_math_solution,
return digest[:12] format_sheet,
get_llm_solutions,
get_math_solutions,
def init_sheet(chat_id: str, messages: list[dict]) -> dict[str, Any]: get_task,
timestamp = _utc_now() init_sheet,
return { last_LLM_solution,
"chat_id": chat_id, set_llm_solutions,
"created_at": timestamp, set_math_solutions,
"updated_at": timestamp, )
"history": messages[:],
"sources": [], # Expose variant modules so callers can opt in explicitly.
"math_solutions": [], from app.deterministic_services.context_stores import context_store_new, context_store_open
"LLM_solutions": [],
"decisions": [], __all__ = [
"initialized": False, # shared/base
} "get_chat_id",
"load_sheet",
"save_sheet",
def load_sheet(chat_id: str) -> dict[str, Any] | None: "format_history",
with _LOCK: "update_history",
if chat_id in _CACHE: "get_history",
return _CACHE[chat_id] "get_history_turns",
"update_retrieval_context",
latest_path = os.path.join(_LOG_DIR, f"{chat_id}_latest.json") "get_retrieval",
if os.path.exists(latest_path): "add_decision",
with open(latest_path, "r", encoding="utf-8") as f: "set_chat_id",
sheet = json.load(f) "get_chat_id_value",
with _LOCK: "set_created_at",
_CACHE[chat_id] = sheet "get_created_at",
return sheet "set_updated_at",
return None "get_updated_at",
"set_history",
"set_decisions",
def save_sheet(sheet: dict[str, Any]) -> None: "get_decisions",
os.makedirs(_LOG_DIR, exist_ok=True) "set_initialized",
sheet["updated_at"] = _utc_now() "get_initialized",
"set_sources",
chat_id = sheet.get("chat_id", "unknown") # default/open variant
latest_path = os.path.join(_LOG_DIR, f"{chat_id}_latest.json") "init_sheet",
"format_sheet",
payload = json.dumps(sheet, ensure_ascii=True, indent=2) "set_math_solutions",
with open(latest_path, "w", encoding="utf-8") as f: "get_math_solutions",
f.write(payload) "add_math_solution",
"first_math_solution",
with _LOCK: "set_llm_solutions",
_CACHE[chat_id] = sheet "get_llm_solutions",
"add_LLM_solution",
"last_LLM_solution",
def format_sheet(sheet: dict[str, Any]) -> str: "get_task",
parts = [] # explicit variants
history = format_history(sheet.get("history", [])) "context_store_open",
parts.append("HISTORY:\n" + (history or "(leer)")) "context_store_new",
]
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 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],
sources: list[Source],
) -> None:
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(
sheet: dict[str, Any],
solution: str,
) -> None:
sheet["math_solutions"].append(
{
"solution": 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,
) -> 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()
math-tutor/backend/app/deterministic_services/context_stores/__init__.py 0 → 100644
View file @ 941f470b
from app.deterministic_services.context_stores import context_store_base
from app.deterministic_services.context_stores import context_store_new
from app.deterministic_services.context_stores import context_store_open
__all__ = ["context_store_base", "context_store_open", "context_store_new"]
math-tutor/backend/app/deterministic_services/context_stores/context_store_base.py 0 → 100644
View file @ 941f470b
from __future__ import annotations
import hashlib
import json
import os
from datetime import datetime
from threading import Lock
from typing import Any
from app.deterministic_services import Source
_CACHE: dict[str, dict[str, Any]] = {}
_LOCK = Lock()
_LOG_DIR = os.path.join("logs", "context_sheets")
def _utc_now() -> str:
return datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")
def _touch(sheet: dict[str, Any]) -> None:
sheet["updated_at"] = _utc_now()
def touch_sheet(sheet: dict[str, Any]) -> None:
_touch(sheet)
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,
)
if not first_user:
return "unknown"
digest = hashlib.sha1(first_user["content"].encode("utf-8")).hexdigest()
return digest[:12]
def init_sheet_base(chat_id: str, messages: list[dict]) -> dict[str, Any]:
timestamp = _utc_now()
return {
"chat_id": chat_id,
"created_at": timestamp,
"updated_at": timestamp,
"history": messages[:],
"sources": [],
"decisions": [],
"initialized": False,
}
def load_sheet(chat_id: str) -> dict[str, Any] | None:
with _LOCK:
if chat_id in _CACHE:
return _CACHE[chat_id]
latest_path = os.path.join(_LOG_DIR, f"{chat_id}_latest.json")
if os.path.exists(latest_path):
with open(latest_path, "r", encoding="utf-8") as file_handle:
sheet = json.load(file_handle)
with _LOCK:
_CACHE[chat_id] = sheet
return sheet
return None
def save_sheet(sheet: dict[str, Any]) -> None:
os.makedirs(_LOG_DIR, exist_ok=True)
_touch(sheet)
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 file_handle:
file_handle.write(payload)
with _LOCK:
_CACHE[chat_id] = sheet
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 format_sheet_base(sheet: dict[str, Any]) -> str:
parts = []
history = format_history(sheet.get("history", []))
parts.append("HISTORY:\n" + (history or "(leer)"))
sources = get_retrieval(sheet)
if sources:
parts.append("SOURCES:\n" + "\n".join([source.to_string() for source in sources]))
else:
parts.append("SOURCES:\n(leer)")
decisions = get_decisions(sheet)
if decisions:
parts.append("DECISIONS:\n" + json.dumps(decisions, ensure_ascii=True, indent=2))
else:
parts.append("DECISIONS:\n(leer)")
parts.append(f"INITIALIZED:\n{get_initialized(sheet)}")
return "\n\n".join(parts)
def set_chat_id(sheet: dict[str, Any], value: str) -> None:
sheet["chat_id"] = value
_touch(sheet)
def get_chat_id_value(sheet: dict[str, Any]) -> str:
return str(sheet.get("chat_id", ""))
def set_created_at(sheet: dict[str, Any], value: str) -> None:
sheet["created_at"] = value
_touch(sheet)
def get_created_at(sheet: dict[str, Any]) -> str:
return str(sheet.get("created_at", ""))
def set_updated_at(sheet: dict[str, Any], value: str) -> None:
sheet["updated_at"] = value
def get_updated_at(sheet: dict[str, Any]) -> str:
return str(sheet.get("updated_at", ""))
def set_history(sheet: dict[str, Any], messages: list[dict]) -> None:
sheet["history"] = messages[:]
_touch(sheet)
def update_history(sheet: dict[str, Any], messages: list[dict]) -> None:
set_history(sheet, messages)
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 set_decisions(sheet: dict[str, Any], decisions: list[dict[str, Any]]) -> None:
sheet["decisions"] = decisions[:]
_touch(sheet)
def get_decisions(sheet: dict[str, Any]) -> list[dict[str, Any]]:
return sheet.get("decisions", [])
def add_decision(sheet: dict[str, Any], decision: dict[str, Any]) -> None:
entry = {"timestamp": _utc_now(), **decision}
sheet.setdefault("decisions", []).append(entry)
_touch(sheet)
def set_initialized(sheet: dict[str, Any], value: bool) -> None:
sheet["initialized"] = bool(value)
_touch(sheet)
def get_initialized(sheet: dict[str, Any]) -> bool:
return bool(sheet.get("initialized", False))
def set_sources(sheet: dict[str, Any], sources: list[Source]) -> None:
sheet["sources"] = [source.model_dump() for source in sources]
_touch(sheet)
def get_retrieval(sheet: dict[str, Any]) -> list[Source]:
raw_sources = sheet.get("sources", [])
if not raw_sources:
return []
return [Source.model_validate(source) for source in raw_sources]
def update_retrieval_context(
sheet: dict[str, Any],
sources: list[Source],
) -> None:
merged = get_retrieval(sheet) + sources
merged_sorted = sorted(merged, key=lambda item: item.score, reverse=True)[:8]
set_sources(sheet, merged_sorted)
math-tutor/backend/app/deterministic_services/context_stores/context_store_new.py 0 → 100644
View file @ 941f470b
from __future__ import annotations
from typing import Any
from app.deterministic_services.context_stores.context_store_base import (
add_decision,
format_history,
format_sheet_base,
get_chat_id,
get_created_at,
get_decisions,
get_history,
get_history_turns,
get_initialized,
get_retrieval,
get_updated_at,
init_sheet_base,
load_sheet,
save_sheet,
set_chat_id,
set_created_at,
set_decisions,
set_history,
set_initialized,
set_sources,
set_updated_at,
touch_sheet,
update_history,
update_retrieval_context,
)
def init_sheet(chat_id: str, messages: list[dict]) -> dict[str, Any]:
sheet = init_sheet_base(chat_id, messages)
sheet["task"] = ""
sheet["hints"] = []
sheet["solution"] = ""
return sheet
def set_task(sheet: dict[str, Any], value: str) -> None:
sheet["task"] = value
touch_sheet(sheet)
def get_task(sheet: dict[str, Any]) -> str:
return str(sheet.get("task", ""))
def set_hints(sheet: dict[str, Any], hints: list[str]) -> None:
sheet["hints"] = hints[:]
touch_sheet(sheet)
def get_hints(sheet: dict[str, Any]) -> list[str]:
return sheet.get("hints", [])
def add_hint(sheet: dict[str, Any], hint: str) -> None:
sheet.setdefault("hints", []).append(hint)
touch_sheet(sheet)
def set_solution(sheet: dict[str, Any], value: str) -> None:
sheet["solution"] = value
touch_sheet(sheet)
def get_solution(sheet: dict[str, Any]) -> str:
return str(sheet.get("solution", ""))
def format_sheet(sheet: dict[str, Any]) -> str:
parts = [format_sheet_base(sheet)]
task = get_task(sheet)
parts.append("TASK:\n" + (task if task else "(leer)"))
hints = get_hints(sheet)
if hints:
parts.append("HINTS:\n" + "\n".join(hints))
else:
parts.append("HINTS:\n(leer)")
solution = get_solution(sheet)
parts.append("SOLUTION:\n" + (solution if solution else "(leer)"))
return "\n\n".join(parts)
math-tutor/backend/app/deterministic_services/context_stores/context_store_open.py 0 → 100644
View file @ 941f470b
from __future__ import annotations
from typing import Any
from app.deterministic_services.context_stores.context_store_base import (
format_history,
format_sheet_base,
get_chat_id,
get_decisions,
get_history,
get_history_turns,
get_initialized,
get_retrieval,
get_updated_at,
load_sheet,
save_sheet,
set_chat_id,
set_created_at,
set_decisions,
set_history,
set_initialized,
set_sources,
set_updated_at,
touch_sheet,
update_history,
update_retrieval_context,
add_decision,
init_sheet_base,
)
def init_sheet(chat_id: str, messages: list[dict]) -> dict[str, Any]:
sheet = init_sheet_base(chat_id, messages)
sheet["math_solutions"] = []
sheet["LLM_solutions"] = []
return sheet
def set_math_solutions(sheet: dict[str, Any], values: list[dict[str, Any]]) -> None:
sheet["math_solutions"] = values[:]
touch_sheet(sheet)
def get_math_solutions(sheet: dict[str, Any]) -> list[dict[str, Any]]:
return sheet.get("math_solutions", [])
def add_math_solution(
sheet: dict[str, Any],
solution: str,
) -> None:
sheet.setdefault("math_solutions", []).append({"solution": solution})
touch_sheet(sheet)
def first_math_solution(sheet: dict[str, Any]) -> str:
math_solutions = get_math_solutions(sheet)
if not math_solutions:
return ""
return math_solutions[0].get("solution", "")
def set_llm_solutions(sheet: dict[str, Any], values: list[dict[str, Any]]) -> None:
sheet["LLM_solutions"] = values[:]
touch_sheet(sheet)
def get_llm_solutions(sheet: dict[str, Any]) -> list[dict[str, Any]]:
return sheet.get("LLM_solutions", [])
def add_LLM_solution(
sheet: dict[str, Any],
solution: str,
) -> None:
sheet.setdefault("LLM_solutions", []).append({"solution": solution})
touch_sheet(sheet)
def last_LLM_solution(sheet: dict[str, Any]) -> str:
llm_solutions = get_llm_solutions(sheet)
if not llm_solutions:
return ""
return llm_solutions[-1].get("solution", "")
def get_task(sheet: dict[str, Any]) -> str:
history = get_history_turns(sheet)
if history:
return history[-1].get("content", "")
return ""
def format_sheet(sheet: dict[str, Any]) -> str:
parts = [format_sheet_base(sheet)]
math_solutions = get_math_solutions(sheet)
if math_solutions:
blocks = []
for item in math_solutions:
blocks.append(f"SOLUTION: {item.get('solution', '')}")
parts.append("MATH_SOLUTIONS:\n" + "\n\n".join(blocks))
else:
parts.append("MATH_SOLUTIONS:\n(leer)")
llm_solutions = get_llm_solutions(sheet)
if llm_solutions:
blocks = []
for item in llm_solutions:
blocks.append(f"SOLUTION: {item.get('solution', '')}")
parts.append("LLM_SOLUTIONS:\n" + "\n\n".join(blocks))
else:
parts.append("LLM_SOLUTIONS:\n(leer)")
return "\n\n".join(parts)
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