From 8f9451a24bcfab389bc7386f9e9a0d7cd36bfd92 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Mon, 25 May 2026 14:14:31 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8E=A5=E5=8F=A3=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/roundtable-planning/api/recommend.ts | 223 +++++++ .../components/RecommendAgentsDialog.tsx | 558 ++++++++++++++++++ .../backend/app/gateway/app.py | 6 + .../backend/app/gateway/routers/recommend.py | 256 ++++++++ 4 files changed, 1043 insertions(+) create mode 100644 frontend-web/src/roundtable-planning/api/recommend.ts create mode 100644 frontend-web/src/roundtable-planning/components/RecommendAgentsDialog.tsx create mode 100644 offline-backend-20260512/backend/app/gateway/routers/recommend.py diff --git a/frontend-web/src/roundtable-planning/api/recommend.ts b/frontend-web/src/roundtable-planning/api/recommend.ts new file mode 100644 index 00000000..c8adca74 --- /dev/null +++ b/frontend-web/src/roundtable-planning/api/recommend.ts @@ -0,0 +1,223 @@ +import { apiFetch } from "@/core/api/fetch-client"; +import { getBackendBaseURL } from "@/core/config"; +import type { Agent } from "@/core/agents"; + +import type { IntentSummary } from "./intent"; + +/** + * Agent-recommendation API. Sits between Step 1 (intent clarification) and + * Step 2 (multi-agent roundtable): given the IntentSummary and a candidate + * agent pool, the built-in `roundtable-recommender` picks 2-6 agents most + * relevant to the task and explains why. + * + * One-shot, stateless: no /init, no thread_id, each call spins up a fresh + * backend-managed thread. Use {@link filterRecommendCandidates} to clean the + * `/api/agents` list before passing it as `candidates`. + */ + +/** Single agent pick returned by the recommender. */ +export interface RecommendPick { + agent_id: string; + reason: string; +} + +/** Final status frame the backend appends after the upstream stream finishes. */ +export interface StreamRecommendResult { + /** + * `done` → picks parsed and validated, ready to populate the dialog as + * default-checked selections. + * `asking` → recommender failed to emit a usable [RECOMMEND_READY] block, + * caller should apply its fallback list (typically the 6 roundtable-* seats). + * `error` → upstream / network failure. + */ + status: "done" | "asking" | "error"; + /** Full final AI text (includes the natural-language rationale). */ + content: string; + /** Populated only when status === "done"; null otherwise. */ + picks: RecommendPick[] | null; +} + +/** Phase mirror of streamIntent so the UI can reuse one loading-label mapper. */ +export type RecommendStreamPhase = "connecting" | "connected" | "streaming"; + +export interface StreamRecommendRequest { + /** Step 1 output. `objective` is required by the backend. */ + intent: IntentSummary; + /** Pre-filtered candidate pool (use {@link filterRecommendCandidates}). */ + candidates: RecommendCandidate[]; + model?: string; + signal?: AbortSignal; + /** Fires for every AI text chunk from upstream LangGraph messages-tuple frames. */ + onTextDelta?: (text: string, messageId: string) => void; + /** Fires whenever the inferred SSE phase changes. */ + onPhaseChange?: (phase: RecommendStreamPhase) => void; +} + +/** Shape the backend expects in `candidates[]`. */ +export interface RecommendCandidate { + agent_id: string; + name: string; + description: string; +} + +/** + * Agent ids that must never enter the candidate pool. Centralized here so + * every caller (recommend dialog, future tools) applies the same rule. + * + * - `roundtable-intent` — Step 1 clarification specialist, no role in research. + * - `roundtable-recommender`— the recommender itself, can't recommend itself. + * - `roundtable-coordinator-*` — disposable Step 2 coordinator shells (DB + * accumulates one per session); meaningless as roundtable participants. + */ +const EXCLUDED_AGENT_ID_PREFIX = "roundtable-coordinator-"; +const EXCLUDED_AGENT_IDS = new Set(["roundtable-intent", "roundtable-recommender"]); + +/** + * Drop excluded ids from a raw `GET /api/agents` list and project each entry + * to the slim shape the backend expects. Returns a fresh array; original + * objects are not mutated. + */ +export function filterRecommendCandidates(agents: Agent[]): RecommendCandidate[] { + const out: RecommendCandidate[] = []; + for (const a of agents) { + if (!a || !a.id) continue; + if (EXCLUDED_AGENT_IDS.has(a.id)) continue; + if (a.id.startsWith(EXCLUDED_AGENT_ID_PREFIX)) continue; + out.push({ + agent_id: a.id, + name: a.name ?? a.id, + description: a.description ?? "", + }); + } + return out; +} + +function isFinalRecommendFrame(value: unknown): value is { + status: "done" | "asking" | "error"; + content?: string; + picks?: RecommendPick[] | null; +} { + if (!value || typeof value !== "object") return false; + if (!("status" in value) || !("content" in value)) return false; + const status = (value as { status: unknown }).status; + return status === "done" || status === "asking" || status === "error"; +} + +/** + * Extract an AI text delta from one LangGraph `messages-tuple` frame. Same + * shape as the helper in intent.ts; duplicated rather than imported to keep + * each helper independently readable. + */ +function extractAiTextDelta(parsed: unknown): { text: string; id: string } | null { + if (!Array.isArray(parsed) || parsed.length < 1) return null; + const chunk = parsed[0]; + if (!chunk || typeof chunk !== "object") return null; + const obj = chunk as Record; + const type = String(obj.type ?? "").toLowerCase(); + if (type !== "ai" && type !== "aimessagechunk" && type !== "aimessage") return null; + const content = obj.content; + let text = ""; + if (typeof content === "string") { + text = content; + } else if (Array.isArray(content)) { + for (const item of content) { + if (typeof item === "string") text += item; + else if (item && typeof item === "object") { + const block = item as Record; + const blockText = block.text; + if (typeof blockText === "string") text += blockText; + } + } + } + if (!text) return null; + const id = typeof obj.id === "string" ? obj.id : ""; + return { text, id }; +} + +/** + * Open an SSE connection to `/api/recommend/stream` and resolve once the + * upstream stream finishes. Each call is one recommendation round — the + * recommender is stateless, so calling twice in a row produces two + * independent runs. + */ +export async function streamRecommend(req: StreamRecommendRequest): Promise { + const res = await apiFetch(`${getBackendBaseURL()}/api/recommend/stream`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + intent: req.intent, + candidates: req.candidates, + model: req.model, + }), + signal: req.signal, + }); + + if (!res.ok || !res.body) { + const text = await res.text().catch(() => ""); + throw new Error(`recommend/stream failed (${res.status}): ${text}`); + } + + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let lastEvent = ""; + + let currentPhase: RecommendStreamPhase = "connecting"; + req.onPhaseChange?.("connecting"); + const setPhase = (next: RecommendStreamPhase) => { + if (next === currentPhase) return; + currentPhase = next; + req.onPhaseChange?.(next); + }; + + const result: StreamRecommendResult = { status: "asking", content: "", picks: null }; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split("\n"); + buffer = lines.pop() ?? ""; + + for (const rawLine of lines) { + const line = rawLine.trimEnd(); + if (!line) continue; + if (line.startsWith("event:")) { + lastEvent = line.slice(6).trim(); + if (lastEvent === "metadata" && currentPhase === "connecting") { + setPhase("connected"); + } + continue; + } + if (!line.startsWith("data:")) continue; + const payload = line.slice(5).trim(); + if (!payload) continue; + + let parsed: unknown; + try { + parsed = JSON.parse(payload); + } catch { + continue; + } + + if (isFinalRecommendFrame(parsed)) { + result.status = parsed.status; + if (typeof parsed.content === "string") result.content = parsed.content; + if (Array.isArray(parsed.picks)) { + result.picks = parsed.picks as RecommendPick[]; + } + continue; + } + + if (lastEvent === "messages") { + const delta = extractAiTextDelta(parsed); + if (delta) { + setPhase("streaming"); + req.onTextDelta?.(delta.text, delta.id); + } + } + } + } + + return result; +} diff --git a/frontend-web/src/roundtable-planning/components/RecommendAgentsDialog.tsx b/frontend-web/src/roundtable-planning/components/RecommendAgentsDialog.tsx new file mode 100644 index 00000000..c2b77ed7 --- /dev/null +++ b/frontend-web/src/roundtable-planning/components/RecommendAgentsDialog.tsx @@ -0,0 +1,558 @@ +"use client"; + +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; +import { CheckIcon, Loader2Icon, RefreshCwIcon, SparklesIcon } from "lucide-react"; + +import { + Dialog, + DialogContent, + DialogDescription, + DialogFooter, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog"; +import { Button } from "@/components/ui/button"; +import { listAgents, type Agent } from "@/core/agents"; +import { cn } from "@/lib/utils"; + +import type { IntentSummary } from "../api/intent"; +import { + filterRecommendCandidates, + streamRecommend, + type RecommendCandidate, + type RecommendPick, + type RecommendStreamPhase, +} from "../api/recommend"; + +/** + * Picks-selection dialog that sits between Step 1 (intent) and Step 2 + * (roundtable). Opens with a streaming recommendation from + * `roundtable-recommender`, defaults its picks to checked, and lets the user + * trim or extend the selection from the rest of the agent pool before + * confirming. + * + * Selection bounds (2 ≤ N ≤ 8) are enforced on the confirm button — the + * backend also clamps but UI feedback is faster here. + */ + +export interface RecommendAgentsDialogProps { + open: boolean; + intent: IntentSummary | null; + /** Falls back to the recommender's own default if omitted. */ + model?: string; + onCancel: () => void; + /** Called with the user's final selection of agent ids (in display order). */ + onConfirm: (agentIds: string[]) => void; +} + +const MIN_SELECTED = 2; +const MAX_SELECTED = 8; + +/** + * Default selection when the recommender fails or returns < 2 valid picks. + * Intersected with the live candidate pool at runtime so missing built-ins + * don't break the dialog. + */ +const FALLBACK_AGENT_IDS = [ + "roundtable-intelligence", + "roundtable-environment", + "roundtable-solution-design", + "roundtable-risk-review", + "roundtable-execution-plan", + "roundtable-summary", +]; + +type DialogPhase = + | "idle" + | "loading_agents" + | "recommending" + | "ready" + | "fallback" + | "error"; + +interface InternalState { + phase: DialogPhase; + /** Streaming + final rationale text from the recommender. */ + rationale: string; + /** Streaming SSE phase from the recommender helper. */ + streamPhase: RecommendStreamPhase | null; + /** Pre-filtered candidate pool fetched from `/api/agents`. */ + candidates: RecommendCandidate[]; + /** Reason-by-agent_id, only for picks returned by the recommender. */ + reasons: Record; + /** Agent ids the recommender returned (in order). */ + recommendedIds: string[]; + /** User-visible warning ("推荐失败,已使用默认席位") for the fallback case. */ + warning: string | null; + /** Latest error message; only shown when `phase === "error"`. */ + errorMessage: string | null; +} + +const INITIAL_STATE: InternalState = { + phase: "idle", + rationale: "", + streamPhase: null, + candidates: [], + reasons: {}, + recommendedIds: [], + warning: null, + errorMessage: null, +}; + +/** + * Resolve which candidate ids should be checked by default given a set of + * recommended ids. The result preserves recommendation order and silently + * drops ids that aren't in the candidate pool. + */ +function intersectInOrder(preferred: string[], pool: RecommendCandidate[]): string[] { + const poolIds = new Set(pool.map((c) => c.agent_id)); + const seen = new Set(); + const out: string[] = []; + for (const id of preferred) { + if (!poolIds.has(id) || seen.has(id)) continue; + out.push(id); + seen.add(id); + } + return out; +} + +export function RecommendAgentsDialog({ + open, + intent, + model, + onCancel, + onConfirm, +}: RecommendAgentsDialogProps) { + const [state, setState] = useState(INITIAL_STATE); + const [selectedIds, setSelectedIds] = useState>(new Set()); + + // Latest stream abort controller, so closing or re-running cancels the + // in-flight request instead of leaking it. + const abortRef = useRef(null); + // Generation token guards against stale completion handlers writing into + // state after a re-run (e.g. user clicks "重新推荐" mid-stream). + const generationRef = useRef(0); + + const cancelInflight = useCallback(() => { + if (abortRef.current) { + abortRef.current.abort(); + abortRef.current = null; + } + }, []); + + /** + * The full pipeline: fetch agents → run recommendation → seed selection. + * Wrapped so the user can re-trigger it via the "重新推荐" button. + */ + const runRecommendation = useCallback(async () => { + if (!intent || !intent.objective) { + setState({ + ...INITIAL_STATE, + phase: "error", + errorMessage: "缺少任务目标,无法启动推荐", + }); + return; + } + + cancelInflight(); + const myGen = ++generationRef.current; + const controller = new AbortController(); + abortRef.current = controller; + + setState({ ...INITIAL_STATE, phase: "loading_agents" }); + setSelectedIds(new Set()); + + let agents: Agent[]; + try { + agents = await listAgents(); + } catch (err) { + if (myGen !== generationRef.current) return; + setState({ + ...INITIAL_STATE, + phase: "error", + errorMessage: `加载智能体列表失败:${err instanceof Error ? err.message : String(err)}`, + }); + return; + } + if (myGen !== generationRef.current) return; + + const candidates = filterRecommendCandidates(agents); + if (candidates.length === 0) { + setState({ + ...INITIAL_STATE, + phase: "error", + errorMessage: "候选智能体池为空,请先在系统里创建至少 2 个智能体", + }); + return; + } + + setState((prev) => ({ + ...prev, + phase: "recommending", + candidates, + rationale: "", + streamPhase: "connecting", + })); + + try { + const result = await streamRecommend({ + intent, + candidates, + model, + signal: controller.signal, + // Append every AI text delta to the rationale display. The + // recommender's [RECOMMEND_READY] + json block ends up in here too; + // that's fine — it's terse and rendering it shows users the model + // actually emitted the structured handoff. + onTextDelta: (text) => { + if (myGen !== generationRef.current) return; + setState((prev) => ({ ...prev, rationale: prev.rationale + text })); + }, + onPhaseChange: (phase) => { + if (myGen !== generationRef.current) return; + setState((prev) => ({ ...prev, streamPhase: phase })); + }, + }); + if (myGen !== generationRef.current) return; + + const reasons: Record = {}; + const recommendedIds: string[] = []; + if (result.picks) { + for (const p of result.picks) { + reasons[p.agent_id] = p.reason; + recommendedIds.push(p.agent_id); + } + } + + if (result.status === "done" && recommendedIds.length >= MIN_SELECTED) { + const seed = intersectInOrder(recommendedIds, candidates); + setSelectedIds(new Set(seed)); + setState((prev) => ({ + ...prev, + phase: "ready", + rationale: result.content || prev.rationale, + recommendedIds, + reasons, + streamPhase: null, + warning: null, + })); + return; + } + + // Fallback: status === "asking" or picks too few. Use the 6 built-in + // roundtable seats, intersected against the live pool. + const fallback = intersectInOrder(FALLBACK_AGENT_IDS, candidates); + const seed = fallback.length >= MIN_SELECTED + ? fallback + : candidates.slice(0, MIN_SELECTED).map((c) => c.agent_id); + setSelectedIds(new Set(seed)); + setState((prev) => ({ + ...prev, + phase: "fallback", + rationale: result.content || prev.rationale, + recommendedIds: seed, + reasons, + streamPhase: null, + warning: "智能推荐未生成有效结果,已为你勾选默认的圆桌席位,可手动调整。", + })); + } catch (err) { + if (myGen !== generationRef.current) return; + // AbortError fires when the user closes the dialog mid-stream — no + // state update needed because the dialog is going away. + if (err instanceof DOMException && err.name === "AbortError") return; + setState((prev) => ({ + ...prev, + phase: "error", + streamPhase: null, + errorMessage: `推荐过程出错:${err instanceof Error ? err.message : String(err)}`, + })); + } + }, [intent, model, cancelInflight]); + + // Trigger the pipeline whenever the dialog opens. Closing it aborts any + // in-flight stream so the next open starts clean. + useEffect(() => { + if (!open) { + cancelInflight(); + return; + } + void runRecommendation(); + return () => { + cancelInflight(); + }; + }, [open, runRecommendation, cancelInflight]); + + const handleToggle = useCallback((agentId: string) => { + setSelectedIds((prev) => { + const next = new Set(prev); + if (next.has(agentId)) next.delete(agentId); + else next.add(agentId); + return next; + }); + }, []); + + // Split candidates into "recommended" and "other" — order within each + // group matches the recommender's pick order / the API's list order so the + // dialog is deterministic between renders. + const { recommendedCards, otherCards } = useMemo(() => { + const recSet = new Set(state.recommendedIds); + const byId = new Map(state.candidates.map((c) => [c.agent_id, c])); + const recommended: RecommendCandidate[] = []; + for (const id of state.recommendedIds) { + const c = byId.get(id); + if (c) recommended.push(c); + } + const others = state.candidates.filter((c) => !recSet.has(c.agent_id)); + return { recommendedCards: recommended, otherCards: others }; + }, [state.candidates, state.recommendedIds]); + + const selectedCount = selectedIds.size; + const canConfirm = + (state.phase === "ready" || state.phase === "fallback") && + selectedCount >= MIN_SELECTED && + selectedCount <= MAX_SELECTED; + + const selectionHint = (() => { + if (selectedCount < MIN_SELECTED) { + return `至少再选 ${MIN_SELECTED - selectedCount} 个`; + } + if (selectedCount > MAX_SELECTED) { + return `已超出上限 ${selectedCount - MAX_SELECTED} 个`; + } + return `已选 ${selectedCount} / ${MAX_SELECTED}`; + })(); + + const handleConfirm = useCallback(() => { + if (!canConfirm) return; + // Order: recommended-and-selected first (in recommender order), then + // user-added picks (in pool order). Keeps the leader's "可调度智能体" + // list visually stable. + const ordered: string[] = []; + for (const c of recommendedCards) { + if (selectedIds.has(c.agent_id)) ordered.push(c.agent_id); + } + for (const c of otherCards) { + if (selectedIds.has(c.agent_id)) ordered.push(c.agent_id); + } + onConfirm(ordered); + }, [canConfirm, recommendedCards, otherCards, selectedIds, onConfirm]); + + const handleOpenChange = useCallback( + (next: boolean) => { + if (!next) onCancel(); + }, + [onCancel], + ); + + return ( + + + + + + 选择参会智能体 + + + 根据你的任务目标,系统为你推荐了最合适的智能体。你可以取消勾选,也可以从下方添加其他智能体。 + + + + {/* Rationale + status block */} +
+ +
+ + {/* Selection area: scrollable, takes remaining height */} +
+ {state.phase === "error" ? ( +
+ {state.errorMessage ?? "未知错误"} +
+ ) : (state.phase === "ready" || state.phase === "fallback") ? ( + <> + {state.warning && ( +
+ {state.warning} +
+ )} + + {recommendedCards.length > 0 && ( +
+ {recommendedCards.map((c) => ( + handleToggle(c.agent_id)} + /> + ))} +
+ )} + + {otherCards.length > 0 && ( +
+ {otherCards.map((c) => ( + handleToggle(c.agent_id)} + /> + ))} +
+ )} + + ) : null} +
+ + +
+ MAX_SELECTED + ? "text-destructive" + : "text-muted-foreground", + )} + > + {selectionHint} + + {(state.phase === "ready" || state.phase === "fallback" || state.phase === "error") && ( + + )} +
+
+ + +
+ + + + ); +} + +function RationaleBlock({ state }: { state: InternalState }) { + if (state.phase === "loading_agents") { + return ( +
+ + 正在加载候选智能体... +
+ ); + } + if (state.phase === "recommending") { + const label = + state.streamPhase === "connecting" + ? "调度推荐智能体..." + : state.streamPhase === "connected" + ? "已连接,等待响应..." + : "正在生成推荐..."; + return ( +
+
+ + {label} +
+ {state.rationale && ( + 
+            {state.rationale}
+
+ )} +
+ ); + } + if (state.phase === "ready" || state.phase === "fallback") { + if (!state.rationale) { + return
推荐完成。
; + } + return ( + 
+        {state.rationale}
+
+ ); + } + if (state.phase === "error") { + return
推荐失败,请使用下方的"重新推荐"重试,或取消后再次进入。
; + } + return null; +} + +function Section({ title, children }: { title: string; children: React.ReactNode }) { + return ( +
+
+ {title} +
+
{children}
+
+ ); +} + +interface AgentCardProps { + candidate: RecommendCandidate; + reason: string | undefined; + recommended: boolean; + checked: boolean; + onToggle: () => void; +} + +function AgentCard({ candidate, reason, recommended, checked, onToggle }: AgentCardProps) { + return ( + + ); +} diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index 9678a42d..72987927 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/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. diff --git a/offline-backend-20260512/backend/app/gateway/routers/recommend.py b/offline-backend-20260512/backend/app/gateway/routers/recommend.py new file mode 100644 index 00000000..83048db5 --- /dev/null +++ b/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: "- (): " — 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": "", + "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", + }, + )