Merge branch 'master' of http://192.168.0.200/knowledge-base-bj/deerflow-server into ji
# Conflicts: # frontend-web/src/components/workspace/chats/chat-box.tsx # offline-backend-20260512/backend/app/gateway/routers/agents.py
This commit is contained in:
commit
6792b4f4c4
200 files changed
+78520
-392
No files matched your search
@@ -503,6 +503,24 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
|
||||
except Exception:
|
||||
logger.exception("Skill ownership migration failed (non-fatal)")
|
||||
|
||||
# 一次性把 .deer-flow/agents/{id}/ 目录里历史遗留的 agent 导入 agents 表。
|
||||
# 路由层原本每次请求都跑一遍这个 sync(扫文件系统 + N 个 agent 各一次 DB
|
||||
# SELECT/UPDATE),把它挪到启动期 + 路由层只保留 TTL 兜底,显著降低
|
||||
# /api/agents 列表延迟。
|
||||
try:
|
||||
from app.gateway.routers.agents import (
|
||||
_mark_legacy_sync_fresh,
|
||||
_sync_legacy_agents,
|
||||
)
|
||||
|
||||
agent_store = getattr(app.state, "agent_store", None)
|
||||
if agent_store is not None:
|
||||
await _sync_legacy_agents(agent_store)
|
||||
_mark_legacy_sync_fresh()
|
||||
logger.info("Legacy folder-backed agents synced at startup")
|
||||
except Exception:
|
||||
logger.exception("Legacy agent sync failed at startup (non-fatal)")
|
||||
|
||||
# 启动 skill curator 调度器:常驻后台任务,每小时检查一次需不需要进化
|
||||
# (per-user interval_hours 仍是真正决定要不要跑进化的闸)。
|
||||
try:
|
||||
|
||||
@@ -406,7 +406,7 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
|
||||
|
||||
> 以下两个示例覆盖**不同任务类型**,目的是教你「思路要呼应任务本身」,**不要**把任何一个示例的具体职能链当模板。
|
||||
|
||||
### 示例 A:合规改造任务(领域候选池)
|
||||
### 示例 A:合规改造任务(职能型候选池)
|
||||
|
||||
**用户输入**:
|
||||
```
|
||||
@@ -415,11 +415,12 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
|
||||
关键假设:现有系统具备基础数字化能力;允许预定窗口停机
|
||||
|
||||
候选智能体池:
|
||||
- compliance-research (合规调研):检索同类历史案例、行业最佳实践与监管口径
|
||||
- infra-capacity (基础设施评估):评估硬件容量与系统量化承载力
|
||||
- solution-architect (方案架构):产出主备方案与关键路径
|
||||
- risk-auditor (风险审查):合规红线检查与收敛建议
|
||||
- execution-planner (执行规划):里程碑与资源排布
|
||||
- roundtable-intelligence (情报收集):检索同类历史案例、行业最佳实践与外部情报
|
||||
- roundtable-environment (环境评估):评估硬件容量与系统量化承载力
|
||||
- roundtable-solution-design (方案设计):产出主备方案与关键路径
|
||||
- roundtable-risk-review (风险审查):合规红线检查与收敛建议
|
||||
- roundtable-execution-plan (执行规划):里程碑与资源排布
|
||||
- roundtable-summary (总结陈述):归纳冲突标记与盖印
|
||||
- my-cost-calc (成本测算):基于预算和资源单价计算总成本
|
||||
```
|
||||
|
||||
@@ -430,10 +431,10 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
|
||||
```json
|
||||
{
|
||||
"picks": [
|
||||
{"agent_id": "compliance-research", "reason": "检索同类等保 2.0 改造案例与监管口径,直接支撑本次合规对标"},
|
||||
{"agent_id": "solution-architect", "reason": "在 6 个月工期约束下产出主备改造方案与关键路径"},
|
||||
{"agent_id": "risk-auditor", "reason": "对照等保 2.0 合规红线审查方案,避免后期返工"},
|
||||
{"agent_id": "execution-planner", "reason": "把改造拆成可在 6 个月内交付的里程碑与资源排布"},
|
||||
{"agent_id": "roundtable-intelligence", "reason": "检索同类等保 2.0 改造案例与监管口径,直接支撑本次合规对标"},
|
||||
{"agent_id": "roundtable-solution-design", "reason": "在 6 个月工期约束下产出主备改造方案与关键路径"},
|
||||
{"agent_id": "roundtable-risk-review", "reason": "对照等保 2.0 合规红线审查方案,避免后期返工"},
|
||||
{"agent_id": "roundtable-execution-plan", "reason": "把改造拆成可在 6 个月内交付的里程碑与资源排布"},
|
||||
{"agent_id": "my-cost-calc", "reason": "校验整体方案落在 500 万元预算硬约束内"}
|
||||
]
|
||||
}
|
||||
@@ -452,7 +453,7 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
|
||||
- channel-planner (渠道规划师):按预算和受众设计跨平台投放组合
|
||||
- audience-insight (受众洞察):画像、关键词、情绪点研究
|
||||
- creative-copy (创意文案):产出落地页与社媒文案变体
|
||||
- ad-compliance (广告合规):审查文案与素材合规风险
|
||||
- roundtable-risk-review (风险审查):合规红线检查与收敛建议
|
||||
```
|
||||
|
||||
**你的输出**:
|
||||
@@ -470,8 +471,8 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
|
||||
}
|
||||
```
|
||||
|
||||
> **关键观察**:两个示例都从候选池里按任务语义挑选,**不要**套用固定职能链(如"调研→设计→风险→执行")。
|
||||
> 候选池里的智能体能不能匹配任务,只看 description 和任务的实际语义。如果候选池里有"广告设计智能体"且任务是广告设计,**你必须挑它**,而不是机械复制示例 A 的链路。
|
||||
> **关键观察**:示例 A 选了 `roundtable-intelligence / solution-design / execution-plan` 等"职能型"智能体,示例 B **完全没用任何 `roundtable-*` 内置**,而是从用户自建的领域智能体里挑。
|
||||
> 这是因为:**候选池里的智能体能不能匹配任务,只看 description 和任务的实际语义,不看 id 是不是 `roundtable-` 开头**。如果候选池里有"广告设计智能体"且任务是广告设计,**你必须挑它**,而不是套用示例 A 的职能链。
|
||||
"""
|
||||
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
import logging
|
||||
import shutil
|
||||
import time
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
@@ -207,6 +208,11 @@ async def _sync_legacy_agents(store: AgentStore) -> None:
|
||||
|
||||
This preserves old deployments where agents already existed under
|
||||
``.deer-flow/agents/{name}`` before the agents table was introduced.
|
||||
|
||||
Heavy: scans the agents directory + parses each ``config.yaml`` + issues a
|
||||
SELECT-then-INSERT/UPDATE per agent. Call once at startup via the lifespan,
|
||||
and rely on :func:`_maybe_sync_legacy_agents` (TTL-gated) inside request
|
||||
handlers as a safety net for environments that bypassed the lifespan hook.
|
||||
"""
|
||||
try:
|
||||
for agent_cfg in list_custom_agents():
|
||||
@@ -225,6 +231,44 @@ async def _sync_legacy_agents(store: AgentStore) -> None:
|
||||
logger.exception("Failed to sync legacy folder-backed agents")
|
||||
|
||||
|
||||
# Throttle the per-request legacy-sync fallback. The real sync runs once at
|
||||
# startup (lifespan); request handlers used to call ``_sync_legacy_agents`` on
|
||||
# every hit, which scanned the filesystem + did N DB upserts per agent each
|
||||
# time. After the first successful sync we skip the work for ``_LEGACY_SYNC_TTL_SECONDS``
|
||||
# — long enough to be effectively "startup-only" in steady state, short enough
|
||||
# that a freshly dropped agent directory gets picked up on the next bucket.
|
||||
_LEGACY_SYNC_TTL_SECONDS = 3600.0
|
||||
_legacy_sync_state: dict[str, Any] = {"last_at": 0.0, "lock": None}
|
||||
|
||||
|
||||
async def _maybe_sync_legacy_agents(store: AgentStore) -> None:
|
||||
"""TTL-gated wrapper around :func:`_sync_legacy_agents`.
|
||||
|
||||
Becomes a no-op in steady state — the heavy lifting happens in the lifespan
|
||||
hook. Inside the TTL window we don't even take the lock; outside it we
|
||||
serialize callers so a thundering herd doesn't fan out into N filesystem
|
||||
scans + N*M DB queries.
|
||||
"""
|
||||
now = time.monotonic()
|
||||
if now - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS:
|
||||
return
|
||||
lock = _legacy_sync_state["lock"]
|
||||
if lock is None:
|
||||
lock = asyncio.Lock()
|
||||
_legacy_sync_state["lock"] = lock
|
||||
async with lock:
|
||||
# Re-check under the lock — a sibling coroutine may have just refreshed.
|
||||
if time.monotonic() - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS:
|
||||
return
|
||||
await _sync_legacy_agents(store)
|
||||
_legacy_sync_state["last_at"] = time.monotonic()
|
||||
|
||||
|
||||
def _mark_legacy_sync_fresh() -> None:
|
||||
"""Called by the lifespan hook after running :func:`_sync_legacy_agents`."""
|
||||
_legacy_sync_state["last_at"] = time.monotonic()
|
||||
|
||||
|
||||
def _ensure_agent_dir_from_record(record: dict[str, Any]) -> None:
|
||||
"""Lazily restore an agent's on-disk artifacts from its DB record.
|
||||
|
||||
@@ -262,14 +306,29 @@ def _record_to_response(
|
||||
record: dict[str, Any],
|
||||
*,
|
||||
include_soul: bool = False,
|
||||
include_extras: bool = True,
|
||||
) -> AgentResponse:
|
||||
_ensure_agent_dir_from_record(record)
|
||||
"""Convert a DB row to an API response.
|
||||
|
||||
Args:
|
||||
include_soul: Read SOUL.md from disk and embed in ``soul``. Set False
|
||||
in list endpoints — the frontend (gallery / recommend / selector)
|
||||
never reads ``soul``, only the edit dialog does. Skipping this
|
||||
avoids one synchronous file read per agent.
|
||||
include_extras: Self-heal a missing agent dir and read deprecated
|
||||
``model`` / ``tool_groups`` / ``skills`` from ``config.yaml``. Set
|
||||
False in list endpoints to skip a stat + yaml parse per agent —
|
||||
those fields are unused by current frontends and the dir restore
|
||||
can happen lazily when the agent is actually opened.
|
||||
"""
|
||||
soul: str | None = None
|
||||
extras: dict[str, Any] = {"skills": None, "model": None, "tool_groups": None}
|
||||
if include_extras:
|
||||
_ensure_agent_dir_from_record(record)
|
||||
extras = _load_config_extras(record["id"])
|
||||
if include_soul:
|
||||
soul = load_agent_soul(record["id"]) or ""
|
||||
|
||||
extras = _load_config_extras(record["id"])
|
||||
|
||||
return AgentResponse(
|
||||
id=record["id"],
|
||||
name=record["name"],
|
||||
@@ -288,6 +347,25 @@ def _record_to_response(
|
||||
)
|
||||
|
||||
|
||||
async def _record_to_response_async(
|
||||
record: dict[str, Any],
|
||||
*,
|
||||
include_soul: bool = False,
|
||||
include_extras: bool = True,
|
||||
) -> AgentResponse:
|
||||
"""Async wrapper: when we actually touch disk, hop to a worker thread so
|
||||
we don't block the asyncio loop. List endpoints (no soul, no extras) skip
|
||||
this overhead and stay on the calling task."""
|
||||
if not include_soul and not include_extras:
|
||||
return _record_to_response(record, include_soul=False, include_extras=False)
|
||||
return await asyncio.to_thread(
|
||||
_record_to_response,
|
||||
record,
|
||||
include_soul=include_soul,
|
||||
include_extras=include_extras,
|
||||
)
|
||||
|
||||
|
||||
async def _get_owned_or_raise(store: AgentStore, agent_id: str, user_id: str) -> dict[str, Any]:
|
||||
owned = await store.get_owned(agent_id, user_id)
|
||||
if owned is not None:
|
||||
@@ -349,7 +427,7 @@ async def list_agents(
|
||||
try:
|
||||
store = get_agent_store(request)
|
||||
tag_store: TagStore = get_tag_store(request)
|
||||
await _sync_legacy_agents(store)
|
||||
await _maybe_sync_legacy_agents(store)
|
||||
user_id = _current_user_id(request)
|
||||
if admin:
|
||||
if not _is_admin_user(request):
|
||||
@@ -357,7 +435,14 @@ async def list_agents(
|
||||
agents = await store.list_all()
|
||||
else:
|
||||
agents = await store.list_visible(user_id)
|
||||
responses = [_record_to_response(a, include_soul=True) for a in agents]
|
||||
# List endpoint stays lean: skip per-agent SOUL.md read, config.yaml
|
||||
# parse, and dir self-heal — none of those feed UI fields the gallery /
|
||||
# recommend dialog / selector actually render. Detail endpoints (single
|
||||
# agent + create/update responses) still load them.
|
||||
responses = [
|
||||
_record_to_response(a, include_soul=False, include_extras=False)
|
||||
for a in agents
|
||||
]
|
||||
|
||||
assignments = await tag_store.list_assignments("agent", [r.id for r in responses])
|
||||
favorite_ids = await store.list_favorite_ids(user_id)
|
||||
@@ -403,11 +488,18 @@ async def get_agent_selector(request: Request) -> AgentSelectorResponse:
|
||||
_require_agents_api_enabled()
|
||||
try:
|
||||
store = get_agent_store(request)
|
||||
await _sync_legacy_agents(store)
|
||||
await _maybe_sync_legacy_agents(store)
|
||||
|
||||
user_id = _current_user_id(request)
|
||||
records = await store.list_featured(user_id)
|
||||
return AgentSelectorResponse(agents=[_record_to_response(r) for r in records])
|
||||
# Selector dropdown only renders name/description/featured_order — same
|
||||
# rationale as list_agents: skip soul + extras + self-heal.
|
||||
return AgentSelectorResponse(
|
||||
agents=[
|
||||
_record_to_response(r, include_soul=False, include_extras=False)
|
||||
for r in records
|
||||
]
|
||||
)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
@@ -427,12 +519,15 @@ async def get_agent(agent_id: str, request: Request) -> AgentResponse:
|
||||
|
||||
try:
|
||||
store = get_agent_store(request)
|
||||
await _sync_legacy_agents(store)
|
||||
await _maybe_sync_legacy_agents(store)
|
||||
user_id = _current_user_id(request)
|
||||
record = await store.get_visible(agent_id, user_id)
|
||||
if record is None:
|
||||
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
|
||||
response = _record_to_response(record, include_soul=True)
|
||||
# Detail view legitimately reads SOUL.md + config.yaml — hop to a
|
||||
# worker thread so the asyncio loop stays responsive under concurrent
|
||||
# detail fetches.
|
||||
response = await _record_to_response_async(record, include_soul=True, include_extras=True)
|
||||
favorite_ids = await store.list_favorite_ids(user_id)
|
||||
response.is_favorite = agent_id in favorite_ids
|
||||
return response
|
||||
@@ -488,7 +583,7 @@ async def create_agent_endpoint(request: AgentCreateRequest, http_request: Reque
|
||||
}
|
||||
)
|
||||
logger.info("Created agent '%s' (%s) at %s", name, agent_id, agent_dir)
|
||||
return _record_to_response(record, include_soul=True)
|
||||
return await _record_to_response_async(record, include_soul=True, include_extras=True)
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
@@ -569,7 +664,7 @@ async def update_agent(agent_id: str, request: AgentUpdateRequest, http_request:
|
||||
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
|
||||
|
||||
logger.info("Updated agent '%s' (%s)", record["name"], agent_id)
|
||||
return _record_to_response(record, include_soul=True)
|
||||
return await _record_to_response_async(record, include_soul=True, include_extras=True)
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
@@ -744,7 +839,7 @@ async def set_agent_featured(
|
||||
|
||||
try:
|
||||
store = get_agent_store(request)
|
||||
await _sync_legacy_agents(store)
|
||||
await _maybe_sync_legacy_agents(store)
|
||||
record = await store.get_any(agent_id)
|
||||
if record is None:
|
||||
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
|
||||
@@ -756,7 +851,7 @@ async def set_agent_featured(
|
||||
if updated is None:
|
||||
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
|
||||
logger.info("Set agent '%s' featured_order=%s by admin", agent_id, body.order)
|
||||
return _record_to_response(updated, include_soul=True)
|
||||
return await _record_to_response_async(updated, include_soul=True, include_extras=True)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
|
||||
@@ -65,6 +65,7 @@ from fastapi.responses import StreamingResponse
|
||||
# identical in connection handling, payload shape, and error mapping.
|
||||
from app.gateway.routers._roundtable_seed import ensure_roundtable_functional_agents
|
||||
from app.gateway.routers.multi_agent import (
|
||||
_ROUNDTABLE_THREAD_METADATA,
|
||||
_auth_headers,
|
||||
_create_thread,
|
||||
_diagnose_exception,
|
||||
@@ -133,6 +134,70 @@ def _normalize_options(raw: object) -> tuple[str, ...]:
|
||||
return tuple(sorted(o for o in out if o))
|
||||
|
||||
|
||||
def _count_past_clarifications(messages: list[dict[str, Any]]) -> int:
|
||||
"""统计当前 thread 内已经发出过的 ``ask_clarification`` 次数。
|
||||
|
||||
用来给 ``stream_intent`` 做"3 轮硬上限"兜底:
|
||||
- 0-1 次:正常流程,模型自行决定继续问还是收敛
|
||||
- 2 次(即将进入第 3 轮):prompt 追加系统提示,要求本轮必须 INTENT_READY
|
||||
- ≥3 次:工具白名单移除 ``ask_clarification``,模型物理上无法再问
|
||||
|
||||
只数"已经成型"的 tool_call,不区分用户是否回答过——LangGraph 把 ask
|
||||
工具的调用永久记录在 AIMessage.tool_calls 里,数它即可。
|
||||
"""
|
||||
if not isinstance(messages, list):
|
||||
return 0
|
||||
count = 0
|
||||
for msg in messages:
|
||||
if not isinstance(msg, dict):
|
||||
continue
|
||||
if (msg.get("type") or "").lower() not in ("ai", "aimessage", "aimessagechunk"):
|
||||
continue
|
||||
for tc in msg.get("tool_calls") or []:
|
||||
if (tc.get("name") or "").lower() == "ask_clarification":
|
||||
count += 1
|
||||
return count
|
||||
|
||||
|
||||
async def _fetch_thread_messages(
|
||||
client: httpx.AsyncClient,
|
||||
base: str,
|
||||
headers: dict[str, str],
|
||||
thread_id: str,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""读取当前 thread checkpoint 的 messages 列表(best-effort)。
|
||||
|
||||
用 ASGITransport loopback 调 ``GET /api/threads/{tid}/state``,失败时返回
|
||||
空列表——计数兜底是个"刹车",失败时退化成原有的纯 SOUL 自律行为,不阻断主流程。
|
||||
"""
|
||||
try:
|
||||
resp = await client.get(f"{base}/api/threads/{thread_id}/state", headers=headers)
|
||||
except Exception as exc:
|
||||
logger.warning("intent: get thread state failed (%s) — falling back to soul-only flow", exc)
|
||||
return []
|
||||
if resp.status_code != 200:
|
||||
logger.warning(
|
||||
"intent: get thread state %s returned %s — falling back to soul-only flow",
|
||||
thread_id,
|
||||
resp.status_code,
|
||||
)
|
||||
return []
|
||||
try:
|
||||
payload = resp.json()
|
||||
except Exception:
|
||||
return []
|
||||
values = payload.get("values") if isinstance(payload, dict) else None
|
||||
if not isinstance(values, dict):
|
||||
return []
|
||||
messages = values.get("messages")
|
||||
return messages if isinstance(messages, list) else []
|
||||
|
||||
|
||||
# 硬上限:超过这个轮数,无论模型怎么想,都强制移除 ask_clarification 工具。
|
||||
# 与 SOUL.md §0 「硬规则 2」对齐——"绝不存在第 4 轮"。
|
||||
_MAX_CLARIFICATION_ROUNDS = 3
|
||||
|
||||
|
||||
def _is_repeated_clarification(
|
||||
messages: list[dict[str, Any]],
|
||||
*,
|
||||
@@ -222,7 +287,7 @@ async def init_intent(request: Request, payload: dict = Body(...)) -> dict[str,
|
||||
# 看不到逐 token 增量。
|
||||
try:
|
||||
async with _make_loopback_client(request) as client:
|
||||
thread_id = await _create_thread(client, base, headers)
|
||||
thread_id = await _create_thread(client, base, headers, metadata=_ROUNDTABLE_THREAD_METADATA)
|
||||
except HTTPException:
|
||||
# 上游已用结构化方式失败,原样传给前端不需要二次包装。
|
||||
raise
|
||||
@@ -279,33 +344,70 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
|
||||
# 因此既无须 trust_env=False 防代理,也不存在 loopback 40s 超时这种坑。
|
||||
async with _make_loopback_client(request) as client:
|
||||
try:
|
||||
# —— 3 轮硬上限兜底 ——
|
||||
# 在发起 run 之前先读一次 thread state,数当前 ask_clarification 已发出次数:
|
||||
# past=0/1 → 正常,不动 message,不动工具白名单
|
||||
# past=2(本轮即将变成第 3 轮)→ 给消息加前缀,强提示"必须 INTENT_READY"
|
||||
# past≥3 → 直接把 ask_clarification 也加进 excluded_tools,工具物理消失
|
||||
# 这一步是个"刹车":SOUL 让模型自律,这里让系统兜底。
|
||||
past_clarifications = _count_past_clarifications(
|
||||
await _fetch_thread_messages(client, base, headers, thread_id)
|
||||
)
|
||||
|
||||
# 1. ask_clarification 的工具白名单——超过硬上限就移除。
|
||||
base_excluded_tools = [
|
||||
"web_search",
|
||||
"present_files",
|
||||
"view_image",
|
||||
"agent_orchestration",
|
||||
"bash",
|
||||
"ls",
|
||||
"read_file",
|
||||
"write_file",
|
||||
"str_replace",
|
||||
"memory",
|
||||
"hindsight_recall",
|
||||
"hindsight_reflect",
|
||||
"hindsight_retain",
|
||||
"task",
|
||||
"write_todos",
|
||||
]
|
||||
if past_clarifications >= _MAX_CLARIFICATION_ROUNDS:
|
||||
base_excluded_tools.append("ask_clarification")
|
||||
logger.info(
|
||||
"intent stream: past_clarifications=%d >= %d, force-disabling "
|
||||
"ask_clarification to break the loop",
|
||||
past_clarifications,
|
||||
_MAX_CLARIFICATION_ROUNDS,
|
||||
)
|
||||
|
||||
# 2. message 前缀——给模型一个明确的"本轮必须收敛"信号。
|
||||
effective_message = message
|
||||
if past_clarifications >= _MAX_CLARIFICATION_ROUNDS:
|
||||
effective_message = (
|
||||
f"[系统强制] 已澄清 {past_clarifications} 轮,触发硬上限。"
|
||||
f"本轮请直接输出 [INTENT_READY],缺的字段用 assumptions 兜底,"
|
||||
f"绝对不要再调任何工具,也不要再询问任何问题。\n\n"
|
||||
f"用户最新输入:{message}"
|
||||
)
|
||||
elif past_clarifications >= _MAX_CLARIFICATION_ROUNDS - 1:
|
||||
effective_message = (
|
||||
f"[系统提示] 已澄清 {past_clarifications} 轮,这是最后一轮。"
|
||||
f"本轮请直接输出 [INTENT_READY],未明确的字段用 assumptions 兜底。\n\n"
|
||||
f"用户最新输入:{message}"
|
||||
)
|
||||
|
||||
body = _run_payload(
|
||||
agent_name=INTENT_AGENT_ID,
|
||||
model_name=model_name,
|
||||
thread_id=thread_id,
|
||||
new_message=message,
|
||||
# Intent agent only ever needs ask_clarification. Block
|
||||
new_message=effective_message,
|
||||
# Intent agent only ever needs ask_clarification (and even that
|
||||
# is yanked once past_clarifications hits the cap above). Block
|
||||
# everything else so the model can't drift into solutioning
|
||||
# (e.g. saving a design doc with write_file) — the model
|
||||
# tends to ignore SOUL prose, hard exclusion is reliable.
|
||||
# 仅保留 ask_clarification;其余工具见模块 docstring「工具白名单」。
|
||||
excluded_tools=[
|
||||
"web_search",
|
||||
"present_files",
|
||||
"view_image",
|
||||
"agent_orchestration",
|
||||
"bash",
|
||||
"ls",
|
||||
"read_file",
|
||||
"write_file",
|
||||
"str_replace",
|
||||
"memory",
|
||||
"hindsight_recall",
|
||||
"hindsight_reflect",
|
||||
"hindsight_retain",
|
||||
"task",
|
||||
"write_todos",
|
||||
],
|
||||
# (e.g. saving a design doc with write_file) — SOUL prose is
|
||||
# not reliable on its own, hard exclusion is.
|
||||
excluded_tools=base_excluded_tools,
|
||||
skill_stop_names=[],
|
||||
)
|
||||
|
||||
@@ -338,7 +440,13 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
|
||||
except json.JSONDecodeError as exc:
|
||||
logger.warning("intent summary JSON parse failed: %s", exc)
|
||||
summary = None
|
||||
ready_text = msg_text
|
||||
# SOUL §0 硬规则 3 要求 ```json``` 反引号闭合后不得再输出任何字符,
|
||||
# 但模型仍偶尔会在后面接 "那么我帮你列一下大纲……" 之类的尾巴。
|
||||
# 终态帧只回到 marker 块结尾 + 一个换行,把废话从 content 里裁掉——
|
||||
# 草稿持久化 / 跨设备重放 / 前端 splitIntentReadyBlock 的 after
|
||||
# 段都因此干净。流式期间用户仍会逐 token 看到那段废话,前端
|
||||
# Step1Panel 已经把 `parts.after` 隐藏掉,体验闭环。
|
||||
ready_text = msg_text[: m.end()].rstrip() + "\n"
|
||||
break
|
||||
if summary and ready_text:
|
||||
final_frame = {
|
||||
|
||||
@@ -53,18 +53,28 @@ 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"])
|
||||
|
||||
|
||||
# 圆桌(intent / recommend / 席位 / 协调)创建的 thread 都是功能型内部线程,
|
||||
# 不是用户的普通对话。打上这个 metadata,前端「最近的对话」列表(recent-chat-list.tsx)
|
||||
# 据 `metadata.thread_type` 过滤掉,与 scheduler / bootstrap 系统会话同一套机制。
|
||||
_ROUNDTABLE_THREAD_METADATA: dict[str, Any] = {"thread_type": "roundtable", "system": True}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 共享:异常诊断器(intent/recommend/multi-agent 共用)
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -285,9 +295,22 @@ def _auth_headers(request: Request) -> dict[str, str]:
|
||||
return {"Authorization": token, "Content-Type": "application/json"}
|
||||
|
||||
|
||||
async def _create_thread(client: httpx.AsyncClient, base: str, headers: dict[str, str]) -> str:
|
||||
"""``POST /api/threads`` → 新 thread_id(intent/recommend/init 席位均用)。"""
|
||||
resp = await client.post(f"{base}/api/threads", json={}, headers=headers)
|
||||
async def _create_thread(
|
||||
client: httpx.AsyncClient,
|
||||
base: str,
|
||||
headers: dict[str, str],
|
||||
*,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
) -> str:
|
||||
"""``POST /api/threads`` → 新 thread_id(intent/recommend/init 席位均用)。
|
||||
|
||||
``metadata`` 写入 thread 元数据,圆桌调用方传 ``_ROUNDTABLE_THREAD_METADATA``
|
||||
把这些功能型线程标记成可被前端「最近的对话」过滤掉的系统会话。
|
||||
"""
|
||||
body: dict[str, Any] = {}
|
||||
if metadata:
|
||||
body["metadata"] = metadata
|
||||
resp = await client.post(f"{base}/api/threads", json=body, headers=headers)
|
||||
resp.raise_for_status()
|
||||
return resp.json()["thread_id"]
|
||||
|
||||
@@ -323,6 +346,181 @@ 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, *, metadata: dict[str, Any] | None = None) -> str:
|
||||
"""同进程版的 ``POST /api/threads``。
|
||||
|
||||
完全镜像 ``routers/threads.create_thread`` 的写盘动作(thread_meta +
|
||||
空 checkpoint),但跳过 TCP + Pydantic 反序列化。仅供 init 阶段批量建席位
|
||||
线程用,失败时抛 HTTPException 由 SSE 帧上报。
|
||||
|
||||
``metadata`` 写入 thread 元数据;init 传 ``_ROUNDTABLE_THREAD_METADATA`` 把
|
||||
席位 / 协调线程标记成系统会话,从前端「最近的对话」里隐藏。
|
||||
"""
|
||||
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=metadata or {})
|
||||
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 +654,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 +663,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 +683,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 +744,132 @@ 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, metadata=_ROUNDTABLE_THREAD_METADATA)
|
||||
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, metadata=_ROUNDTABLE_THREAD_METADATA)
|
||||
)
|
||||
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 +1056,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(
|
||||
|
||||
@@ -49,6 +49,7 @@ from fastapi.responses import StreamingResponse
|
||||
|
||||
from app.gateway.routers._roundtable_seed import ensure_roundtable_functional_agents
|
||||
from app.gateway.routers.multi_agent import (
|
||||
_ROUNDTABLE_THREAD_METADATA,
|
||||
_auth_headers,
|
||||
_create_thread,
|
||||
_flatten_content,
|
||||
@@ -237,7 +238,7 @@ async def stream_recommend(request: Request, payload: dict = Body(...)) -> Strea
|
||||
async with _make_loopback_client(request) as client:
|
||||
try:
|
||||
# 每次推荐独立 thread;prompt 已包含完整 intent + 候选列表,无需多轮。
|
||||
thread_id = await _create_thread(client, base, headers)
|
||||
thread_id = await _create_thread(client, base, headers, metadata=_ROUNDTABLE_THREAD_METADATA)
|
||||
|
||||
body = _run_payload(
|
||||
agent_name=RECOMMENDER_AGENT_ID,
|
||||
@@ -253,6 +254,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] = []
|
||||
|
||||
+10
@@ -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()
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
"""Configuration and loaders for custom agents."""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
@@ -13,6 +15,25 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
SOUL_FILENAME = "SOUL.md"
|
||||
AGENT_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$")
|
||||
|
||||
# Per-file mtime caches. These are read on every /api/agents list/detail request
|
||||
# (one read per agent) — caching keyed by (path, mtime_ns, size) is correct
|
||||
# because any edit to the file updates mtime, and an external tool re-writing
|
||||
# the same bytes doesn't actually change the result we'd return. Cache size is
|
||||
# unbounded but lookup is O(1) and entries are tiny (one parsed dict / string
|
||||
# per agent file); for the order of N≤10⁴ agents the memory cost is negligible.
|
||||
_config_cache: dict[str, tuple[tuple[int, int], "AgentConfig | None"]] = {}
|
||||
_soul_cache: dict[str, tuple[tuple[int, int], str | None]] = {}
|
||||
_cache_lock = threading.Lock()
|
||||
|
||||
|
||||
def _file_stat_key(path) -> tuple[int, int] | None:
|
||||
"""Return (mtime_ns, size) for cache invalidation, or None when missing."""
|
||||
try:
|
||||
st = os.stat(path)
|
||||
except OSError:
|
||||
return None
|
||||
return (st.st_mtime_ns, st.st_size)
|
||||
# Backward-compatible alias. Historically this validated the display name;
|
||||
# after the DB-backed agent table, display names can be Chinese and duplicated.
|
||||
# The pattern now applies only to storage/runtime ids.
|
||||
@@ -94,6 +115,18 @@ def load_agent_config(agent_id: str | None) -> AgentConfig | None:
|
||||
)
|
||||
return None
|
||||
|
||||
# mtime-based cache. The config file is rewritten every time the agent is
|
||||
# edited via PUT /api/agents/{id}, which bumps mtime — so the next read
|
||||
# picks up the change automatically. Missing config_file = key None below,
|
||||
# bypassing the cache (cheap path: just return AgentConfig with id/name).
|
||||
cache_path_key = str(config_file)
|
||||
stat_key = _file_stat_key(config_file) if config_file.exists() else None
|
||||
if stat_key is not None:
|
||||
with _cache_lock:
|
||||
cached = _config_cache.get(cache_path_key)
|
||||
if cached is not None and cached[0] == stat_key:
|
||||
return cached[1]
|
||||
|
||||
data: dict[str, Any] = {}
|
||||
if config_file.exists():
|
||||
try:
|
||||
@@ -112,7 +145,11 @@ def load_agent_config(agent_id: str | None) -> AgentConfig | None:
|
||||
known_fields = set(AgentConfig.model_fields.keys())
|
||||
data = {k: v for k, v in data.items() if k in known_fields}
|
||||
|
||||
return AgentConfig(**data)
|
||||
result = AgentConfig(**data)
|
||||
if stat_key is not None:
|
||||
with _cache_lock:
|
||||
_config_cache[cache_path_key] = (stat_key, result)
|
||||
return result
|
||||
|
||||
|
||||
def load_agent_soul(agent_id: str | None) -> str | None:
|
||||
@@ -131,8 +168,21 @@ def load_agent_soul(agent_id: str | None) -> str | None:
|
||||
soul_path = agent_dir / SOUL_FILENAME
|
||||
if not soul_path.exists():
|
||||
return None
|
||||
# mtime-based cache, same rationale as load_agent_config: SOUL.md is
|
||||
# rewritten on edit (PUT /api/agents/{id}), which bumps mtime.
|
||||
cache_path_key = str(soul_path)
|
||||
stat_key = _file_stat_key(soul_path)
|
||||
if stat_key is not None:
|
||||
with _cache_lock:
|
||||
cached = _soul_cache.get(cache_path_key)
|
||||
if cached is not None and cached[0] == stat_key:
|
||||
return cached[1]
|
||||
content = soul_path.read_text(encoding="utf-8").strip()
|
||||
return content or None
|
||||
result = content or None
|
||||
if stat_key is not None:
|
||||
with _cache_lock:
|
||||
_soul_cache[cache_path_key] = (stat_key, result)
|
||||
return result
|
||||
|
||||
|
||||
def list_custom_agents() -> list[AgentConfig]:
|
||||
|
||||
Reference in new issue
Block a user