"""Agentic tool loop: stream model output, execute tools (in parallel), loop, persist.""" from __future__ import annotations import asyncio import json import time from collections.abc import Awaitable, Callable from datetime import date from typing import Any from app.core.config import Settings, get_settings from app.core.logging import get_logger from app.core.security import hash_user_id from app.llm.openrouter import LLMError, get_llm from app.llm.prompts import load_prompt from app.llm.router import router from app.llm.schemas import ToolCallReq, ToolResult from app.models import Conversation, Message, User, new_id from app.services import analytics, costs from app.services import conversations as conv_service from app.services import courses as course_service from app.services import files as file_service from app.tools.all import registry from app.tools.registry import ToolContext log = get_logger("agent") Emit = Callable[[str, dict[str, Any]], Awaitable[None]] TOOL_HINTS = """## Outils - `search_course_content` : matière du cours (à consulter d'abord). - `appraisal_calc` : calculateurs d'évaluation déterministes (méthode du coût, ventilation de la dépréciation, coût indexé/unitaire, terrain : extraction/allocation/résiduelle/lotissement, capitalisation directe, MRB, grille de comparables, âge effectif par le marché). - `financial_calc` : six fonctions du dollar, VAN/TRI, âge-vie simple. - `unit_convert` : m² ↔ pi², arpent, acre, hectare, $/m² ↔ $/pi². - `execute_python` : calculs libres, statistiques, régressions, graphiques élaborés (sandbox sans réseau ; les fichiers listés ci-dessous sont dans inputs/, y compris les classeurs déjà générés). - `make_chart` : graphique rapide à partir de données (sans code). - `create_excel` : nouveau classeur à formules vivantes (gabarits : comparables_ajustes, methode_du_cout, age_vie, tableau_amortissement, six_fonctions, sensibilite ; ou spec libre). - `inspect_excel` puis `edit_excel` : lire et MODIFIER un classeur existant (déposé par l'étudiant ou généré plus tôt) — ajouter une colonne/ligne/feuille/graphique, corriger une cellule, formater. Toujours inspecter avant de modifier. - `create_docx` : document Word (fiche de révision, plan d'étude, gabarit de structure de rapport). - `analyze_file` : lire un fichier déposé (xlsx/csv/pdf/image/docx). - `web_search` : données actuelles seulement (marché, taux, coûts, règlements). - `generate_quiz` : quiz interactif. Enchaîne plusieurs outils si nécessaire (ex. : search_course_content → appraisal_calc → create_excel → make_chart). Quand l'étudiant demande de changer un fichier existant, modifie-le avec edit_excel plutôt que d'en recréer un.""" def build_system_prompt(course_code: str, user: User, course_extra: str, announcement: str, files: list[dict[str, Any]], settings: Settings, deadlines: list) -> str: prefs = user.preferences or {} parts = [load_prompt("system_tutor"), load_prompt("guardrails"), load_prompt(f"course_{course_code.lower()}") or f"## Cours actif : {course_code}"] if course_extra: parts.append("## Consignes additionnelles du professeur\n" + course_extra) parts.append(TOOL_HINTS) ctx = [f"Date du jour : {date.today().isoformat()}. Trimestre : {settings.TERM_LABEL}.", f"Étudiant : {user.display_name or 'étudiant'} ; " f"{'tutoiement' if prefs.get('tutoiement', True) else 'vouvoiement'} ; " f"langue : {prefs.get('locale', 'fr-CA')}."] if deadlines: ctx.append("Échéances : " + "; ".join(f"{d.get('label')} ({d.get('date')})" for d in deadlines)) if announcement: ctx.append(f"Annonce du professeur : {announcement}") if files: ctx.append("Fichiers de cette conversation (file_id — nom — origine) ; utilisables avec analyze_file, " "inspect_excel/edit_excel, et disponibles dans inputs/ pour execute_python :\n" + "\n".join(f"- {f['id']} — {f['filename']} ({f['type']}, " f"{'déposé par l’étudiant' if f.get('kind') == 'upload' else 'généré par toi'})" for f in files)) parts.append("## Contexte\n" + "\n".join(ctx)) return "\n\n".join(p for p in parts if p) def history_to_messages(history: list[Message], max_chars: int = 60000) -> list[dict[str, Any]]: """Rebuild OpenAI-style messages, replaying tool calls compactly.""" out: list[dict[str, Any]] = [] for m in history: if m.role == "user": content = m.content if m.attachments: content += "\n\n(Fichiers joints : " + ", ".join(m.attachments) + ")" out.append({"role": "user", "content": content}) elif m.role == "assistant": if m.tool_calls: calls = [{"id": t.id, "type": "function", "function": {"name": t.name, "arguments": json.dumps(t.arguments, ensure_ascii=False)}} for t in m.tool_calls] out.append({"role": "assistant", "content": None, "tool_calls": calls}) for t in m.tool_calls: out.append({"role": "tool", "tool_call_id": t.id, "content": (t.result_summary or "(résultat)")[:1500]}) if m.content: out.append({"role": "assistant", "content": m.content}) # trim from the front to respect budget total = sum(len(json.dumps(x, ensure_ascii=False)) for x in out) while out and total > max_chars: dropped = out.pop(0) total -= len(json.dumps(dropped, ensure_ascii=False)) # never start with a tool message while out and out[0]["role"] == "tool": total -= len(json.dumps(out.pop(0), ensure_ascii=False)) return out class Turn: def __init__(self, conversation: Conversation, user: User, emit: Emit, deep: bool = False, settings: Settings | None = None) -> None: self.conv = conversation self.user = user self.emit = emit self.deep = deep self.settings = settings or get_settings() self.cancel = asyncio.Event() self.message_id = new_id() self.text_parts: list[str] = [] self.tool_records: list[dict[str, Any]] = [] self.usage_total = {"input_tokens": 0, "output_tokens": 0, "cost_usd": 0.0, "model": ""} self.model_used = "" # ------------------------------------------------------------------ helpers async def _tool_ctx(self, tool_call_id: str, file_ids: list[str]) -> ToolContext: async def progress(status: str, detail: str) -> None: await self.emit("tool_progress", {"id": tool_call_id, "status": status, "detail": detail}) return ToolContext(user_id=self.user.id, user_id_hash=hash_user_id(self.user.id), conversation_id=self.conv.id, course_code=self.conv.course_code, settings=self.settings, role=self.user.role, file_ids=file_ids, progress=progress) async def _run_tool(self, call: ToolCallReq, file_ids: list[str]) -> tuple[ToolCallReq, ToolResult]: args = call.arguments() preview = json.dumps({k: v for k, v in args.items() if not k.startswith("__")}, ensure_ascii=False)[:200] await self.emit("tool_call", {"id": call.id, "name": call.name, "args_preview": preview, "arguments": {k: v for k, v in args.items() if not k.startswith("__")}}) ctx = await self._tool_ctx(call.id, file_ids) if call.truncated: result = ToolResult( content=f"Appel {call.name} interrompu : tes arguments ont dépassé la limite de " f"longueur de sortie ({len(call.arguments_json)} caractères reçus). " "Refais un appel plus court : pour create_excel utilise un `template` ou " "réduis le nombre de lignes/feuilles (ou fais une feuille par appel) ; pour " "execute_python écris un code plus concis. N'annonce rien à l'étudiant avant " "que l'appel réussisse.", error=True) else: result = await registry.run(call.name, args, ctx) artifacts = [a.to_dict() for a in result.artifacts] await self.emit("tool_result", {"id": call.id, "name": call.name, "summary": result.summary(), "artifacts": artifacts, "payload": result.payload, "status": "error" if result.error else "ok", "duration_ms": result.meta.get("duration_ms", 0)}) self.tool_records.append({"id": call.id, "name": call.name, "arguments": args, "summary": result.summary(600), "payload": result.payload, "status": "error" if result.error else "ok", "duration_ms": result.meta.get("duration_ms", 0)}) asyncio.create_task(analytics.record_event(hash_user_id(self.user.id), self.conv.course_code, "tool", call.name)) return call, result # ------------------------------------------------------------------ main loop async def run(self, user_text: str, attachments: list[str]) -> None: t0 = time.perf_counter() course = await course_service.get_course(self.conv.course_code) files = await file_service.list_for_conversation(self.conv.id) file_dicts = [{"id": f.id, "filename": f.filename, "type": f.type, "kind": f.kind} for f in files if f.kind == "upload" or f.type in {"xlsx", "csv", "docx"}] file_dicts = file_dicts[-16:] file_ids = [f["id"] for f in file_dicts if f["kind"] == "upload" or f["type"] in {"xlsx", "csv"}] system = build_system_prompt( self.conv.course_code, self.user, course.extra_system_prompt if course else "", course.announcement if course else "", file_dicts, self.settings, (course.syllabus.get("deadlines") if course else []) or []) history = await conv_service.history(self.conv.id, limit=40) messages: list[dict[str, Any]] = [{"role": "system", "content": system}] messages += history_to_messages(history) # the freshly stored user message is already in history; make sure it is the last one if not messages or messages[-1].get("role") != "user": messages.append({"role": "user", "content": user_text}) enabled = None if course and (course.settings or {}).get("tools_enabled"): enabled = set(course.settings["tools_enabled"]) tools = registry.openai_tools(enabled) if course and (course.settings or {}).get("model_primary"): router.overrides["primary"] = course.settings["model_primary"] budget = await costs.budget_status((course.settings or {}).get("budget_usd") if course else None) plan = router.plan("tutor", deep=self.deep) if budget["exceeded"]: await self.emit("warning", {"code": "budget", "message_fr": "Budget mensuel atteint : modèle économique utilisé."}) await self.emit("message_start", {"message_id": self.message_id, "model": plan.models[0]}) llm = get_llm() user_hash = hash_user_id(self.user.id) finish = "stop" try: for _iteration in range(self.settings.LLM_MAX_TOOL_ITERATIONS + 1): if self.cancel.is_set(): finish = "cancelled" break pending: dict[int, ToolCallReq] = {} finish = "stop" call_t0 = time.perf_counter() iteration_text: list[str] = [] async for ev in llm.stream_chat(messages, tools, plan.models, plan.temperature, reasoning=plan.reasoning, user_id_hash=user_hash): if self.cancel.is_set(): finish = "cancelled" break if ev.type == "text_delta": iteration_text.append(ev.data["delta"]) self.text_parts.append(ev.data["delta"]) await self.emit("text_delta", {"delta": ev.data["delta"]}) elif ev.type == "tool_call_end": pending[ev.data["index"]] = ToolCallReq( id=ev.data["id"] or f"call_{new_id()[:8]}", name=ev.data["name"], arguments_json=ev.data["arguments"], index=ev.data["index"]) elif ev.type == "usage": for k in ("input_tokens", "output_tokens", "cost_usd"): self.usage_total[k] += ev.data.get(k, 0) self.model_used = ev.data.get("model") or self.model_used asyncio.create_task(costs.record_usage( "tutor", self.conv.course_code, ev.data, int((time.perf_counter() - call_t0) * 1000))) elif ev.type == "error": raise LLMError(ev.data.get("message", "erreur LLM")) elif ev.type == "done": finish = ev.data.get("finish_reason") or "stop" self.model_used = ev.data.get("model") or self.model_used if finish == "cancelled": break if finish == "length" and pending: # the last tool call was cut by the output limit: never run a truncated call last = max(pending) pending[last].truncated = True log.warning("tool_call_truncated", tool=pending[last].name, chars=len(pending[last].arguments_json)) if not pending: if finish == "length": await self.emit("warning", {"code": "length", "message_fr": "Réponse coupée (limite de longueur)."}) break # append assistant tool-call message, run tools in parallel, append results calls = sorted(pending.values(), key=lambda c: c.index) messages.append({"role": "assistant", "content": "".join(iteration_text) or None, "tool_calls": [{"id": c.id, "type": "function", "function": {"name": c.name, "arguments": c.arguments_json or "{}"}} for c in calls]}) results = await asyncio.gather(*(self._run_tool(c, file_ids) for c in calls)) for call, result in results: messages.append({"role": "tool", "tool_call_id": call.id, "content": result.content[:24000]}) if iteration_text: self.text_parts.append("\n\n") await self.emit("text_delta", {"delta": "\n\n"}) else: await self.emit("warning", {"code": "max_iterations", "message_fr": "Nombre maximal d'étapes atteint."}) except LLMError as exc: log.error("llm_error", error=str(exc)) await self.emit("error", {"code": "llm", "message_fr": "Le modèle n'a pas pu répondre. Réessaie dans un instant."}) finish = "error" except asyncio.CancelledError: finish = "cancelled" finally: content = "".join(self.text_parts).strip() latency = int((time.perf_counter() - t0) * 1000) if content or self.tool_records: try: await asyncio.shield(conv_service.save_assistant_message( self.conv.id, self.message_id, content, self.model_used or plan.models[0], self.usage_total, latency, self.tool_records)) except Exception as exc: # noqa: BLE001 log.warning("persist_failed", error=str(exc)) try: await self.emit("usage", {**self.usage_total, "model_used": self.model_used, "latency_ms": latency}) await self.emit("done", {"message_id": self.message_id, "finish_reason": finish}) except Exception: # noqa: BLE001 — client already gone pass # ------------------------------------------------------------------ side tasks async def generate_title(conv_id: str, first_message: str, emit: Emit | None = None) -> str | None: try: plan = router.plan("fast") text, usage = await get_llm().complete( [{"role": "system", "content": load_prompt("title_generator")}, {"role": "user", "content": first_message[:1500]}], plan.models, 0.0, 30) title = text.strip().strip('"').strip("«»").strip()[:80] if title: await conv_service.set_title(conv_id, title) asyncio.create_task(costs.record_usage("title", "", usage, 0)) if emit: await emit("title", {"conversation_id": conv_id, "title": title}) return title except Exception as exc: # noqa: BLE001 log.warning("title_failed", error=str(exc)) return None