350 lines
15 KiB
Python
350 lines
15 KiB
Python
"""圆桌规划 · Step 1→2 之间「智能体推荐」Gateway 路由。
|
||
|
||
前端页面：推荐弹窗 ``RecommendAgentsDialog``
|
||
前端模块：``frontend-web/src/roundtable-planning/api/recommend.ts``
|
||
|
||
职责
|
||
----
|
||
根据 Step 1 的 ``IntentSummary`` 与前端过滤后的候选池，驱动内置
|
||
``roundtable-recommender``（SOUL 见 ``.deer-flow/agents/roundtable-recommender/``），
|
||
输出 2–6 个 ``{agent_id, reason}`` 推荐席位。
|
||
|
||
设计为**无状态单次调用**：
|
||
- **不暴露** ``/init``：每次请求内部 ``_create_thread``，推荐 agent 不需要历史轮次。
|
||
- 候选池过滤由前端完成（剔除 ``roundtable-intent``、``roundtable-recommender``、
|
||
``roundtable-coordinator-*``），本路由只做校验与 clamp。
|
||
|
||
SSE 协议
|
||
--------
|
||
1. 透传上游 LangGraph 流（前端用 ``parseRecommendStreamContent`` 把正文拆成
|
||
``[RECOMMEND_READY]`` 前的 ``rationale`` 与 JSON 里的 ``picks``，实现卡片增量渲染）。
|
||
2. 末尾状态帧::
|
||
|
||
{ "status": "done"|"asking"|"error", "content": "<全文>", "picks": [...]|null }
|
||
|
||
- ``done``：``picks`` 经 ``_validate_picks`` 后 ≥2 条且 id 均在候选池内；
|
||
- ``asking``：未匹配 ``[RECOMMEND_READY]`` 或校验后不足 2 条（前端走 FALLBACK 默认 6 席）；
|
||
- ``error``：HTTP/未捕获异常。
|
||
|
||
与 SOUL 的契约
|
||
--------------
|
||
模型须在自然语言思路后输出::
|
||
|
||
[RECOMMEND_READY]
|
||
```json
|
||
{"picks": [{"agent_id": "...", "reason": "..."}, ...]}
|
||
```
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
from typing import Any, AsyncIterator
|
||
|
||
import httpx
|
||
from fastapi import APIRouter, Body, HTTPException, Request
|
||
from fastapi.responses import StreamingResponse
|
||
|
||
from app.gateway.roundtable_diag import record_foreground_diag
|
||
from app.gateway.routers._roundtable_seed import ensure_roundtable_functional_agents
|
||
from app.gateway.routers.multi_agent import (
|
||
_ROUNDTABLE_THREAD_METADATA,
|
||
_auth_headers,
|
||
_create_thread,
|
||
_flatten_content,
|
||
_loopback_base,
|
||
_make_loopback_client,
|
||
_parse_last_messages,
|
||
_run_payload,
|
||
_stream_upstream,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter(prefix="/api/recommend", tags=["recommend"])
|
||
|
||
RECOMMENDER_AGENT_ID = "roundtable-recommender"
|
||
|
||
# 从模型最终 AIMessage 全文提取 picks；非贪婪 ``.*?`` 只匹配第一个 json 块。
|
||
_RECOMMEND_READY_PATTERN = re.compile(
|
||
r"\[RECOMMEND_READY\]\s*```json\s*(\{.*?\})\s*```",
|
||
re.DOTALL,
|
||
)
|
||
|
||
# 与 SOUL / 前端 ``RecommendAgentsDialog`` 一致：少于 2 无法成桌，多于 6 由网关截断。
|
||
# 前端确认按钮另限制 2≤N≤8（用户可手动多选），后端此处只负责模型输出校验。
|
||
_MIN_PICKS = 2
|
||
_MAX_PICKS = 6
|
||
|
||
|
||
def _build_prompt(intent: dict[str, Any], candidates: list[dict[str, Any]]) -> str:
|
||
"""Render the intent + candidate pool into the single user message the
|
||
recommender agent reads. Keep the format close to what the SOUL example
|
||
shows so the model has a clean pattern to mirror."""
|
||
objective = str(intent.get("objective") or "").strip() or "(未提供)"
|
||
constraints = intent.get("constraints") or []
|
||
assumptions = intent.get("assumptions") or []
|
||
|
||
constraint_lines = "; ".join(str(c).strip() for c in constraints if str(c).strip()) or "(无)"
|
||
assumption_lines = "; ".join(str(a).strip() for a in assumptions if str(a).strip()) or "(无)"
|
||
|
||
candidate_lines: list[str] = []
|
||
for c in candidates:
|
||
cid = str(c.get("agent_id") or c.get("id") or "").strip()
|
||
cname = str(c.get("name") or "").strip()
|
||
cdesc = str(c.get("description") or "").strip()
|
||
if not cid:
|
||
continue
|
||
# Format: "- <id> (<name>): <description>" — matches the SOUL example.
|
||
suffix = f" ({cname})" if cname else ""
|
||
desc_part = f": {cdesc}" if cdesc else ""
|
||
candidate_lines.append(f"- {cid}{suffix}{desc_part}")
|
||
|
||
candidate_block = "\n".join(candidate_lines) if candidate_lines else "(候选池为空)"
|
||
|
||
return (
|
||
f"任务目标:{objective}\n"
|
||
f"约束:{constraint_lines}\n"
|
||
f"关键假设:{assumption_lines}\n\n"
|
||
f"候选智能体池:\n{candidate_block}\n\n"
|
||
f"请按 SOUL 中规定的格式输出推荐结果(自然语言思路 + [RECOMMEND_READY] + json 块)。"
|
||
)
|
||
|
||
|
||
def _validate_picks(
|
||
raw_picks: Any,
|
||
candidates: list[dict[str, Any]],
|
||
) -> list[dict[str, str]]:
|
||
"""Filter the model's picks against the candidate pool and clamp to 2-6.
|
||
|
||
Drops picks whose ``agent_id`` isn't in the candidate set (guards against
|
||
model hallucination), de-duplicates, and trims to ``_MAX_PICKS`` if the
|
||
model over-picked. Returns ``[]`` if validation leaves us with fewer than
|
||
``_MIN_PICKS`` valid picks — the caller treats that as "fall back to
|
||
defaults".
|
||
|
||
容错策略(2026-XX 修复 "用户自建 agent 被推荐丢失" 的问题):
|
||
模型有时把候选池里的 ``name`` 当成 ``agent_id`` 回填(尤其当 id 是 UUID
|
||
形式时, LLM 倾向用人类可读名),或者顺便也回了 ``name`` 字段。本函数:
|
||
1. 先按 agent_id 命中候选池(严格匹配,原行为);
|
||
2. 否则按 ``agent_id`` 的字面值反查 name → 找回真实 id;
|
||
3. 否则按 ``name`` 字段反查 name → 找回真实 id。
|
||
只要能定位到候选池里**唯一**一个 agent,就算 pick 成功。
|
||
"""
|
||
if not isinstance(raw_picks, list):
|
||
return []
|
||
|
||
id_set: set[str] = set()
|
||
name_to_id: dict[str, str] = {}
|
||
for c in candidates:
|
||
aid = str(c.get("agent_id") or "").strip()
|
||
if not aid:
|
||
continue
|
||
id_set.add(aid)
|
||
nm = str(c.get("name") or "").strip()
|
||
if nm:
|
||
# 大小写不敏感反查 —— 模型可能用空格 / 不同大小写 echo 中文名。
|
||
name_to_id.setdefault(nm.lower(), aid)
|
||
|
||
picked: list[dict[str, str]] = []
|
||
seen: set[str] = set()
|
||
for item in raw_picks:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
raw_id = str(item.get("agent_id") or "").strip()
|
||
raw_name = str(item.get("name") or "").strip()
|
||
reason = str(item.get("reason") or "").strip()
|
||
|
||
aid = ""
|
||
if raw_id and raw_id in id_set:
|
||
aid = raw_id
|
||
elif raw_id and raw_id.lower() in name_to_id:
|
||
# 模型把"广告设计智能体"这种 name 当成 agent_id 回填了
|
||
aid = name_to_id[raw_id.lower()]
|
||
elif raw_name and raw_name.lower() in name_to_id:
|
||
# 模型 agent_id 字段乱了,但 name 字段对得上
|
||
aid = name_to_id[raw_name.lower()]
|
||
|
||
if not aid or aid in seen:
|
||
continue
|
||
picked.append({"agent_id": aid, "reason": reason})
|
||
seen.add(aid)
|
||
if len(picked) >= _MAX_PICKS:
|
||
break
|
||
|
||
if len(picked) < _MIN_PICKS:
|
||
return []
|
||
return picked
|
||
|
||
|
||
@router.post("/stream")
|
||
async def stream_recommend(request: Request, payload: dict = Body(...)) -> StreamingResponse:
|
||
"""Drive one recommendation turn via SSE.
|
||
|
||
Body::
|
||
|
||
{
|
||
"intent": {"objective": "...", "constraints": [...], "assumptions": [...]},
|
||
"candidates": [{"agent_id": "...", "name": "...", "description": "..."}, ...],
|
||
"model": "<可选>" # optional;不传由 lead_agent 回退到 config.yaml models[0]
|
||
}
|
||
|
||
Stream: transparent upstream LangGraph frames, followed by one status
|
||
frame::
|
||
|
||
{
|
||
"status": "done" | "asking" | "error",
|
||
"content": "<final AI text>",
|
||
"picks": [{"agent_id": "...", "reason": "..."}] | null
|
||
}
|
||
|
||
``done`` means picks were parsed and validated; ``asking`` means the
|
||
marker was missing or validation left fewer than 2 valid picks (caller
|
||
should apply its fallback list); ``error`` is upstream failure.
|
||
"""
|
||
intent = payload.get("intent") or {}
|
||
candidates = payload.get("candidates") or []
|
||
# 不写死任何默认模型;空串会让 lead_agent._resolve_model_name() 回退到
|
||
# config.yaml 的 models[0]。内网部署可能没有 deepseek-chat,前端传当前选中的
|
||
# model 名即可,缺失也不会硬错。
|
||
model_name = str(payload.get("model") or "")
|
||
|
||
if not isinstance(intent, dict) or not str(intent.get("objective") or "").strip():
|
||
raise HTTPException(status_code=400, detail="intent.objective is required")
|
||
if not isinstance(candidates, list) or not candidates:
|
||
raise HTTPException(status_code=400, detail="candidates must be a non-empty array")
|
||
|
||
candidate_ids = {
|
||
str(c.get("agent_id") or c.get("id") or "").strip()
|
||
for c in candidates
|
||
if isinstance(c, dict)
|
||
}
|
||
candidate_ids.discard("")
|
||
if not candidate_ids:
|
||
raise HTTPException(status_code=400, detail="candidates must contain at least one valid agent_id")
|
||
|
||
# 调用前自检:内网部署 .deer-flow/ 被 gitignore,roundtable-recommender 目录可能不存在,
|
||
# 缺失则用嵌入模板补建,避免下游 lead_agent 解析失败抛 500。
|
||
ensure_roundtable_functional_agents()
|
||
|
||
prompt = _build_prompt(intent, candidates)
|
||
base = _loopback_base(request)
|
||
headers = _auth_headers(request)
|
||
|
||
async def stream_generator() -> AsyncIterator[str]:
|
||
# TCP loopback 到 127.0.0.1:<gateway port>(详见 multi_agent._make_loopback_client)。
|
||
# 不要在这里替换成 httpx.ASGITransport——曾试过,会整段 buffer SSE 流导致前端
|
||
# 看不到逐 token 增量。
|
||
async with _make_loopback_client(request) as client:
|
||
try:
|
||
# 每次推荐独立 thread；prompt 已包含完整 intent + 候选列表，无需多轮。
|
||
thread_id = await _create_thread(client, base, headers, metadata=_ROUNDTABLE_THREAD_METADATA)
|
||
|
||
body = _run_payload(
|
||
agent_name=RECOMMENDER_AGENT_ID,
|
||
model_name=model_name,
|
||
thread_id=thread_id,
|
||
new_message=prompt,
|
||
# 推荐只做判断：禁止澄清、派活、文件等，避免跑题或二次追问。
|
||
excluded_tools=[
|
||
"web_search",
|
||
"present_files",
|
||
"view_image",
|
||
"agent_orchestration",
|
||
"ask_clarification",
|
||
],
|
||
skill_stop_names=[],
|
||
# 推荐是"从候选池挑 2-6 个"的短分类任务，不需要 CoT。
|
||
# 关掉 thinking + 用 low reasoning，避免模型先吐数千 reasoning
|
||
# tokens 才开始输出 [RECOMMEND_READY]，前端"等待推荐结果..."
|
||
# 从 10-30s 降到 2-5s。
|
||
thinking_enabled=False,
|
||
reasoning_effort="low",
|
||
)
|
||
|
||
captured: list[str] = []
|
||
async for line, done in _stream_upstream(client, base, headers, thread_id, body):
|
||
if line is not None:
|
||
yield line if line.endswith("\n") else line + "\n"
|
||
elif done is not None:
|
||
captured = done
|
||
|
||
messages = _parse_last_messages(captured)
|
||
last_ai_content = ""
|
||
for msg in reversed(messages):
|
||
if not isinstance(msg, dict):
|
||
continue
|
||
if (msg.get("type") or "").lower() not in ("ai", "aimessage", "aimessagechunk"):
|
||
continue
|
||
text = _flatten_content(msg.get("content", "")).strip()
|
||
if text:
|
||
last_ai_content = text
|
||
break
|
||
|
||
# 只扫描「最后一条非空 AIMessage」：推荐 agent 单轮输出，无多轮 READY 问题。
|
||
picks: list[dict[str, str]] = []
|
||
match = _RECOMMEND_READY_PATTERN.search(last_ai_content)
|
||
# 诊断:打印模型输出末尾(含 RECOMMEND_READY 块)。grep "recommend stream debug"
|
||
# 即可看到 LLM 实际推荐了什么、_validate_picks 解析前/后的差异。
|
||
logger.info(
|
||
"recommend stream debug | last_ai_tail=%r | match=%s | candidate_count=%d",
|
||
last_ai_content[-600:] if last_ai_content else "",
|
||
bool(match),
|
||
len(candidates),
|
||
)
|
||
if match:
|
||
try:
|
||
parsed = json.loads(match.group(1))
|
||
except json.JSONDecodeError as exc:
|
||
logger.warning("recommend picks JSON parse failed: %s | raw=%r", exc, match.group(1)[:300])
|
||
parsed = {}
|
||
if isinstance(parsed, dict):
|
||
raw_picks = parsed.get("picks")
|
||
# 传完整 candidates 列表(含 name) —— _validate_picks 容错按 name 反查 id,
|
||
# 修复 LLM 把"广告设计智能体"这种 name 当 agent_id 回填导致全丢的问题。
|
||
picks = _validate_picks(raw_picks, candidates)
|
||
logger.info(
|
||
"recommend stream debug | raw_picks=%r | validated_picks=%r",
|
||
raw_picks,
|
||
picks,
|
||
)
|
||
|
||
# asking ≠ 模型在追问用户，而是「网关未能产出合法 picks」，由前端 fallback。
|
||
status_value = "done" if picks else "asking"
|
||
final_frame: dict[str, Any] = {
|
||
"status": status_value,
|
||
"content": last_ai_content,
|
||
"picks": picks if picks else None,
|
||
}
|
||
yield f"data: {json.dumps(final_frame, ensure_ascii=False)}\n\n"
|
||
|
||
except HTTPException as exc:
|
||
logger.warning("recommend stream HTTPException: %s %s", exc.status_code, exc.detail)
|
||
await record_foreground_diag(
|
||
request, stage="step2_recommend", level="error", event="recommend_http_error",
|
||
message=f"智能体推荐失败 [{exc.status_code}]：{exc.detail}",
|
||
detail={"status_code": exc.status_code, "detail": str(exc.detail)},
|
||
)
|
||
err = {"status": "error", "content": f"[{exc.status_code}] {exc.detail}", "picks": None}
|
||
yield f"data: {json.dumps(err, ensure_ascii=False)}\n\n"
|
||
except Exception as exc:
|
||
logger.exception("recommend stream unexpected failure")
|
||
await record_foreground_diag(
|
||
request, stage="step2_recommend", level="error", event="recommend_unexpected",
|
||
message=f"智能体推荐未预期异常：{exc!r}",
|
||
detail={"type": type(exc).__name__, "error": str(exc)},
|
||
)
|
||
err = {"status": "error", "content": f"unexpected: {exc!r}", "picks": None}
|
||
yield f"data: {json.dumps(err, ensure_ascii=False)}\n\n"
|
||
|
||
return StreamingResponse(
|
||
stream_generator(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no",
|
||
"Connection": "keep-alive",
|
||
},
|
||
)
|