feat(llm): Web-Suche per Tool-Calling (Weg 2) — Sonar + ToolCallingLLM
Frische-/Web-Such-Funktion: Das zentrale Modell entscheidet selbst via web_search-Tool, ob es tagesaktuelle Fakten braucht, holt sie über perplexity/sonar und formuliert die Antwort in Persona (Augment). - SonarTool (app/tools/web_search.py): Fakten via perplexity/sonar, Citations als Metadaten, honest-punt-Sentinel bei Fehler/Timeout. - ToolCallingLLM (app/providers/llm/tool_calling.py): agentischer Loop als LLMProvider; complete() + gestreamtes stream() mit SSE-Tool-Assembler; Persona- + Trigger- + Vorrang-Prompt (Tool-Ergebnis schlaegt Gedaechtnis). - Verdrahtung: Registry-Eintrag openrouter-tools; web_search_enabled (global an, pro Nutzer/Profil abschaltbar) via Route-Layering; build_orchestrator waehlt tool-faehig vs. plain, Fallback-Kette erhalten. - Filler: ephemerer Beruhigungssatz beim Tool-Start (sofort angezeigt UND gesprochen als Satz null), nie in semantic_response/History; on_tool_start defensiv durch die stream()-Kette gefaedelt (kein Bruch bestehender Provider). - Modellwechsel: Standard auf mistralai/mistral-small-3.2-24b-instruct (tool-faehig; im Eval einziger Recall-Gate-Passer). 2501 ist tool-unfaehig. - Eval-Harness (eval/tool_calling/): Datensatz + Runner zur Modellauswahl. Doc: Docs/weg2-tool-calling.md. Tests: 290 gruen. Offen (Schritt 5): Koreferenz-Vorstufe (nl-Pronomen) + Metrik-Zaehler. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
19b3998948
commit
703a695bed
19 changed files with 1470 additions and 9 deletions
|
|
@ -65,13 +65,14 @@ class FallbackLLMProvider(_Chain):
|
|||
self._on_error(name)
|
||||
raise last_exc
|
||||
|
||||
async def stream(self, text, history=None, session_id=None, language=None) -> AsyncIterator[str]:
|
||||
async def stream(self, text, history=None, session_id=None, language=None,
|
||||
**kwargs) -> AsyncIterator[str]:
|
||||
last_exc = None
|
||||
for index, (name, provider) in enumerate(self.entries):
|
||||
produced = False
|
||||
try:
|
||||
async for delta in provider.stream(
|
||||
text, history=history, session_id=session_id, language=language
|
||||
text, history=history, session_id=session_id, language=language, **kwargs
|
||||
):
|
||||
produced = True
|
||||
yield delta
|
||||
|
|
|
|||
|
|
@ -70,9 +70,11 @@ class LLMProvider(ABC):
|
|||
history: list[dict] | None = None,
|
||||
session_id: str | None = None,
|
||||
language: str | None = None,
|
||||
**kwargs,
|
||||
) -> AsyncIterator[str]:
|
||||
"""Token-Stream. Default: kein echtes Streaming -> komplette Antwort als ein Chunk.
|
||||
|
||||
Provider mit SSE-Unterstuetzung ueberschreiben diese Methode.
|
||||
Provider mit SSE-Unterstuetzung ueberschreiben diese Methode. Unbekannte
|
||||
kwargs (z. B. on_tool_start) werden ignoriert.
|
||||
"""
|
||||
yield await self.complete(text, history=history, session_id=session_id, language=language)
|
||||
|
|
|
|||
|
|
@ -102,6 +102,7 @@ class LocalOpenAICompatibleLLM(LLMProvider):
|
|||
history: list[dict] | None = None,
|
||||
session_id: str | None = None,
|
||||
language: str | None = None,
|
||||
**kwargs, # z. B. on_tool_start — hier ignoriert (kein Tool-Calling)
|
||||
) -> AsyncIterator[str]:
|
||||
async with httpx.AsyncClient(timeout=120) as client:
|
||||
async with client.stream(
|
||||
|
|
|
|||
|
|
@ -175,6 +175,7 @@ class OpenRouterLLMProvider(LLMProvider):
|
|||
history: list[dict] | None = None,
|
||||
session_id: str | None = None,
|
||||
language: str | None = None,
|
||||
**kwargs, # z. B. on_tool_start — hier ignoriert (kein Tool-Calling)
|
||||
) -> AsyncIterator[str]:
|
||||
payload = {
|
||||
"model": self.model,
|
||||
|
|
|
|||
284
app/providers/llm/tool_calling.py
Normal file
284
app/providers/llm/tool_calling.py
Normal file
|
|
@ -0,0 +1,284 @@
|
|||
"""ToolCallingLLM: agentischer Wrapper um ein tool-fähiges OpenRouter-Modell.
|
||||
|
||||
Implementiert die LLMProvider-Schnittstelle und wickelt intern die Tool-Schleife
|
||||
ab (Modell entscheidet via web_search, ob es sucht; Werkzeug liefert Fakten;
|
||||
Modell formuliert die finale Antwort in Persona). Der Orchestrator ruft nur
|
||||
`complete()`/`stream()` und weiß nichts von Tools. Siehe Docs/weg2-tool-calling.md.
|
||||
|
||||
Schritt 2: nur `complete()` (nicht gestreamt). Der ABC-Default `stream()` fällt
|
||||
auf `complete()` zurück, sodass der Loop schon durch den Orchestrator nutzbar ist;
|
||||
echtes Streaming + Filler folgen in Schritt 4.
|
||||
"""
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from collections.abc import AsyncIterator
|
||||
from datetime import date
|
||||
|
||||
import httpx
|
||||
|
||||
from app.providers.llm.base import LLMProvider, lang_instruction, with_lang_reminder
|
||||
from app.providers.llm.openrouter import SYSTEM_PROMPT as PERSONA_PROMPT
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
ENDPOINT = "https://openrouter.ai/api/v1/chat/completions"
|
||||
_RETRY_STATUS = {404, 429, 500, 502, 503}
|
||||
|
||||
# Tool-Schema inkl. Trigger-Kategorien — wortgleich zur im Eval validierten Fassung.
|
||||
WEB_SEARCH_TOOL = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "web_search",
|
||||
"description": (
|
||||
"Look up real-world information that may be newer than your "
|
||||
"knowledge or may have changed since your last update. Call it "
|
||||
"whenever the true answer could plausibly have changed, INCLUDING "
|
||||
"when no exact place or date is named. This covers: who currently "
|
||||
"holds an office; where a living person now lives; whether someone "
|
||||
"is still alive; current prices, rates or crypto; current weather "
|
||||
"or outdoor conditions (even phrased as 'is it cold/raining right "
|
||||
"now', using the user's location); sports results and standings; "
|
||||
"the latest version or model of a product; recent news; and "
|
||||
"time-sensitive logistics such as opening hours, schedules, and "
|
||||
"public-transport or train/bus departure times. Do NOT use it for "
|
||||
"timeless knowledge, opinions, jokes, small talk, the current "
|
||||
"clock time or today's date (you already have those), anything "
|
||||
"you can derive yourself, or anything about the user themselves."
|
||||
),
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"query": {
|
||||
"type": "string",
|
||||
"description": "Concise search query, in the user's language or English.",
|
||||
},
|
||||
},
|
||||
"required": ["query"],
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
# Trigger- + Vorrang-Anweisung. cutoff ist ein bewusst KONSERVATIVER Anker (eher
|
||||
# zu früh als zu spät) — das verschiebt im Zweifel Richtung „suchen", die sichere
|
||||
# Seite (höherer Recall). Im Eval mit genau dieser Formulierung validiert.
|
||||
_TRIGGER_AND_PRIORITY = (
|
||||
"Your knowledge was last updated around {cutoff}. Today is {today}. "
|
||||
"When a question could require information newer than {cutoff} that you "
|
||||
"cannot reason out yourself, call web_search before answering. Never call "
|
||||
"web_search merely to find the current time or today's date — you already "
|
||||
"have them. Tool results are authoritative and newer than your memory: if a "
|
||||
"tool result conflicts with what you believe, follow the tool and never "
|
||||
"contradict it."
|
||||
)
|
||||
|
||||
|
||||
def _system_prompt(cutoff: str, language: str | None) -> str:
|
||||
today = date.today().isoformat()
|
||||
parts = [PERSONA_PROMPT, _TRIGGER_AND_PRIORITY.format(cutoff=cutoff, today=today)]
|
||||
instr = lang_instruction(language)
|
||||
if instr:
|
||||
parts.append(instr)
|
||||
return "\n\n".join(parts)
|
||||
|
||||
|
||||
class _StreamAcc:
|
||||
"""Sammelt einen gestreamten Modell-Schritt: Text-Buffer + (fragmentierte) tool_calls.
|
||||
|
||||
Beim Streaming kommen tool_calls in Bruchstücken (je `index`: id/name einmal,
|
||||
`arguments` über viele Deltas). Hier zusammengesetzt; Text wird zugleich
|
||||
durchgereicht (siehe _stream_chat).
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.text: list[str] = []
|
||||
self._tc: dict[int, dict] = {}
|
||||
|
||||
def add_text(self, s: str) -> None:
|
||||
self.text.append(s)
|
||||
|
||||
def add_tool_fragments(self, frags: list[dict]) -> None:
|
||||
for tc in frags:
|
||||
idx = tc.get("index", 0)
|
||||
entry = self._tc.setdefault(idx, {"id": "", "name": "", "arguments": ""})
|
||||
if tc.get("id"):
|
||||
entry["id"] = tc["id"]
|
||||
fn = tc.get("function") or {}
|
||||
if fn.get("name"):
|
||||
entry["name"] = fn["name"]
|
||||
if fn.get("arguments"):
|
||||
entry["arguments"] += fn["arguments"]
|
||||
|
||||
@property
|
||||
def tool_calls(self) -> list[dict]:
|
||||
return [
|
||||
{"id": e["id"] or f"call_{idx}", "type": "function",
|
||||
"function": {"name": e["name"], "arguments": e["arguments"]}}
|
||||
for idx, e in sorted(self._tc.items())
|
||||
]
|
||||
|
||||
def assistant_message(self) -> dict:
|
||||
"""Assistant-Message zum Re-Threading der tool_calls an das Modell."""
|
||||
return {"role": "assistant", "content": "".join(self.text) or None,
|
||||
"tool_calls": self.tool_calls}
|
||||
|
||||
|
||||
class ToolCallingLLM(LLMProvider):
|
||||
def __init__(self, api_key: str, model: str, tools: list,
|
||||
knowledge_cutoff: str = "fall 2024", max_rounds: int = 3,
|
||||
max_retries: int = 4, temperature: float = 0.3):
|
||||
self.api_key = (api_key or "").strip()
|
||||
self.model = (model or "").strip()
|
||||
self.tools = {t.name: t for t in tools}
|
||||
self.cutoff = knowledge_cutoff
|
||||
self.max_rounds = max(1, max_rounds)
|
||||
self.max_retries = max(1, max_retries)
|
||||
self.temperature = temperature
|
||||
|
||||
def _initial_messages(self, text: str, history, language) -> list[dict]:
|
||||
if not self.api_key:
|
||||
raise ValueError("OPENROUTER_API_KEY is empty")
|
||||
if not self.model:
|
||||
raise ValueError("ToolCallingLLM model is empty")
|
||||
if not text or not text.strip():
|
||||
raise ValueError("LLM input text is empty")
|
||||
messages = [{"role": "system", "content": _system_prompt(self.cutoff, language)}]
|
||||
if history:
|
||||
messages.extend(history)
|
||||
messages.append({"role": "user", "content": with_lang_reminder(text.strip(), language)})
|
||||
return messages
|
||||
|
||||
async def _chat(self, messages: list[dict], allow_tools: bool = True) -> dict:
|
||||
"""Ein Modell-Aufruf; liefert die Assistant-Message (mit/ohne tool_calls)."""
|
||||
payload = {"model": self.model, "messages": messages, "temperature": self.temperature}
|
||||
if allow_tools:
|
||||
payload["tools"] = [WEB_SEARCH_TOOL]
|
||||
payload["tool_choice"] = "auto"
|
||||
timeout = httpx.Timeout(connect=10.0, read=120.0, write=30.0, pool=10.0)
|
||||
headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
|
||||
last_error: Exception | None = None
|
||||
|
||||
for attempt in range(1, self.max_retries + 1):
|
||||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||||
resp = await client.post(ENDPOINT, json=payload, headers=headers)
|
||||
if resp.status_code in _RETRY_STATUS and attempt < self.max_retries:
|
||||
last_error = RuntimeError(f"OpenRouter {resp.status_code}: {resp.text[:200]}")
|
||||
await asyncio.sleep(min(2.0 * attempt, 30.0))
|
||||
continue
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
if "choices" not in data: # transiente Fehler kommen teils als 200 mit {"error":...}
|
||||
if attempt < self.max_retries:
|
||||
last_error = RuntimeError(f"Antwort ohne 'choices': {str(data)[:200]}")
|
||||
await asyncio.sleep(min(2.0 * attempt, 30.0))
|
||||
continue
|
||||
raise RuntimeError(f"OpenRouter-Antwort ohne 'choices': {str(data)[:300]}")
|
||||
return data["choices"][0]["message"]
|
||||
|
||||
raise last_error or RuntimeError("ToolCallingLLM: alle Versuche fehlgeschlagen")
|
||||
|
||||
async def _run_tool(self, tool_call: dict, language: str | None) -> str:
|
||||
"""Führt einen Tool-Aufruf aus und liefert den tool-Message-Inhalt."""
|
||||
fn = tool_call.get("function", {})
|
||||
name = fn.get("name", "")
|
||||
tool = self.tools.get(name)
|
||||
if tool is None:
|
||||
logger.warning("Unbekanntes Tool angefragt: %r", name)
|
||||
return f"ERROR: unknown tool {name!r}."
|
||||
try:
|
||||
args = json.loads(fn.get("arguments") or "{}")
|
||||
except json.JSONDecodeError:
|
||||
args = {}
|
||||
query = str(args.get("query", "")).strip()
|
||||
result = await tool.run(query, language=language)
|
||||
# result.citations -> UI-Bubble (Fast-follow); hier (noch) nicht durchgereicht.
|
||||
return result.text
|
||||
|
||||
async def complete(self, text: str, history: list[dict] | None = None,
|
||||
session_id: str | None = None, language: str | None = None) -> str:
|
||||
messages = self._initial_messages(text, history, language)
|
||||
|
||||
for _ in range(self.max_rounds):
|
||||
msg = await self._chat(messages, allow_tools=True)
|
||||
tool_calls = msg.get("tool_calls")
|
||||
if not tool_calls:
|
||||
content = (msg.get("content") or "").strip()
|
||||
if content:
|
||||
return content
|
||||
break # leer ohne Tool-Call -> finale Runde erzwingen
|
||||
messages.append(msg) # Assistant-Message mit tool_calls (unverändert zurück)
|
||||
for tc in tool_calls:
|
||||
result_text = await self._run_tool(tc, language)
|
||||
messages.append({"role": "tool", "tool_call_id": tc.get("id", ""),
|
||||
"content": result_text})
|
||||
|
||||
# max_rounds erschöpft (oder leer): finale Antwort ohne weitere Tools erzwingen.
|
||||
final = await self._chat(messages, allow_tools=False)
|
||||
content = (final.get("content") or "").strip()
|
||||
if not content:
|
||||
raise RuntimeError("ToolCallingLLM returned an empty response")
|
||||
return content
|
||||
|
||||
async def stream(self, text: str, history: list[dict] | None = None,
|
||||
session_id: str | None = None, language: str | None = None,
|
||||
on_tool_start=None, **kwargs) -> AsyncIterator[str]:
|
||||
"""Gestreamter Loop: Text-Deltas durchreichen; bei tool_call Sonar+nächste Runde.
|
||||
|
||||
Kein Latenz-Regress im Normalfall (kein Tool): die Antwort streamt direkt
|
||||
durch. `on_tool_start(language)` (optional) feuert, bevor ein Tool läuft —
|
||||
Aufhänger für den ephemeren Filler (Schritt 4b).
|
||||
"""
|
||||
messages = self._initial_messages(text, history, language)
|
||||
|
||||
for _ in range(self.max_rounds):
|
||||
acc = _StreamAcc()
|
||||
async for piece in self._stream_chat(messages, allow_tools=True, acc=acc):
|
||||
yield piece
|
||||
if not acc.tool_calls:
|
||||
return # war Text -> fertig (durchgereicht)
|
||||
if on_tool_start is not None:
|
||||
await on_tool_start(language) # 4b: Filler sofort sprechen/anzeigen
|
||||
messages.append(acc.assistant_message())
|
||||
for tc in acc.tool_calls:
|
||||
result_text = await self._run_tool(tc, language)
|
||||
messages.append({"role": "tool", "tool_call_id": tc["id"],
|
||||
"content": result_text})
|
||||
|
||||
# max_rounds erschöpft: finale Runde ohne weitere Tools, gestreamt.
|
||||
acc = _StreamAcc()
|
||||
async for piece in self._stream_chat(messages, allow_tools=False, acc=acc):
|
||||
yield piece
|
||||
|
||||
async def _stream_chat(self, messages: list[dict], allow_tools: bool,
|
||||
acc: _StreamAcc) -> AsyncIterator[str]:
|
||||
"""Ein gestreamter Modell-Call. Yieldet Text-Deltas; füllt acc (Text + tool_calls)."""
|
||||
payload = {"model": self.model, "messages": messages,
|
||||
"temperature": self.temperature, "stream": True}
|
||||
if allow_tools:
|
||||
payload["tools"] = [WEB_SEARCH_TOOL]
|
||||
payload["tool_choice"] = "auto"
|
||||
timeout = httpx.Timeout(connect=10.0, read=120.0, write=30.0, pool=10.0)
|
||||
headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
|
||||
|
||||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||||
async with client.stream("POST", ENDPOINT, json=payload, headers=headers) as resp:
|
||||
if resp.status_code >= 400:
|
||||
body = await resp.aread()
|
||||
raise RuntimeError(
|
||||
f"OpenRouter {resp.status_code}: {body.decode(errors='replace')[:200]}")
|
||||
async for line in resp.aiter_lines():
|
||||
if not line.startswith("data:"):
|
||||
continue
|
||||
data = line[len("data:"):].strip()
|
||||
if not data or data == "[DONE]":
|
||||
continue
|
||||
try:
|
||||
delta = json.loads(data)["choices"][0]["delta"]
|
||||
except (ValueError, KeyError, IndexError, TypeError):
|
||||
continue
|
||||
content = delta.get("content")
|
||||
if content:
|
||||
acc.add_text(content)
|
||||
yield content
|
||||
if delta.get("tool_calls"):
|
||||
acc.add_tool_fragments(delta["tool_calls"])
|
||||
Loading…
Add table
Add a link
Reference in a new issue