1"""Agentic tool loop: stream model output, execute tools (in parallel), loop, persist."""23from __future__ import annotations45import asyncio6import json7import time8from collections.abc import Awaitable, Callable9from datetime import date10from typing import Any1112from app.core.config import Settings, get_settings13from app.core.logging import get_logger14from app.core.security import hash_user_id15from app.llm.openrouter import LLMError, get_llm16from app.llm.prompts import load_prompt17from app.llm.router import router18from app.llm.schemas import ToolCallReq, ToolResult19from app.models import Conversation, Message, User, new_id20from app.services import analytics, costs21from app.services import conversations as conv_service22from app.services import courses as course_service23from app.services import files as file_service24from app.tools.all import registry25from app.tools.registry import ToolContext2627log = get_logger("agent")28Emit = Callable[[str, dict[str, Any]], Awaitable[None]]2930TOOL_HINTS = """## Outils31- `search_course_content` : matière du cours (à consulter d'abord).32- `execute_python` : calculs, statistiques, graphiques (sandbox, sans réseau).33- `financial_calc` : six fonctions du dollar, âge-vie, coût unitaire, extraction du marché.34- `create_excel` : classeur à formules vivantes (gabarits : methode_du_cout, comparables_ajustes, age_vie, tableau_amortissement, six_fonctions, sensibilite).35- `web_search` : données actuelles seulement (marché, taux, coûts, règlements).36- `analyze_file` : fichiers déposés (identifiants ci-dessous).37- `generate_quiz` : quiz interactif.38Enchaîne plusieurs outils si nécessaire (ex. : search_course_content puis execute_python puis create_excel)."""394041def build_system_prompt(course_code: str, user: User, course_extra: str, announcement: str,42 files: list[dict[str, Any]], settings: Settings, deadlines: list) -> str:43 prefs = user.preferences or {}44 parts = [load_prompt("system_tutor"), load_prompt("guardrails"),45 load_prompt(f"course_{course_code.lower()}") or f"## Cours actif : {course_code}"]46 if course_extra:47 parts.append("## Consignes additionnelles du professeur\n" + course_extra)48 parts.append(TOOL_HINTS)49 ctx = [f"Date du jour : {date.today().isoformat()}. Trimestre : {settings.TERM_LABEL}.",50 f"Étudiant : {user.display_name or 'étudiant'} ; "51 f"{'tutoiement' if prefs.get('tutoiement', True) else 'vouvoiement'} ; "52 f"langue : {prefs.get('locale', 'fr-CA')}."]53 if deadlines:54 ctx.append("Échéances : " + "; ".join(f"{d.get('label')} ({d.get('date')})" for d in deadlines))55 if announcement:56 ctx.append(f"Annonce du professeur : {announcement}")57 if files:58 ctx.append("Fichiers déposés dans cette conversation (file_id — nom) :\n" +59 "\n".join(f"- {f['id']} — {f['filename']} ({f['type']})" for f in files))60 parts.append("## Contexte\n" + "\n".join(ctx))61 return "\n\n".join(p for p in parts if p)626364def history_to_messages(history: list[Message], max_chars: int = 60000) -> list[dict[str, Any]]:65 """Rebuild OpenAI-style messages, replaying tool calls compactly."""66 out: list[dict[str, Any]] = []67 for m in history:68 if m.role == "user":69 content = m.content70 if m.attachments:71 content += "\n\n(Fichiers joints : " + ", ".join(m.attachments) + ")"72 out.append({"role": "user", "content": content})73 elif m.role == "assistant":74 if m.tool_calls:75 calls = [{"id": t.id, "type": "function",76 "function": {"name": t.name, "arguments": json.dumps(t.arguments,77 ensure_ascii=False)}}78 for t in m.tool_calls]79 out.append({"role": "assistant", "content": None, "tool_calls": calls})80 for t in m.tool_calls:81 out.append({"role": "tool", "tool_call_id": t.id,82 "content": (t.result_summary or "(résultat)")[:1500]})83 if m.content:84 out.append({"role": "assistant", "content": m.content})85 # trim from the front to respect budget86 total = sum(len(json.dumps(x, ensure_ascii=False)) for x in out)87 while out and total > max_chars:88 dropped = out.pop(0)89 total -= len(json.dumps(dropped, ensure_ascii=False))90 # never start with a tool message91 while out and out[0]["role"] == "tool":92 total -= len(json.dumps(out.pop(0), ensure_ascii=False))93 return out949596class Turn:97 def __init__(self, conversation: Conversation, user: User, emit: Emit,98 deep: bool = False, settings: Settings | None = None) -> None:99 self.conv = conversation100 self.user = user101 self.emit = emit102 self.deep = deep103 self.settings = settings or get_settings()104 self.cancel = asyncio.Event()105 self.message_id = new_id()106 self.text_parts: list[str] = []107 self.tool_records: list[dict[str, Any]] = []108 self.usage_total = {"input_tokens": 0, "output_tokens": 0, "cost_usd": 0.0, "model": ""}109 self.model_used = ""110111 # ------------------------------------------------------------------ helpers112 async def _tool_ctx(self, tool_call_id: str, file_ids: list[str]) -> ToolContext:113 async def progress(status: str, detail: str) -> None:114 await self.emit("tool_progress", {"id": tool_call_id, "status": status, "detail": detail})115116 return ToolContext(user_id=self.user.id, user_id_hash=hash_user_id(self.user.id),117 conversation_id=self.conv.id, course_code=self.conv.course_code,118 settings=self.settings, role=self.user.role, file_ids=file_ids,119 progress=progress)120121 async def _run_tool(self, call: ToolCallReq, file_ids: list[str]) -> tuple[ToolCallReq, ToolResult]:122 args = call.arguments()123 preview = json.dumps(args, ensure_ascii=False)[:200]124 await self.emit("tool_call", {"id": call.id, "name": call.name, "args_preview": preview,125 "arguments": args})126 ctx = await self._tool_ctx(call.id, file_ids)127 result = await registry.run(call.name, args, ctx)128 artifacts = [a.to_dict() for a in result.artifacts]129 await self.emit("tool_result", {"id": call.id, "name": call.name,130 "summary": result.summary(), "artifacts": artifacts,131 "payload": result.payload, "status": "error" if result.error else "ok",132 "duration_ms": result.meta.get("duration_ms", 0)})133 self.tool_records.append({"id": call.id, "name": call.name, "arguments": args,134 "summary": result.summary(600), "payload": result.payload,135 "status": "error" if result.error else "ok",136 "duration_ms": result.meta.get("duration_ms", 0)})137 asyncio.create_task(analytics.record_event(hash_user_id(self.user.id),138 self.conv.course_code, "tool", call.name))139 return call, result140141 # ------------------------------------------------------------------ main loop142 async def run(self, user_text: str, attachments: list[str]) -> None:143 t0 = time.perf_counter()144 course = await course_service.get_course(self.conv.course_code)145 files = await file_service.list_for_conversation(self.conv.id)146 file_dicts = [{"id": f.id, "filename": f.filename, "type": f.type} for f in files147 if f.kind == "upload"]148 file_ids = [f["id"] for f in file_dicts]149 system = build_system_prompt(150 self.conv.course_code, self.user, course.extra_system_prompt if course else "",151 course.announcement if course else "", file_dicts, self.settings,152 (course.syllabus.get("deadlines") if course else []) or [])153 history = await conv_service.history(self.conv.id, limit=40)154 messages: list[dict[str, Any]] = [{"role": "system", "content": system}]155 messages += history_to_messages(history)156 # the freshly stored user message is already in history; make sure it is the last one157 if not messages or messages[-1].get("role") != "user":158 messages.append({"role": "user", "content": user_text})159160 enabled = None161 if course and (course.settings or {}).get("tools_enabled"):162 enabled = set(course.settings["tools_enabled"])163 tools = registry.openai_tools(enabled)164 if course and (course.settings or {}).get("model_primary"):165 router.overrides["primary"] = course.settings["model_primary"]166 budget = await costs.budget_status((course.settings or {}).get("budget_usd") if course else None)167 plan = router.plan("tutor", deep=self.deep)168 if budget["exceeded"]:169 await self.emit("warning", {"code": "budget", "message_fr":170 "Budget mensuel atteint : modèle économique utilisé."})171 await self.emit("message_start", {"message_id": self.message_id, "model": plan.models[0]})172173 llm = get_llm()174 user_hash = hash_user_id(self.user.id)175 finish = "stop"176 try:177 for _iteration in range(self.settings.LLM_MAX_TOOL_ITERATIONS + 1):178 if self.cancel.is_set():179 finish = "cancelled"180 break181 pending: dict[int, ToolCallReq] = {}182 finish = "stop"183 call_t0 = time.perf_counter()184 iteration_text: list[str] = []185 async for ev in llm.stream_chat(messages, tools, plan.models, plan.temperature,186 reasoning=plan.reasoning, user_id_hash=user_hash):187 if self.cancel.is_set():188 finish = "cancelled"189 break190 if ev.type == "text_delta":191 iteration_text.append(ev.data["delta"])192 self.text_parts.append(ev.data["delta"])193 await self.emit("text_delta", {"delta": ev.data["delta"]})194 elif ev.type == "tool_call_end":195 pending[ev.data["index"]] = ToolCallReq(196 id=ev.data["id"] or f"call_{new_id()[:8]}", name=ev.data["name"],197 arguments_json=ev.data["arguments"], index=ev.data["index"])198 elif ev.type == "usage":199 for k in ("input_tokens", "output_tokens", "cost_usd"):200 self.usage_total[k] += ev.data.get(k, 0)201 self.model_used = ev.data.get("model") or self.model_used202 asyncio.create_task(costs.record_usage(203 "tutor", self.conv.course_code, ev.data,204 int((time.perf_counter() - call_t0) * 1000)))205 elif ev.type == "error":206 raise LLMError(ev.data.get("message", "erreur LLM"))207 elif ev.type == "done":208 finish = ev.data.get("finish_reason") or "stop"209 self.model_used = ev.data.get("model") or self.model_used210 if finish == "cancelled":211 break212 if not pending:213 break214 # append assistant tool-call message, run tools in parallel, append results215 calls = sorted(pending.values(), key=lambda c: c.index)216 messages.append({"role": "assistant",217 "content": "".join(iteration_text) or None,218 "tool_calls": [{"id": c.id, "type": "function",219 "function": {"name": c.name,220 "arguments": c.arguments_json or "{}"}}221 for c in calls]})222 results = await asyncio.gather(*(self._run_tool(c, file_ids) for c in calls))223 for call, result in results:224 messages.append({"role": "tool", "tool_call_id": call.id,225 "content": result.content[:24000]})226 if iteration_text:227 self.text_parts.append("\n\n")228 await self.emit("text_delta", {"delta": "\n\n"})229 else:230 await self.emit("warning", {"code": "max_iterations",231 "message_fr": "Nombre maximal d'étapes atteint."})232 except LLMError as exc:233 log.error("llm_error", error=str(exc))234 await self.emit("error", {"code": "llm", "message_fr":235 "Le modèle n'a pas pu répondre. Réessaie dans un instant."})236 finish = "error"237 except asyncio.CancelledError:238 finish = "cancelled"239 finally:240 content = "".join(self.text_parts).strip()241 latency = int((time.perf_counter() - t0) * 1000)242 if content or self.tool_records:243 await conv_service.save_assistant_message(244 self.conv.id, self.message_id, content, self.model_used or plan.models[0],245 self.usage_total, latency, self.tool_records)246 await self.emit("usage", {**self.usage_total, "model_used": self.model_used,247 "latency_ms": latency})248 await self.emit("done", {"message_id": self.message_id, "finish_reason": finish})249250251# ------------------------------------------------------------------ side tasks252async def generate_title(conv_id: str, first_message: str, emit: Emit | None = None) -> str | None:253 try:254 plan = router.plan("fast")255 text, usage = await get_llm().complete(256 [{"role": "system", "content": load_prompt("title_generator")},257 {"role": "user", "content": first_message[:1500]}], plan.models, 0.0, 30)258 title = text.strip().strip('"').strip("«»").strip()[:80]259 if title:260 await conv_service.set_title(conv_id, title)261 asyncio.create_task(costs.record_usage("title", "", usage, 0))262 if emit:263 await emit("title", {"conversation_id": conv_id, "title": title})264 return title265 except Exception as exc: # noqa: BLE001266 log.warning("title_failed", error=str(exc))267 return None268