接口修改

offline-backend-20260512/backend/app/gateway/app.py

@@ -40,6 +40,7 @@ from app.gateway.routers import (
     multi_agent,
     notifications,
     public_scheduled_tasks,
+    recommend,
     recommended_questions,
     runs,
     scheduled_tasks,
@@ -871,6 +872,11 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # roundtable as its initial intent.
     app.include_router(intent.router)
 
+    # Agent recommendation (between Step 1 and Step 2): given the intent
+    # summary + a frontend-filtered candidate pool, the built-in
+    # roundtable-recommender picks 2-6 agents most relevant to the task.
+    app.include_router(recommend.router)
+
     @app.get("/health", tags=["health"])
     async def health_check() -> dict:
         """Health check endpoint.

offline-backend-20260512/backend/app/gateway/routers/recommend.py

@@ -0,0 +1,256 @@
+"""Agent recommendation endpoint (between Step 1 and Step 2 of the roundtable flow).
+
+Given the Step-1 intent summary (``objective`` + ``constraints`` + ``assumptions``)
+and a frontend-filtered candidate agent pool, drives the built-in
+``roundtable-recommender`` agent to pick 2-6 agents most relevant to the task
+and explain why.
+
+Single one-shot endpoint — no ``/init`` is exposed because the recommendation
+is stateless: the agent never needs the prior turn's history. We create a
+fresh thread internally per request and let the LangGraph runtime clean it up
+in the background. The frontend handles filtering ``roundtable-intent`` /
+``roundtable-recommender`` / ``roundtable-coordinator-*`` out of the candidate
+pool before calling here.
+
+SSE shape is the same family as ``intent.py`` / ``multi_agent.py``:
+transparent passthrough of the upstream LangGraph stream, then a single
+status frame appended at the end so the frontend can finalize.
+"""
+
+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,
+    _parse_last_messages,
+    _run_payload,
+    _stream_upstream,
+)
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/recommend", tags=["recommend"])
+
+RECOMMENDER_AGENT_ID = "roundtable-recommender"
+
+# Matches the SOUL-mandated handoff signal: a [RECOMMEND_READY] line followed
+# by a ```json ... ``` block carrying {"picks": [{agent_id, reason}, ...]}.
+_RECOMMEND_READY_PATTERN = re.compile(
+    r"\[RECOMMEND_READY\]\s*```json\s*(\{.*?\})\s*```",
+    re.DOTALL,
+)
+
+# Mirrors the SOUL constraint so a misbehaving model can't push >6 picks to
+# the frontend or fall below the "this isn't a roundtable" floor.
+_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": "deepseek-chat"   # optional
+        }
+
+    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 []
+    model_name = str(payload.get("model") or "deepseek-chat")
+
+    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]:
+        # trust_env=False: same-process loopback to 127.0.0.1 — honoring
+        # HTTP_PROXY / HTTPS_PROXY env vars (commonly set on dev machines)
+        # causes httpx to route loopback calls through an unreachable upstream
+        # proxy and wait ~40 seconds for timeout before falling back.
+        async with httpx.AsyncClient(timeout=None, trust_env=False) as client:
+            try:
+                # One-shot recommendation: spin up a fresh thread per request.
+                # The recommender never needs to see prior turns.
+                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,
+                    # Recommender is pure judgment: no tools, no dispatching.
+                    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
+
+                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)
+
+                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",
+        },
+    )