diff --git a/frontend-web/src/components/workspace/artifacts/context.tsx b/frontend-web/src/components/workspace/artifacts/context.tsx index e15481af..f864d965 100644 --- a/frontend-web/src/components/workspace/artifacts/context.tsx +++ b/frontend-web/src/components/workspace/artifacts/context.tsx @@ -32,7 +32,10 @@ interface ArtifactsProviderProps { } export function ArtifactsProvider({ children }: ArtifactsProviderProps) { - const [artifacts, setArtifacts] = useState([]); + const [artifacts, setArtifactsState] = useState([]); + const setArtifacts = useCallback((next: string[]) => { + setArtifactsState(Array.isArray(next) ? next : []); + }, []); const [selectedArtifact, setSelectedArtifact] = useState(null); const [autoSelect, setAutoSelect] = useState(true); const [open, setOpen] = useState( diff --git a/frontend-web/src/roundtable-planning/components/Step2SandboxLayout.tsx b/frontend-web/src/roundtable-planning/components/Step2SandboxLayout.tsx index 5998cbf6..3b33d59e 100644 --- a/frontend-web/src/roundtable-planning/components/Step2SandboxLayout.tsx +++ b/frontend-web/src/roundtable-planning/components/Step2SandboxLayout.tsx @@ -50,11 +50,12 @@ export function Step2SandboxLayout({ }: Step2SandboxLayoutProps) { const layoutRef = useRef(null); const { - artifacts, + artifacts: rawArtifacts, open: artifactsOpen, setOpen: setArtifactsOpen, selectedArtifact, } = useArtifacts(); + const artifacts = rawArtifacts ?? []; const artifactPanelOpen = artifactsOpen; diff --git a/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts b/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts index 54c65b56..5d7c54ce 100644 --- a/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts +++ b/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts @@ -234,7 +234,7 @@ export function useStep2Orchestration( } = useArtifacts(); const artifactPathsRef = useRef([]); useEffect(() => { - artifactPathsRef.current = artifacts; + artifactPathsRef.current = artifacts ?? []; }, [artifacts]); // 用户手动关掉沙箱后,autoOpen 会被 ArtifactsProvider 置 false —— 把它们打到 ref // 上,让 openSeatVirtualArtifact 不用进 useCallback deps,避免每次 toggle 重建函数。 diff --git a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py index bcf6bb01..5f92a3e6 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py +++ b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py @@ -53,13 +53,17 @@ import os import re import time import traceback +import uuid from typing import Any, AsyncIterator import httpx from deerflow.config import get_app_config from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple +from deerflow.runtime.user_context import get_effective_user_id +from deerflow.utils.time import now_iso from fastapi import APIRouter, Body, HTTPException, Request from fastapi.responses import StreamingResponse +from langgraph.checkpoint.base import empty_checkpoint logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/multi-agent", tags=["multi-agent"]) @@ -323,6 +327,178 @@ async def _create_agent( return False, resp.text +# --------------------------------------------------------------------------- +# 同进程直调 helpers(仅供 init_multi_agent 用,不要扩散到其它地方) +# +# init 阶段串了 N+1 个 thread create + N 个 GET /api/agents + 1 个 POST /api/agents, +# 用 HTTP loopback 串走每次都得 TCP + 鉴权中间件 + Pydantic 反序列化一遍。每个 +# 调用 10-30ms,加起来对前端就是「点进 Step 2 后席位灯一颗一颗亮」的卡顿来源。 +# +# 这里把这三个调用改成直接吃 app.state 里的 store/checkpointer —— 跑在同一个事 +# 件循环里,省掉 TCP + 序列化 + 中间件链。鉴权 / 校验 / 文件副作用按需在直调 +# helper 里显式做掉,不要"绕过"任何一项。 +# +# 设计边界:这里只覆盖 init 阶段。SSE 长连接(/runs/stream / state)留在 HTTP +# loopback,因为它们本就是别的进程模型(LangGraph 子树),抽出来风险大收益小。 +# --------------------------------------------------------------------------- + + +def _resolve_user_id(request: Request) -> str: + """与 routers/agents._current_user_id 完全等价,直调路径里用。""" + user = getattr(request.state, "user", None) + if user is not None: + return str(user.id) + return get_effective_user_id() + + +async def _create_thread_direct(request: Request) -> str: + """同进程版的 ``POST /api/threads``。 + + 完全镜像 ``routers/threads.create_thread`` 的写盘动作(thread_meta + + 空 checkpoint),但跳过 TCP + Pydantic 反序列化。仅供 init 阶段批量建席位 + 线程用,失败时抛 HTTPException 由 SSE 帧上报。 + """ + from app.gateway.deps import get_checkpointer, get_thread_store + + checkpointer = get_checkpointer(request) + thread_store = get_thread_store(request) + thread_id = str(uuid.uuid4()) + now = now_iso() + + try: + await thread_store.create(thread_id, assistant_id=None, metadata={}) + except Exception: + logger.exception("[init-direct] thread_meta create failed for %s", thread_id) + raise HTTPException(status_code=500, detail="Failed to create thread") + + config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} + ckpt_metadata = { + "step": -1, + "source": "input", + "writes": None, + "parents": {}, + "created_at": now, + } + try: + await checkpointer.aput(config, empty_checkpoint(), ckpt_metadata, {}) + except Exception: + logger.exception("[init-direct] checkpoint create failed for %s", thread_id) + raise HTTPException(status_code=500, detail="Failed to create thread") + return thread_id + + +async def _get_agents_batch_direct( + request: Request, + agent_ids: list[str], +) -> dict[str, dict[str, Any]]: + """同进程版的 N 个并行 ``GET /api/agents/{id}`` 合并为 1 次 DB 查询。 + + 返回 ``{agent_id: {id, name, description}}``。**严格模式**:任一席位对当前 + 用户不可见就抛 404 —— 与原 HTTP loopback 的 ``_get_agent`` 行为一致(原路径 + 单查 404 会 raise,在外层被 except 捕获并 yield error 帧),不要悄悄回退。 + """ + from app.gateway.deps import get_agent_store + + store = get_agent_store(request) + user_id = _resolve_user_id(request) + records = await store.list_by_ids(agent_ids, user_id) + out: dict[str, dict[str, Any]] = {} + for r in records: + rid = str(r.get("id") or "").lower() + if not rid: + continue + out[rid] = { + "id": rid, + "name": str(r.get("name") or rid), + "description": str(r.get("description") or ""), + } + missing = [sid for sid in agent_ids if sid.lower() not in out] + if missing: + raise HTTPException( + status_code=404, + detail=f"Agent(s) not visible to current user: {', '.join(missing)}", + ) + return out + + +async def _create_coordinator_direct( + request: Request, + *, + agent_id: str, + name: str, + description: str, + soul: str, + model: str | None, + skills: list[str] | None, +) -> tuple[bool, dict[str, Any] | str]: + """同进程版的协调 agent 创建。 + + 完整复刻 ``routers/agents.create_agent_endpoint`` 的关键副作用: + - ``agents_api.enabled`` master switch + - 显示名校验(空白则 422 → 这里转 (False, msg)) + - 重复 id 检查(409 → (False, msg)) + - 写 ``config.yaml`` + ``SOUL.md``(用 ``asyncio.to_thread`` 防同步 IO 阻塞) + - ``agent_store.create`` 入表 + + 返回 ``(True, record)`` 或 ``(False, error_text)``;调用方决定要不要发 SSE error 帧。 + """ + from app.gateway.deps import get_agent_store + from app.gateway.routers.agents import ( + _require_agents_api_enabled, + _write_agent_config, + _write_agent_soul, + ) + from deerflow.config.paths import get_paths + + _require_agents_api_enabled() + normalized_name = (name or "").strip() + if not normalized_name: + return False, "Agent name cannot be empty" + + store = get_agent_store(request) + if await store.get_any(agent_id) is not None: + return False, f"Agent id '{agent_id}' already exists" + + agent_dir = get_paths().agent_dir(agent_id) + if agent_dir.exists(): + return False, f"Agent id '{agent_id}' already exists" + + try: + # config.yaml / SOUL.md 是同步 IO —— 全部丢线程池,不阻塞 init SSE 的发帧节奏。 + await asyncio.to_thread( + _write_agent_config, + agent_id, + name=normalized_name, + description=description or "", + skills=skills, + model=model, + tool_groups=None, + ) + await asyncio.to_thread(_write_agent_soul, agent_id, soul) + record = await store.create( + { + "id": agent_id, + "name": normalized_name, + "description": description or "", + "user_id": _resolve_user_id(request), + "published": False, + } + ) + return True, record + except Exception as exc: + # 与 routers/agents 失败回滚一致 —— 文件已写出但 DB 失败时清掉 dir,避免下次 + # `agent_dir.exists()` 误判为"已存在"导致后续 init 永远 409。 + try: + if agent_dir.exists(): + import shutil + + await asyncio.to_thread(shutil.rmtree, agent_dir) + except Exception: + logger.exception("[init-direct] cleanup after coordinator create failure") + logger.exception("[init-direct] coordinator create failed") + return False, str(exc) + + async def _append_thread_message( client: httpx.AsyncClient, base: str, @@ -456,6 +632,8 @@ def _run_payload( new_message: str, excluded_tools: list[str], skill_stop_names: list[str], + thinking_enabled: bool = True, + reasoning_effort: str = "medium", ) -> dict[str, Any]: """构造 ``POST /api/threads/{thread_id}/runs/stream`` 的请求体。 @@ -463,6 +641,10 @@ def _run_payload( - ``excluded_tools``:按路由裁剪工具面(intent 仅澄清、special 禁派活等)。 - ``skill_stop_names``:leader 侧配合 SkillStopMiddleware 在 ``agent_orchestration`` 调用后截断图,以便网关从 AIMessage.tool_calls 解析派活列表。 + - ``thinking_enabled`` / ``reasoning_effort``:默认开 CoT + medium,适合 Step 1 澄清 + 与子智能体交付这类需要推理的场景。**推荐弹窗与总控派活**是短决策任务,调用方 + 应传 ``thinking_enabled=False`` + ``reasoning_effort="low"``,省掉模型先吐 + reasoning tokens 才开口的时间,显著降低前端"调度中"首字节延迟。 """ return { "input": { @@ -479,11 +661,17 @@ def _run_payload( "agent_name": agent_name, "model_name": model_name, "mode": "pro", - "reasoning_effort": "medium", - "thinking_enabled": True, + "reasoning_effort": reasoning_effort, + "thinking_enabled": thinking_enabled, "is_plan_mode": False, "subagent_enabled": False, "thread_id": thread_id, + # 圆桌四个 router(intent / recommend / leader / special)都是临时任务, + # 用户个人记忆与本次研讨无关。MemoryMiddleware 的 Hindsight recall 是 + # 同步阻塞 HTTP 调用(单次 timeout 15s),如果记忆服务慢/不通会让前端 + # "调度中"卡很久;这里默认跳过,builtin USER.md/MEMORY.md 仍通过系统 + # 提示注入不受影响。 + "memory_recall_disabled": True, }, "stream_mode": ["messages-tuple", "values"], "stream_subgraphs": True, @@ -534,111 +722,130 @@ async def init_multi_agent(request: Request, payload: dict = Body(...)) -> Strea _validate_agent_id(name, "agent_names entry") _validate_agent_id(main_agent_name, "main_agent_name") - base = _loopback_base(request) - headers = _auth_headers(request) - def _frame(payload: dict[str, Any]) -> str: return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n" async def stream_generator() -> AsyncIterator[str]: - # TCP loopback 到 127.0.0.1:。client 配置见 _make_loopback_client: - # trust_env=False 防 HTTP_PROXY 拦截,timeout=None 容纳 SSE 长连接。 - async with _make_loopback_client(request) as client: - try: - # Per-seat prepare task — same body as before, but we no longer - # await them with asyncio.gather. Instead, asyncio.as_completed - # surfaces them in REAL completion order so we can emit a - # seat_ready frame the moment each finishes. - async def _prepare_sub_agent(sub_id: str) -> tuple[str, str, str, str]: - info = await _get_agent(client, base, headers, sub_id) - display_name = str(info.get("name") or sub_id) - description = info.get("description", "") or "" - thread_id = await _create_thread(client, base, headers) - return sub_id.lower(), thread_id, display_name, description + # init 阶段 fast path:全部走同进程直调(_create_thread_direct / + # _get_agents_batch_direct / _create_coordinator_direct),不再走 HTTP + # loopback —— TCP + 鉴权中间件 + Pydantic 反序列化每次 10-30ms,N+1 个席位 + # 累计就是"点进 Step 2 后席位一颗颗亮"的卡顿来源。 + try: + t_init_start = time.perf_counter() + # 1) 一次 DB 查询拿到所有席位的 name/description,替代之前的 N 个并行 + # GET /api/agents/{id}。SOUL 拼接需要描述,这里一次性拿齐。 + agent_meta = await _get_agents_batch_direct(request, agent_names) + t_meta_done = time.perf_counter() - tasks = [asyncio.create_task(_prepare_sub_agent(sid)) for sid in agent_names] - prepared: list[tuple[str, str, str, str]] = [] - try: - for fut in asyncio.as_completed(tasks): - sid, thread_id, display_name, description = await fut - prepared.append((sid, thread_id, display_name, description)) - yield _frame({ - "event": "seat_ready", - "agent_name": sid, - "thread_id": thread_id, - "display_name": display_name, - "description": description, - }) - except Exception: - # Cancel any still-pending sibling so we don't leak tasks - # past the AsyncClient close. - for t in tasks: - if not t.done(): - t.cancel() - raise - - agent_threads: dict[str, str] = {sid: thread_id for sid, thread_id, _, _ in prepared} - descriptions: list[str] = [ - f"- agent_name: {sid}({display_name}):{description}" - for sid, _, display_name, description in prepared - ] - - # 动态 SOUL:把本次 init 的席位列表写进总控 prompt,供 agent_orchestration 派活。 - soul = ( - "#核心定位:\n" - "用途:根据用户的任务目标,调用其他智能体干活,只负责协调汇总," - "当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果\n" - "#调度方法:\n" - "通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活。" - "注意:agent_name 必须严格使用下方列出的英文 id(例如 roundtable-intelligence," - "或形如 02714cbe6e64406a8a4659f0e3d9c0b8 这样的 UUID)," - "不要把括号里的中文显示名拼进 agent_name 里。\n" - "#关键约束(必读):\n" - "1. **不要对同一子智能体重复派同样的活**:每次派活前回顾对话历史," - "如果某个子智能体已经针对你正要派的任务完成过交付(" - "看到形如\"子智能体[名称]完成XX工作,交付内容为:...\"的 human 消息)," - "就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题。\n" - "2. **看到交付要会用**:子智能体的交付内容会作为 human 消息追加到你的对话历史里," - "你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具)," - "让讨论收敛。\n" - "3. **避免无意义循环**:如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论。\n" - "#可调度智能体:\n" - + "\n".join(descriptions) + # 2) 每个席位 thread 并行创建(直调 thread_store + checkpointer), + # 用 asyncio.as_completed 保持"先完先发 seat_ready"的体验。 + async def _prepare_sub_agent(sub_id: str) -> tuple[str, str, str, str]: + thread_id = await _create_thread_direct(request) + meta = agent_meta.get(sub_id.lower(), {}) + return ( + sub_id.lower(), + thread_id, + str(meta.get("name") or sub_id), + str(meta.get("description") or ""), ) - coord_thread = await _create_thread(client, base, headers) - agent_threads[main_agent_name] = coord_thread - ok, result = await _create_agent( - client, - base, - headers, + tasks = [asyncio.create_task(_prepare_sub_agent(sid)) for sid in agent_names] + prepared: list[tuple[str, str, str, str]] = [] + try: + for fut in asyncio.as_completed(tasks): + sid, thread_id, display_name, description = await fut + prepared.append((sid, thread_id, display_name, description)) + yield _frame({ + "event": "seat_ready", + "agent_name": sid, + "thread_id": thread_id, + "display_name": display_name, + "description": description, + }) + except Exception: + # Cancel any still-pending sibling so partially-prepared seats + # don't keep running past the stream close. + for t in tasks: + if not t.done(): + t.cancel() + raise + t_seats_done = time.perf_counter() + + agent_threads: dict[str, str] = {sid: thread_id for sid, thread_id, _, _ in prepared} + descriptions: list[str] = [ + f"- agent_name: {sid}({display_name}):{description}" + for sid, _, display_name, description in prepared + ] + + # 动态 SOUL:把本次 init 的席位列表写进总控 prompt,供 agent_orchestration 派活。 + soul = ( + "#核心定位:\n" + "用途:根据用户的任务目标,调用其他智能体干活,只负责协调汇总," + "当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果\n" + "#调度方法:\n" + "通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活。" + "注意:agent_name 必须严格使用下方列出的英文 id(例如 roundtable-intelligence," + "或形如 02714cbe6e64406a8a4659f0e3d9c0b8 这样的 UUID)," + "不要把括号里的中文显示名拼进 agent_name 里。\n" + "#关键约束(必读):\n" + "1. **不要对同一子智能体重复派同样的活**:每次派活前回顾对话历史," + "如果某个子智能体已经针对你正要派的任务完成过交付(" + "看到形如\"子智能体[名称]完成XX工作,交付内容为:...\"的 human 消息)," + "就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题。\n" + "2. **看到交付要会用**:子智能体的交付内容会作为 human 消息追加到你的对话历史里," + "你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具)," + "让讨论收敛。\n" + "3. **避免无意义循环**:如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论。\n" + "#可调度智能体:\n" + + "\n".join(descriptions) + ) + + # 3) coordinator thread + agent 并行创建。thread 与 agent 之间无依赖, + # asyncio.gather 比串行省一个直调往返的开销。 + coord_thread_task = asyncio.create_task(_create_thread_direct(request)) + coord_agent_task = asyncio.create_task( + _create_coordinator_direct( + request, + agent_id=main_agent_name, name=main_agent_name, description="你是总控控制协调智能体,负责协调其他智能体干活", + soul=soul, model=model, skills=["agent_orchestration"], - soul=soul, - agent_id=main_agent_name, ) - if not ok and isinstance(result, str) and "exists" not in result: - yield _frame({"event": "error", "detail": f"create coordinator agent failed: {result}"}) - return + ) + coord_thread = await coord_thread_task + agent_threads[main_agent_name] = coord_thread + ok, result = await coord_agent_task + if not ok and isinstance(result, str) and "exists" not in result: + yield _frame({"event": "error", "detail": f"create coordinator agent failed: {result}"}) + return + t_coord_done = time.perf_counter() + logger.info( + "[init-direct] meta=%.3fs seats=%.3fs coord=%.3fs total=%.3fs seats_count=%d", + t_meta_done - t_init_start, + t_seats_done - t_meta_done, + t_coord_done - t_seats_done, + t_coord_done - t_init_start, + len(prepared), + ) - yield _frame({ - "event": "coordinator_ready", - "main_agent_name": main_agent_name, - "thread_id": coord_thread, - }) - yield _frame({ - "event": "init_done", - "main_agent_name": main_agent_name, - "thread_ids": agent_threads, - }) - except HTTPException as exc: - logger.warning("multi-agent init HTTPException: %s %s", exc.status_code, exc.detail) - yield _frame({"event": "error", "detail": f"[{exc.status_code}] {exc.detail}"}) - except Exception as exc: - logger.exception("multi-agent init unexpected failure") - yield _frame({"event": "error", "detail": f"unexpected: {exc!r}"}) + yield _frame({ + "event": "coordinator_ready", + "main_agent_name": main_agent_name, + "thread_id": coord_thread, + }) + yield _frame({ + "event": "init_done", + "main_agent_name": main_agent_name, + "thread_ids": agent_threads, + }) + except HTTPException as exc: + logger.warning("multi-agent init HTTPException: %s %s", exc.status_code, exc.detail) + yield _frame({"event": "error", "detail": f"[{exc.status_code}] {exc.detail}"}) + except Exception as exc: + logger.exception("multi-agent init unexpected failure") + yield _frame({"event": "error", "detail": f"unexpected: {exc!r}"}) return StreamingResponse( stream_generator(), @@ -825,6 +1032,11 @@ async def _leader_run( new_message=new_message, excluded_tools=["web_search"], skill_stop_names=effective_stop_names, + # 总控派活是"下一步派谁、什么任务"的短决策,不需要 CoT。 + # 关掉 thinking + low reasoning 让"调度中"几乎瞬间转入"正在派活", + # 子智能体仍保留 thinking(在 _special_run 里默认 True),保证交付质量。 + thinking_enabled=False, + reasoning_effort="low", ) t_payload_built = time.perf_counter() logger.info( diff --git a/offline-backend-20260512/backend/app/gateway/routers/recommend.py b/offline-backend-20260512/backend/app/gateway/routers/recommend.py index 6ff43224..f6b50494 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/recommend.py +++ b/offline-backend-20260512/backend/app/gateway/routers/recommend.py @@ -253,6 +253,12 @@ async def stream_recommend(request: Request, payload: dict = Body(...)) -> Strea "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] = [] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py index 6a067ef0..b48c06be 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py @@ -137,7 +137,17 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]): 召回结果包在 ```` 标签里并带系统提示,告诉模型这是 背景信息而非用户的新输入。 + + **跳过开关**:若 ``runtime.context.memory_recall_disabled`` 为真,直接 + 返回 None。圆桌规划等临时性任务(intent / recommend / leader / special) + 不需要从用户个人记忆里召回,Hindsight HTTP 同步阻塞会显著增加 LLM 首 + 字节延迟(前端"调度中"卡顿);由路由层通过 ``_run_payload`` 的 context + 字段统一关闭。 """ + runtime = getattr(request, "runtime", None) + if runtime is not None and getattr(runtime, "context", None): + if runtime.context.get("memory_recall_disabled"): + return None if not self._injection_enabled: return None config = self._memory_config or get_memory_config()