LuminousRuoxi
94a26227c1
把对话页从固定状态机换成 skill 驱动的 Agent：顾问说什么都原样发给 agent， 由 SKILL.md + 工具决定该聊还是该调工具。工作台（workbenchView）未改动。 后端 - llm.py：新增 stream_chat()（流式 + tools），并带 reasoning 端点自动降级 （gemini-3.7-flash 拒绝 effort:none，首次 400 后锁定重试） - agent.py：最小对话底座内核——skill 注入 + 工具循环 + SSE 事件 - tools.py：8 个工具包住既有能力（analyze/diagnose/scaffold/reference/ translate/recheck/record_note/confirm），延迟 import main 复用已验证的 endpoint 处理器与测试接缝，prompts.py/schemas.py/evidence.py 未改一行 - main.py：POST /api/chat/stream（SSE） skill - skills/hvr-rewrite/SKILL.md：分诊规则、首轮盘点、AI 味清单对齐 blader/humanizer 的 A–E 分类（含强度校准与反误报）、诚实性硬约束 前端 - 流式正文 + 工具卡片 + 选项按钮；工具 payload 直接喂既有渲染函数 （renderUnderstanding/renderDiagnosis/renderRecheck），没有第二套 UI - 删除随 agent 化失效的 runDiagnose / applyUnderstandingCorrection 死代码 验证（真实模型，非 mock） - 65 单测全绿（test_agent.py 30 + test_demo.py 35） - 逐轮实跑：盘点 → 确认 → 诊断 → 复检，SSE 事件序列与 payload kind 均符合契约 - 前端 readSSE 用真实响应字节按 7/64/全量三种切块回放，事件序列一致 - 实测修掉两个只在真跑时暴露的问题：模型调完分析直接调确认工具导致正文为空 （SKILL.md 补「正文先行」硬规则）；伪标题 `**N. 标题**` 让列表判定失败、 短横线漏成字面字符（前端改逐行分组渲染） 未验证：浏览器人工走查（无浏览器自动化环境），仅到「真实 HTTP + 真实字节回放」这一层。
309 lines
13 KiB
Python
309 lines
13 KiB
Python
"""最小对话底座内核。
|
||
|
||
复刻 prodream dreami engine 的三件事（形态与契约，不是那套服务）：
|
||
1. **skill 注入**：SKILL.md 全文拼进 system prompt（固定注入，不做相关性排序）
|
||
—— 与 engine 的 skill injection 同形，业务逻辑留在 host 这一侧。
|
||
2. **工具循环**：LLM ⇄ 工具，直到没有 tool_call 为止。engine 跑的就是这个循环。
|
||
3. **SSE 事件**：产出 transport 无关的事件字典，由 main.py 序列化成 wire 格式。
|
||
|
||
明确不做（demo 不需要，做了只是负担）：计费、鉴权、附件、MongoDB、
|
||
caller assertion、向量相关性排序、独立的 engine 进程。
|
||
|
||
分层原则（决定了失败模式）：
|
||
- 对话层 = 流式纯文本 + tools，不强制 JSON，不重试（失败即报错，用户看得见）
|
||
- 工具层 = 非流式 JSON，走 llm.complete_json 的既有重试兜底（那套是验证过的）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Any, Iterator
|
||
|
||
from llm import LLMError, LlmClient
|
||
from tools import ToolContext, execute, openai_tools
|
||
|
||
logger = logging.getLogger("hvr.agent")
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
SKILL_ROOT = HERE / "skills"
|
||
|
||
# 一次用户输入最多触发几轮 LLM（每轮可能带一次工具调用）。
|
||
# 4 足够走完「判断 → 调工具 → 汇报」；给到 6 是为了容忍模型多调一次工具。
|
||
MAX_TURNS = 6
|
||
# 会话历史上限（条）。工具结果进历史的是紧凑摘要，但仍要有上限兜底。
|
||
MAX_HISTORY = 40
|
||
|
||
# 新会话没有用户输入时的开场（SKILL.md 规定首轮由 agent 主动盘点）
|
||
KICKOFF = "请开始首轮盘点。"
|
||
|
||
_FRONTMATTER = re.compile(r"^---\n.*?\n---\n", re.DOTALL)
|
||
|
||
_SKILL_CACHE: dict[str, str] = {}
|
||
_SESSIONS: dict[str, dict[str, Any]] = {}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# skill 加载
|
||
def load_skill(name: str) -> str:
|
||
"""读 SKILL.md 并剥掉 frontmatter —— 拼进 system prompt 的是正文。
|
||
|
||
frontmatter 是给注册链路读的元数据（engine 用 name/description/tags 做场景
|
||
命中）；这里按名直取，没有命中环节，所以元数据对模型是纯噪声。"""
|
||
if name in _SKILL_CACHE:
|
||
return _SKILL_CACHE[name]
|
||
path = SKILL_ROOT / name / "SKILL.md"
|
||
if not path.is_file():
|
||
raise LLMError(f"找不到 skill：{name}（{path}）")
|
||
text = _FRONTMATTER.sub("", path.read_text(encoding="utf-8")).strip()
|
||
if not text:
|
||
raise LLMError(f"skill 内容为空：{name}")
|
||
# 缓存是刻意的：SKILL.md 是静态资产，会话期间不会变；改完 skill 需重启进程
|
||
_SKILL_CACHE[name] = text
|
||
return text
|
||
|
||
|
||
def _context_block(ctx: ToolContext) -> str:
|
||
"""业务上下文块。原文只在这里注入一次，工具入参不重抄 —— 这是省 token
|
||
与防篡改（模型重抄原文可能改写原文）的关键。"""
|
||
lines: list[str] = ["<essay_context>"]
|
||
lines.append(f"Essay Prompt: {ctx.prompt or '（未提供）'}")
|
||
lines.append(f"Word Limit: {ctx.word_limit or '（未提供）'}")
|
||
lines.append("段落原文：")
|
||
for i, text in enumerate(ctx.paragraphs, start=1):
|
||
lines.append(f"<p{i}>{text}</p{i}>")
|
||
if ctx.rewrite_paragraphs:
|
||
lines.append("顾问在工作台的当前改写稿：")
|
||
for i, text in enumerate(ctx.rewrite_paragraphs, start=1):
|
||
lines.append(f"<p{i}>{text}</p{i}>")
|
||
else:
|
||
lines.append("顾问在工作台的当前改写稿：（尚未开始改写）")
|
||
if ctx.constraints:
|
||
lines.append("顾问全局要求：" + "；".join(ctx.constraints))
|
||
lines.append("</essay_context>")
|
||
|
||
done: list[str] = []
|
||
if ctx.analysis:
|
||
done.append("已完成理解（analyze_essay）")
|
||
if ctx.diagnosis:
|
||
done.append("已完成诊断（diagnose_essay）")
|
||
lines.append("已完成步骤：" + ("；".join(done) if done else "无"))
|
||
return "\n".join(lines)
|
||
|
||
|
||
def build_system_prompt(ctx: ToolContext, skill_name: str = "hvr-rewrite") -> str:
|
||
return f"{load_skill(skill_name)}\n\n---\n\n# 当前业务上下文\n\n{_context_block(ctx)}"
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 会话
|
||
def _new_session(req: dict[str, Any]) -> dict[str, Any]:
|
||
ctx = ToolContext(
|
||
prompt=str(req.get("prompt") or ""),
|
||
word_limit=req.get("word_limit") or None,
|
||
paragraphs=[str(p) for p in (req.get("paragraphs") or [])],
|
||
constraints=[str(c) for c in (req.get("constraints") or [])],
|
||
rewrite_paragraphs=[str(p) for p in (req.get("rewrite_paragraphs") or [])],
|
||
confirmed_anchors=[str(a) for a in (req.get("confirmed_anchors") or [])],
|
||
paragraph_constraints={
|
||
str(k): [str(x) for x in (v or [])]
|
||
for k, v in (req.get("paragraph_constraints") or {}).items()
|
||
},
|
||
# 前端在会话第一轮把已完成的分析/诊断带过来（页面刷新、进程重启后新开会话时
|
||
# 服务端手里没有）。丢了 diagnosis 会让复检的「原 Pattern 是否缓解」失去对照，
|
||
# 丢了 analysis 会让语义锚点变空 —— 都是静默降级，所以这里必须收。
|
||
analysis=req.get("analysis") or None,
|
||
diagnosis=req.get("diagnosis") or None,
|
||
)
|
||
return {"sid": uuid.uuid4().hex[:16], "ctx": ctx, "messages": []}
|
||
|
||
|
||
def get_session(session_id: str | None, req: dict[str, Any]) -> dict[str, Any]:
|
||
"""续接已有会话；没给 id 或 id 不认识就开新会话。
|
||
|
||
id 不认识时开新会话而不是报错：进程重启会清空 _SESSIONS，前端手里的旧 id
|
||
不该让用户卡死。代价是丢历史，但原文由前端每轮重发，业务上下文不丢。"""
|
||
if session_id and session_id in _SESSIONS:
|
||
sess = _SESSIONS[session_id]
|
||
_refresh_ctx(sess["ctx"], req)
|
||
return sess
|
||
sess = _new_session(req)
|
||
_SESSIONS[sess["sid"]] = sess
|
||
# 会话数上限：demo 场景不会有几百个会话，超了丢最早的防内存无限增长
|
||
while len(_SESSIONS) > 200:
|
||
_SESSIONS.pop(next(iter(_SESSIONS)))
|
||
return sess
|
||
|
||
|
||
def _refresh_ctx(ctx: ToolContext, req: dict[str, Any]) -> None:
|
||
"""前端每轮重发业务上下文 —— 顾问可能刚在工作台改完稿、刚补了要求。
|
||
已完成的分析/诊断结果保留，不因重发而丢。"""
|
||
if req.get("paragraphs"):
|
||
ctx.paragraphs = [str(p) for p in req["paragraphs"]]
|
||
if "rewrite_paragraphs" in req:
|
||
ctx.rewrite_paragraphs = [str(p) for p in (req.get("rewrite_paragraphs") or [])]
|
||
if "constraints" in req:
|
||
ctx.constraints = [str(c) for c in (req.get("constraints") or [])]
|
||
if "confirmed_anchors" in req:
|
||
ctx.confirmed_anchors = [str(a) for a in (req.get("confirmed_anchors") or [])]
|
||
if "paragraph_constraints" in req:
|
||
ctx.paragraph_constraints = {
|
||
str(k): [str(x) for x in (v or [])]
|
||
for k, v in (req.get("paragraph_constraints") or {}).items()
|
||
}
|
||
if "prompt" in req:
|
||
ctx.prompt = str(req.get("prompt") or "")
|
||
if "word_limit" in req:
|
||
ctx.word_limit = req.get("word_limit") or None
|
||
|
||
|
||
def drop_session(session_id: str) -> bool:
|
||
return _SESSIONS.pop(session_id, None) is not None
|
||
|
||
|
||
def _trim(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""保留最近 MAX_HISTORY 条，但**不能从 tool 消息开头切**——OpenAI 协议要求
|
||
tool 消息必须紧跟带 tool_calls 的 assistant 消息，悬空的 tool 消息会被拒。"""
|
||
if len(messages) <= MAX_HISTORY:
|
||
return messages
|
||
cut = len(messages) - MAX_HISTORY
|
||
while cut < len(messages) and messages[cut].get("role") == "tool":
|
||
cut += 1
|
||
return messages[cut:]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 事件
|
||
def _ev(name: str, **data: Any) -> dict[str, Any]:
|
||
return {"event": name, "data": data}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 工具循环
|
||
def run_agent_stream(
|
||
sess: dict[str, Any],
|
||
user_message: str,
|
||
client: LlmClient,
|
||
skill_name: str = "hvr-rewrite",
|
||
) -> Iterator[dict[str, Any]]:
|
||
"""跑一轮对话，逐条 yield 事件字典。
|
||
|
||
事件契约（照搬 prodream，见 docs/Agent化改造方案_20260911.md §四）：
|
||
meta / session_id / token / thinking / tool_call / tool_result /
|
||
tool_request_clarify / done / error
|
||
"""
|
||
ctx: ToolContext = sess["ctx"]
|
||
history: list[dict[str, Any]] = sess["messages"]
|
||
trace_id = uuid.uuid4().hex[:16]
|
||
|
||
yield _ev("meta", trace_id=trace_id)
|
||
yield _ev("session_id", session_id=sess["sid"])
|
||
|
||
if not ctx.paragraphs:
|
||
yield _ev("error", code="no_essay", msg="还没有拿到文书原文，请先在左侧粘贴并切分段落")
|
||
return
|
||
|
||
history.append({"role": "user", "content": user_message or KICKOFF})
|
||
messages = [{"role": "system", "content": build_system_prompt(ctx, skill_name)}, *_trim(history)]
|
||
|
||
try:
|
||
for turn in range(MAX_TURNS):
|
||
text_parts: list[str] = []
|
||
tool_calls: list[dict[str, str]] = []
|
||
finish = ""
|
||
for ev in client.stream_chat(messages, tools=openai_tools()):
|
||
kind = ev.get("type")
|
||
if kind == "text":
|
||
text_parts.append(ev["text"])
|
||
yield _ev("token", text=ev["text"])
|
||
elif kind == "thinking":
|
||
yield _ev("thinking", text=ev["text"])
|
||
elif kind == "tool_calls":
|
||
tool_calls = ev["tool_calls"]
|
||
elif kind == "done":
|
||
finish = ev.get("finish_reason") or "stop"
|
||
|
||
if not tool_calls:
|
||
history.append({"role": "assistant", "content": "".join(text_parts)})
|
||
yield _ev("done", finish_reason=finish or "stop")
|
||
return
|
||
|
||
# 工具调用轮：正文通常是空的，仍要落历史（部分模型会给一句过渡语）
|
||
history.append(
|
||
{
|
||
"role": "assistant",
|
||
"content": "".join(text_parts) or None,
|
||
"tool_calls": [
|
||
{
|
||
"id": c["id"],
|
||
"type": "function",
|
||
"function": {"name": c["name"], "arguments": c["arguments"] or "{}"},
|
||
}
|
||
for c in tool_calls
|
||
],
|
||
}
|
||
)
|
||
messages.append(history[-1])
|
||
|
||
for call in tool_calls:
|
||
name = call.get("name") or ""
|
||
try:
|
||
args = json.loads(call.get("arguments") or "{}")
|
||
if not isinstance(args, dict):
|
||
args = {}
|
||
except json.JSONDecodeError:
|
||
args = {}
|
||
logger.warning("tool args 不是合法 JSON：%s", (call.get("arguments") or "")[:200])
|
||
|
||
yield _ev("tool_call", tool=name, tool_call_id=call["id"], args=args)
|
||
|
||
result = execute(name, args, ctx)
|
||
if not result.ok:
|
||
logger.warning("tool failed name=%s error=%s", name, result.error)
|
||
|
||
yield _ev(
|
||
"tool_result",
|
||
tool=name,
|
||
tool_call_id=call["id"],
|
||
ok=result.ok,
|
||
result_summary=result.summary or result.error,
|
||
payload=result.payload,
|
||
)
|
||
|
||
# 进历史的是摘要（或错误原文）——诊断 JSON 有 2.5k tokens，
|
||
# 原样回灌会让每轮成本滚雪球
|
||
tool_msg = {
|
||
"role": "tool",
|
||
"tool_call_id": call["id"],
|
||
"content": json.dumps(
|
||
{"ok": result.ok, "result": result.summary or result.error},
|
||
ensure_ascii=False,
|
||
),
|
||
}
|
||
history.append(tool_msg)
|
||
messages.append(tool_msg)
|
||
|
||
if result.stop:
|
||
# 交互型工具：本轮到此为止，等顾问点选后由下一条消息继续
|
||
logger.warning("agent await user name=%s turn=%s", name, turn)
|
||
yield _ev(
|
||
"tool_request_clarify",
|
||
clarify_id=call["id"],
|
||
question=(result.clarify or {}).get("question", ""),
|
||
choices=(result.clarify or {}).get("choices", []),
|
||
)
|
||
yield _ev("done", finish_reason="await_user")
|
||
return
|
||
|
||
# 轮次用尽：不是错误，但要如实告诉前端为什么停
|
||
logger.warning("agent hit MAX_TURNS=%s", MAX_TURNS)
|
||
yield _ev("error", code="max_turns", msg=f"本次操作步骤过多（超过 {MAX_TURNS} 轮），请把要求拆开再问")
|
||
except LLMError as exc:
|
||
logger.warning("agent llm error: %s", exc)
|
||
yield _ev("error", code="llm_error", msg=str(exc))
|
||
finally:
|
||
sess["messages"] = _trim(history)
|