速度优化

This commit is contained in:
董宏凯 authored and xuhd committed 2026-05-28 18:29:44 +08:00
1 parent a4b7ab5e9d
commit d018ecd805
6 files changed
+329 -97

No files matched your search

@@ -32,7 +32,10 @@ interface ArtifactsProviderProps {
}
export function ArtifactsProvider({ children }: ArtifactsProviderProps) {
const [artifacts, setArtifacts] = useState<string[]>([]);
const [artifacts, setArtifactsState] = useState<string[]>([]);
const setArtifacts = useCallback((next: string[]) => {
setArtifactsState(Array.isArray(next) ? next : []);
}, []);
const [selectedArtifact, setSelectedArtifact] = useState<string | null>(null);
const [autoSelect, setAutoSelect] = useState(true);
const [open, setOpen] = useState(
@@ -50,11 +50,12 @@ export function Step2SandboxLayout({
}: Step2SandboxLayoutProps) {
const layoutRef = useRef<GroupImperativeHandle>(null);
const {
artifacts,
artifacts: rawArtifacts,
open: artifactsOpen,
setOpen: setArtifactsOpen,
selectedArtifact,
} = useArtifacts();
const artifacts = rawArtifacts ?? [];
const artifactPanelOpen = artifactsOpen;
@@ -234,7 +234,7 @@ export function useStep2Orchestration(
} = useArtifacts();
const artifactPathsRef = useRef<string[]>([]);
useEffect(() => {
artifactPathsRef.current = artifacts;
artifactPathsRef.current = artifacts ?? [];
}, [artifacts]);
// 用户手动关掉沙箱后,autoOpen 会被 ArtifactsProvider 置 false —— 把它们打到 ref
// 上,让 openSeatVirtualArtifact 不用进 useCallback deps,避免每次 toggle 重建函数。
@@ -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:<gateway port>。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(
@@ -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] = []
@@ -137,7 +137,17 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
召回结果包在 ``<memory-context>`` 标签里并带系统提示,告诉模型这是
背景信息而非用户的新输入。
**跳过开关**:若 ``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()