278 lines
11 KiB
Python
278 lines
11 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.routers.multi_agent import (
|
||
_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,
|
||
candidate_ids: set[str],
|
||
) -> 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".
|
||
"""
|
||
if not isinstance(raw_picks, list):
|
||
return []
|
||
|
||
picked: list[dict[str, str]] = []
|
||
seen: set[str] = set()
|
||
for item in raw_picks:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
aid = str(item.get("agent_id") or "").strip()
|
||
reason = str(item.get("reason") or "").strip()
|
||
if not aid or aid not in candidate_ids 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")
|
||
|
||
prompt = _build_prompt(intent, candidates)
|
||
base = _loopback_base(request)
|
||
headers = _auth_headers(request)
|
||
|
||
async def stream_generator() -> AsyncIterator[str]:
|
||
# ASGITransport：进程内直调本机 ASGI app，**不走 TCP / 不经过网络栈**，
|
||
# 因此既无须 trust_env=False 防代理，也不存在 loopback 40s 超时这种坑。
|
||
async with _make_loopback_client(request) as client:
|
||
try:
|
||
# 每次推荐独立 thread；prompt 已包含完整 intent + 候选列表，无需多轮。
|
||
thread_id = await _create_thread(client, base, headers)
|
||
|
||
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=[],
|
||
)
|
||
|
||
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)
|
||
if match:
|
||
try:
|
||
parsed = json.loads(match.group(1))
|
||
except json.JSONDecodeError as exc:
|
||
logger.warning("recommend picks JSON parse failed: %s", exc)
|
||
parsed = {}
|
||
if isinstance(parsed, dict):
|
||
picks = _validate_picks(parsed.get("picks"), candidate_ids)
|
||
|
||
# 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)
|
||
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")
|
||
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",
|
||
},
|
||
)
|