d07218307f
Nach der Prüfung glättet ein Lektor ohne Werkzeuge jeden Abschnitt (Satzbau, Zusammenhang, Wortwahl, Begriffe erinnern). Eingebaute Verlustprüfung: Bezeichner, Zahlen, Überschriften, Code, Diagramme, Querverweise, Umfang ±20 % – sonst bleibt das Original. Alte Zustände bekommen die Stufe beim Laden nachgetragen. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
94 lines
4.8 KiB
Python
94 lines
4.8 KiB
Python
"""Ein Agent = eine Claude-Agent-SDK-Query mit dem Themenverzeichnis als Arbeitsverzeichnis."""
|
||
import asyncio
|
||
import logging
|
||
|
||
from claude_agent_sdk import (AssistantMessage, ClaudeAgentOptions, ResultMessage, TextBlock, ToolUseBlock, query)
|
||
|
||
from . import config, state
|
||
|
||
log = logging.getLogger("buchwerk")
|
||
|
||
SYSTEM_ZUSATZ = (
|
||
"Du bist ein Agent in einer Pipeline, die ein deutschsprachiges Sachbuch erstellt. Das aktuelle Arbeitsverzeichnis "
|
||
"ist das Buchprojekt; verlasse es nicht (außer für Klone und Downloads, die du unter quellen/ ablegst). "
|
||
"Es gibt keinen menschlichen Ansprechpartner: Rückfragen sind nicht möglich, entscheide selbst. "
|
||
"Arbeite gründlich und bis zum Ende der Aufgabe. Antworte auf Deutsch. Deine letzte Nachricht ist der Abschlussbericht."
|
||
)
|
||
|
||
|
||
def _kurz(block: ToolUseBlock) -> str:
|
||
i = block.input or {}
|
||
for k in ("file_path", "command", "pattern", "query", "url", "description"):
|
||
if k in i:
|
||
return f"{block.name} {str(i[k])[:160]}"
|
||
return block.name
|
||
|
||
|
||
async def laufen(name: str, agent_id: str, stufe_id: str, titel: str, prompt: str, *,
|
||
effort: str = config.EFFORT, max_turns: int = config.MAX_TURNS) -> dict:
|
||
"""Führt einen Agenten aus; schreibt Ereignisse und Bericht. → {"status", "bericht", "kosten", "tokens"}"""
|
||
a = state.agent_eintrag(name, agent_id, stufe_id, titel)
|
||
state.ereignis(name, agent_id, "start", titel)
|
||
options = ClaudeAgentOptions(
|
||
model=config.MODELL, effort=effort, cwd=str(state.verzeichnis(name)),
|
||
permission_mode="bypassPermissions", max_turns=max_turns,
|
||
system_prompt={"type": "preset", "preset": "claude_code", "append": SYSTEM_ZUSATZ},
|
||
# Workflow läuft im Hintergrund und ließ einen Agenten ohne Bericht enden – gesperrt.
|
||
disallowed_tools=["Workflow"],
|
||
)
|
||
letzter_text, kosten, tokens, status, fehler = "", 0.0, 0, "fehler", ""
|
||
try:
|
||
async for m in query(prompt=prompt, options=options):
|
||
if isinstance(m, AssistantMessage):
|
||
for b in m.content:
|
||
if isinstance(b, ToolUseBlock):
|
||
a["werkzeuge"] += 1
|
||
if a["werkzeuge"] % 5 == 1:
|
||
state.ereignis(name, agent_id, "tool", _kurz(b))
|
||
elif isinstance(b, TextBlock) and b.text.strip():
|
||
letzter_text = b.text
|
||
elif isinstance(m, ResultMessage):
|
||
kosten = m.total_cost_usd or 0.0
|
||
u = m.usage or {}
|
||
tokens = sum(int(u.get(k) or 0) for k in
|
||
("input_tokens", "output_tokens", "cache_creation_input_tokens", "cache_read_input_tokens"))
|
||
if m.result and m.result.strip():
|
||
letzter_text = m.result
|
||
if m.is_error or m.subtype != "success":
|
||
fehler = f"{m.subtype}: " + "; ".join(str(e) for e in (m.errors or [])[:3])
|
||
else:
|
||
status = "fertig"
|
||
except asyncio.CancelledError:
|
||
state.agent_fertig(name, a, "abgebrochen", letzter_text, kosten, tokens)
|
||
state.ereignis(name, agent_id, "ende", "abgebrochen")
|
||
raise
|
||
except Exception as e: # ProcessError, CLIConnectionError, …
|
||
fehler = f"{type(e).__name__}: {e}"
|
||
log.exception("Agent %s", agent_id)
|
||
bericht = letzter_text.strip()
|
||
if status != "fertig":
|
||
bericht = f"**Fehler:** {fehler}\n\n{bericht}".strip()
|
||
state.agent_fertig(name, a, status, bericht, kosten, tokens)
|
||
if status == "fertig":
|
||
state.bericht_schreiben(name, agent_id, bericht)
|
||
state.ereignis(name, agent_id, "ende" if status == "fertig" else "fehler",
|
||
f"{status} · {kosten:.2f} USD · {a['werkzeuge']} Werkzeugaufrufe" + (f" · {fehler}" if fehler else ""))
|
||
return {"status": status, "bericht": bericht, "kosten": kosten, "tokens": tokens}
|
||
|
||
|
||
async def fragen(prompt: str, *, system: str, effort: str = "medium") -> tuple[str, float, int]:
|
||
"""Eine einzelne Modellantwort ohne Werkzeuge (für das Lektorat). → (Text, Kosten, Tokens)"""
|
||
options = ClaudeAgentOptions(model=config.MODELL, effort=effort, max_turns=1, allowed_tools=[], system_prompt=system)
|
||
text, kosten, tokens = [], 0.0, 0
|
||
async for m in query(prompt=prompt, options=options):
|
||
if isinstance(m, AssistantMessage):
|
||
for b in m.content:
|
||
if isinstance(b, TextBlock):
|
||
text.append(b.text)
|
||
elif isinstance(m, ResultMessage):
|
||
kosten = m.total_cost_usd or 0.0
|
||
u = m.usage or {}
|
||
tokens = sum(int(u.get(k) or 0) for k in
|
||
("input_tokens", "output_tokens", "cache_creation_input_tokens", "cache_read_input_tokens"))
|
||
return "".join(text).strip(), kosten, tokens
|