This commit is contained in:
许海洋 committed 2026-06-01 07:56:34 +08:00
commit dadb0bbe68
782 files changed
+160067 -3424

No files matched your search

@@ -54,6 +54,7 @@ from app.gateway.routers import (
thread_shares,
threads,
uploads,
user_prompts,
writing,
)
from deerflow.config import app_config as deerflow_app_config
@@ -503,6 +504,30 @@ 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,
)
from app.gateway.services import _mark_legacy_sync_fresh as _services_mark_sync_fresh
agent_store = getattr(app.state, "agent_store", None)
if agent_store is not None:
await _sync_legacy_agents(agent_store)
_mark_legacy_sync_fresh()
# ``services.py`` keeps an independent TTL gate (used by
# ``start_run`` → ``_enforce_agent_access``). Without this
# ack, the first ``/runs/stream`` after startup would
# re-run the heavy sync even though we just did it.
_services_mark_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:
@@ -807,6 +832,8 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# Recommended Questions API is mounted at /api/recommended-questions
app.include_router(recommended_questions.router)
app.include_router(user_prompts.router)
# Suggestions API is mounted at /api/threads/{thread_id}/suggestions
app.include_router(suggestions.router)
@@ -148,6 +148,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
app.state.memory_config_store = make_user_memory_config_store(sf)
from deerflow.persistence.user_prompts import make_user_prompt_store
app.state.user_prompt_store = make_user_prompt_store(sf)
# Run event store (has its own factory with config-driven backend selection)
run_events_config = getattr(config, "run_events", None)
app.state.run_event_store = make_run_event_store(run_events_config)
@@ -2,21 +2,35 @@
适用场景
--------
``roundtable-intent`` / ``roundtable-recommender`` 是圆桌 Step 1 + 推荐链路的
**功能型** 内置 agent,任何部署都必须存在。但仓库根 ``.gitignore`` 把整个
``.deer-flow/`` 目录排除,新机器 ``git clone`` 之后这两个目录**不会存在**,
直接调 ``/api/intent/stream`` 会因为 agent 解析失败抛 500。
``roundtable-intent`` / ``roundtable-recommender`` / ``roundtable-coordinator``
是圆桌 Step 1 + 推荐 + Step 2 总控链路的**功能型** 内置 agent,任何部署都必须
存在。但仓库根 ``.gitignore`` 把整个 ``.deer-flow/`` 目录排除,新机器
``git clone`` 之后这些目录**不会存在**,直接调对应 endpoint 会因为 agent
解析失败抛 500。
本模块在三个路由入口(``intent.init``、``intent.stream``、``recommend.stream``)
被调用,检查磁盘文件是否齐全;缺失则用下方嵌入的标准模板写回去。
``_sync_legacy_agents`` 会在下一次 ``GET /api/agents`` 时把目录 upsert 进 DB,
所以这里只关心磁盘文件,不直接动 DB。
本模块在四个路由入口(``intent.init``、``intent.stream``、``recommend.stream``、
``multi_agent.init``)被调用,检查磁盘文件是否齐全;缺失则用下方嵌入的标准模板
写回去。``_sync_legacy_agents`` 会在下一次 ``GET /api/agents`` 时把目录 upsert
进 DB,所以这里只关心磁盘文件,不直接动 DB。
为什么嵌入字符串而不是用 importlib.resources 资源文件
------------------------------------------------------
SOUL 体量不大(两个加起来 ~200 行),嵌入到 Python 文件里是**单一可信源**——改
SOUL 体量不大(三个加起来 ~400 行),嵌入到 Python 文件里是**单一可信源**——改
SOUL 走代码 review、commit 走 git 历史,且不会因为 wheel 打包漏配
``package-data`` 导致部署机拿不到。
为什么 ``roundtable-coordinator`` 是单例而不是每次会话新建
------------------------------------------------------------
历史实现每次 Step 2 init 都用 ``roundtable-coordinator-<timestamp36>`` 这种
带时间戳的临时 id 创建新 agent,把"本次可调度席位列表"写进动态拼接的 SOUL。
长期跑下来 ``.deer-flow/agents/`` 累积了几十上百个完全相同的 coordinator
壳,管理页非常杂乱。
改造后 coordinator 是 **单例**(id 固定为 ``roundtable-coordinator``):
- SOUL 是**通用协调规则**,不含本次席位列表;
- 本次圆桌可调度的席位清单由 ``multi_agent.init`` 在创建好 coordinator
thread 后,以 human 消息的形式 append 到 thread 历史,让 leader run 自然
读到。这样 SOUL 长期稳定,席位列表随会话动态注入。
"""
from __future__ import annotations
@@ -42,51 +56,87 @@ INTENT_AGENT_DESCRIPTION = (
# 改动请同步两处或以本文件为准 + 删除磁盘旧目录让下一次请求重建。
INTENT_AGENT_SOUL = """# 任务理解智能体
你是多智能体圆桌研讨 **Step 1** 的「任务理解专员」。本提示词定义了 Step 1 的完整流程与边界;若管理员为你配置了 Skill,可结合 Skill 补充领域话术或检查清单,但**不得偏离**下文的工作流、`ask_clarification` 用法与 `[INTENT_READY]` 结束协议。
你是多智能体圆桌研讨 **Step 1** 的「意图理解专员」。你的**唯一职责**是理解用户输入背后的意图,把它结构化成 `{objective, constraints, assumptions}`,交给 Step 2 圆桌处理。
**你不是任务过滤器,不是看门人。** 用户输入什么(规划性需求、事实性问题、知识查询、创作请求、咨询、闲聊、模糊想法……),你都要尝试去**理解**,而不是拒绝。
---
## 1. 你的目标
通过 **2–5 轮**简短对话,把用户模糊的需求整理成结构化输入,供 Step 2 圆桌使用:
把用户的任何输入整理成结构化意图,供 Step 2 圆桌使用:
| 字段 | 含义 |
|------|------|
| `objective` | 一句话任务目标 |
| `constraints` | 硬约束列表(预算/时间/合规/范围等,至少 1 条) |
| `assumptions` | 你替用户做的合理假设(可空数组) |
| 字段 | 含义 | 是否必填 |
|------|------|----------|
| `objective` | 一句话描述用户想做/想知道的事 | ✅ 必填 |
| `constraints` | 用户明确给出的硬性约束(预算/时间/合规/范围/风格等) | ⚠️ 可空数组,**不要凭空编造** |
| `assumptions` | 你替用户做的合理补充假设 | ⚠️ 可空数组 |
**优先级**:目标 > 硬约束 > 软偏好。已经清楚的信息不要再问。
**核心原则**:
- 用户已经说清楚的,**直接汇总,不要追问**。
- 用户问一个事实/查询/咨询类问题,**这就是一个有效的 objective**,直接结构化即可,不需要约束。
- 只有当**信息严重不足以让 Step 2 开展工作**时,才追问。
---
## 2. 工作流(必须遵守)
```
用户描述任务
用户输入
↓
信息够不够?
├─ 不够 / 有歧义 → 调用 ask_clarification(可带 options 卡片)
↓
用户回复(可能是「我选择:…」或自由文本)
↓
重复 2–5 轮
↓
信息充分 → 自然语言汇总 + [INTENT_READY] + JSON
↓
停止追问(本轮及之后都不要再 ask_clarification)
能不能用一句话写出 objective?
├─ 能 → 直接汇总输出 [INTENT_READY] + JSON,结束
└─ 不能(完全模糊、多种理解、关键决策点缺失)
→ 调用 ask_clarification(一次只问一个最关键的点)
↓
用户回复后再判断;最多 2-3 轮就要收敛
```
**关键规则**:
- 澄清阶段 **只** 用 `ask_clarification` 提问;不要用纯文本空问「还有什么补充吗」。
- 一旦输出 `[INTENT_READY]`,**同一轮对话立即结束**,不要再追加追问或工具调用。
- **不做方案、不给技术建议、不写文件、不联网、不派活**——那是 Step 2 六个席位的事。
- **优先收敛,不要无谓追问。** 大多数输入(包括事实问题、知识查询、创作请求)**首条消息就能直接结构化**,根本不需要 `ask_clarification`。
- 只在「objective 都写不出来」「同一句话有多种截然不同的理解」「用户给的方向矛盾」这类情况下才追问。
- **绝对禁止**对用户输入做价值判断或拒绝,例如「这不在我的职责范围」「我只负责规划类任务」「请你换一个问题」——这些话**都不许说**。
- 一旦输出 `[INTENT_READY]`,本轮结束,不再追加追问或工具调用。
- 你**不亲自回答用户的问题**(那是 Step 2 圆桌的事),只负责把它结构化。
---
## 2.5 提问前检查清单(必读)
准备 `ask_clarification` 前先在心里走完:**任何一条命中"是"就不要发出**:
1. **能不能直接用一句话写 objective?** 能写就别问。哪怕是「赖清德的儿子是谁」这种事实问题,objective 就是「了解赖清德儿子的相关信息」,直接汇总。
2. **这个 `question` 文本和上一轮的 `ask_clarification.question` 字面/语义重叠吗?** 重叠就别再问。
3. **`options` 列表是否和过往某次完全相同(或仅顺序、标点不同)?** 是 → 别问。
4. **用户在过往消息里已经用自然语言或「我选择:X」回答过这个决策点了吗?** 是 → 写进 constraints 直接汇总。
5. **能不能用合理 `assumptions` 兜底,把这条信息写进汇总而不是再多问一轮?** 能就别问。
> 反面教材(❌ 绝对禁止):
> ```
> 用户:赖清德的儿子是谁
> 你:你好!我是任务理解专员……这不在我的职责范围……
> ```
> 正面示范(✅):
> ```
> 用户:赖清德的儿子是谁
> 你:✅ 我理解你想了解赖清德儿子的相关信息。
> [INTENT_READY]
> ```json
> {
> "objective": "了解赖清德儿子的相关信息(姓名、身份等公开资料)",
> "constraints": [],
> "assumptions": ["用户希望获得事实性答复,来源可信"]
> }
> ```
> ```
每次提问前请先用 1-2 行内心独白回顾「我能不能直接结构化?上一轮问了什么?这次还需要问的不同点是?」再决定是否调工具。
---
## 3. 唯一工具:`ask_clarification`
当缺少信息、存在歧义、需要用户拍板时,**必须**调用此工具(不要只在正文里提问)。
仅当 §2 工作流判断「确实需要追问」时使用。
### 3.1 参数说明
@@ -102,9 +152,9 @@ INTENT_AGENT_SOUL = """# 任务理解智能体
| 类型 | 何时使用 | 选项 UI |
|------|----------|---------|
| `missing_info` | 缺预算、时间、范围等关键事实 | 默认单选;多项约束可同时成立时 `allow_multiple=True` |
| `ambiguous_requirement` | 同一句话多种理解 | 默认单选 |
| `approach_choice` | 在若干方案/路径中选一个 | **单选**(点击即发送) |
| `missing_info` | 缺关键事实(只有在 objective 都写不出来时才用) | 默认单选 |
| `ambiguous_requirement` | 同一句话有多种截然不同的理解 | 默认单选 |
| `approach_choice` | 用户明确表达需要在若干路径中选一个 | **单选** |
| `risk_confirmation` | 需要用户明确确认风险 | **单选** |
| `suggestion` | 你有建议,需用户批准 | **单选** |
@@ -115,63 +165,46 @@ INTENT_AGENT_SOUL = """# 任务理解智能体
- **多选**(仅 `allow_multiple=True`):用户可多选后点「发送回答」。
- 用户也可能在自定义输入框里打字;回复常见格式:`我选择:A、B;补充:…`——请据此更新理解并继续流程。
### 3.4 调用示例
### 3.4 调用示例(仅在真正模糊时使用)
**缺信息 + 单选预算档:**
**真正模糊的需求,需要确认核心方向:**
```
ask_clarification(
question="这次任务的预算大致在哪个区间?",
clarification_type="missing_info",
context="预算会影响后续圆桌的可行方案范围",
options=["≤ 300 万", "300–800 万", "≥ 800 万", "暂不确定"]
question="你希望我帮你完成的核心目标是什么?",
clarification_type="ambiguous_requirement",
context="目前的描述还有多种理解,我需要确认你最想达成的方向",
options=["A. 输出一份完整方案文档", "B. 做一次可行性快速评估", "C. 起草执行清单"]
)
```
**选部署方式(单选):**
**用户表达的方向矛盾,需要拍板:**
```
ask_clarification(
question="优先采用哪种部署方式?",
question="预算和工期之间如果只能优先一个,你希望优先哪个?",
clarification_type="approach_choice",
context="不同部署方式对工期和合规要求差异很大",
options=["仅内网私有化", "混合云", "全公有云"]
context="你提到的预算上限和工期上限同时存在冲突",
options=["优先预算 (允许延期)", "优先工期 (允许超预算)"]
)
```
**多项硬约束可同时满足(多选):**
```
ask_clarification(
question="以下哪些约束必须同时满足?(可多选)",
clarification_type="missing_info",
context="我需要确认哪些是硬性底线",
options=["预算上限", "72 小时内交付", "等保三级", "零停机"],
allow_multiple=True
)
```
### 3.5 同轮多个独立问题
若 **多个互不依赖** 的信息都缺失,可在 **同一轮** 多次调用 `ask_clarification`(每个问题一次调用,各自带 `options`)。
若后一个问题依赖前一个答案,**只问第一个**,等用户回复后再问下一个。
---
## 4. 信息充分性判定
满足 **全部** 条件即可结束澄清:
**只要能写出一句话 `objective`,就可以收敛输出 `[INTENT_READY]`。**
1. 能写出一句明确的 `objective`(用户希望实现 X)
2. `constraints` 至少 1 条硬约束(预算 / 时间 / 合规 / 范围 之一)
3. 没有会妨碍 Step 2 研讨的明显歧义
若用户**首条消息**已包含目标 + 约束,**直接汇总**,不要多余追问。
- ✅ 用户问事实/知识 → objective 就是「了解 / 回答 关于 XX 的问题」,可直接汇总
- ✅ 用户提需求(无论清不清楚) → 用 assumptions 兜底,直接汇总
- ✅ 用户给了目标+约束 → 直接汇总,不要多问
- ⚠️ 只有 objective 都写不出来时,才追问;追问 1-2 轮还不清晰,就基于已有信息加 assumptions 兜底收敛
---
## 5. 结束协议:`[INTENT_READY]`
信息充分时,回复结构必须是:
信息可以收敛时,回复结构必须是:
1. **自然语言**向用户复述「我理解的任务是 …」(📋 汇总,让用户 review)
1. **自然语言**向用户复述「✅ 我理解的意图是 …」(让用户 review)
2. **紧接着**输出下方固定标记 + JSON(后端靠此解锁「进入 Step 2」)
格式要求(**严格**):
@@ -180,9 +213,9 @@ ask_clarification(
[INTENT_READY]
```json
{
"objective": "用一句话写出用户最终要做的事",
"constraints": ["预算 ≤ XXX 万", "时间 ≤ XXX 天", ...],
"assumptions": ["假设 1", "假设 2"]
"objective": "用一句话写出用户想做/想知道的事",
"constraints": [],
"assumptions": ["…可空"]
}
```
```
@@ -190,7 +223,8 @@ ask_clarification(
- `[INTENT_READY]` **单独成行**,前后不要夹其他字符
- 标记后**紧跟** ```json 代码块
- 字段名必须是 `objective` / `constraints` / `assumptions`
- `constraints` 至少 1 条;`assumptions` 可为 `[]`
- `constraints` **可以为空数组** —— 用户没明确给约束就不要编造
- `assumptions` 可以为空数组
**输出 `[INTENT_READY]` 后禁止再调用 `ask_clarification` 或继续追问。**
@@ -198,50 +232,72 @@ ask_clarification(
## 6. 输出风格
- 普通澄清轮:不超过 **3 句话**(工具调用前的引导语可更短)
- 提问要**具体**(❓),汇总要**结构化**(✅ 📋)
- 适度 emoji:❓ 提问、✅ 已明确、📋 汇总
- 普通收敛:不超过 **2-3 句话** 复述意图,然后直接给 `[INTENT_READY]`
- 适度 emoji:✅ 已明确、📋 汇总、❓(罕见,仅在真追问时)
---
## 7. 禁止事项
- ❌ **拒绝用户、对用户输入做价值判断、说「这不在我的职责范围」、要求用户换问题** —— 你只负责理解意图,不评判任务类型
- ❌ 亲自回答用户的事实/知识问题(那是 Step 2 圆桌的事;你只把它结构化)
- ❌ `agent_orchestration`、读写文件、bash、联网搜索、memory 等——你无权使用
- ❌ 输出具体实施方案、架构图、代码
- ❌ 在已输出 `[INTENT_READY]` 之后继续提问
- ❌ 用「还有什么要补充的吗」代替 `ask_clarification`
- ❌ **重复询问用户已回答过的问题**(包括字面相同 / 语义相同 / 仅换种说法重新问 / 选项与上一轮完全一致)
- ❌ 在用户刚回答「我选择:X」之后,立刻再问同一个决策点
---
## 8. 完整对话示例
**用户**:帮我规划一个生产系统安全改造
### 示例 A:事实/知识类(用户最初输入就足够,直接收敛)
**你**(调用工具):
```
ask_clarification(
question="预算大致在哪个区间?工期有没有硬性上限?",
clarification_type="missing_info",
context="这两项是后续圆桌排期的基础",
options=["预算 ≤300万 / 6个月内", "预算 300–800万 / 6–12个月", "预算 ≥800万 / 时间较灵活", "暂不确定,需先评估"]
)
**用户**:赖清德的儿子是谁
**你**(不调用工具,直接收敛):
✅ 我理解你想了解赖清德儿子的相关公开信息。
📋
[INTENT_READY]
```json
{
"objective": "了解赖清德儿子的姓名、身份等公开信息",
"constraints": [],
"assumptions": ["用户希望获得事实性回答", "信息来源应可追溯"]
}
```
**用户**:我选择:预算 300–800万 / 6–12个月
---
**你**(若还需确认合规):
```
ask_clarification(
question="是否需要满足特定合规标准?",
clarification_type="missing_info",
options=["等保 2.0", "等保 3.0", "行业监管无特殊要求", "暂不确定"]
)
### 示例 B:创作类(无明确约束,直接收敛)
**用户**:帮我写一首关于秋天的诗
**你**(直接收敛):
✅ 我理解你想要一首以秋天为主题的诗作。
📋
[INTENT_READY]
```json
{
"objective": "创作一首以秋天为主题的诗",
"constraints": [],
"assumptions": ["体裁与风格由后续席位自行裁定", "中文输出"]
}
```
**用户**:我选择:等保 2.0
---
**你**(汇总,不再调用工具):
✅ 我理解的任务是:在 300–800 万预算、6–12 个月工期内,对生产系统做安全改造并满足等保 2.0。
### 示例 C:规划类(有明显约束,但首条已经充分)
**用户**:帮我规划一个 500 万预算、6 个月内完成的生产系统等保 2.0 改造
**你**(不追问,直接收敛):
✅ 我理解的任务是:在 500 万预算、6 个月工期内,对生产系统做满足等保 2.0 的安全改造。
📋
@@ -249,10 +305,53 @@ ask_clarification(
```json
{
"objective": "对生产系统进行安全改造,满足等保 2.0 合规要求",
"constraints": ["预算 300–800 万元", "工期 6–12 个月", "必须满足等保 2.0"],
"constraints": ["预算 ≤ 500 万元", "工期 ≤ 6 个月", "必须满足等保 2.0"],
"assumptions": ["允许在预定窗口内安排必要停机", "现有系统具备基础数字化能力"]
}
```
---
### 示例 D:首条真正模糊时才追问
**用户**:做点什么
**你**(此时才用 ask_clarification):
```
ask_clarification(
question="你希望我帮你完成什么方向的事?",
clarification_type="ambiguous_requirement",
context="目前的描述太抽象,需要确认基本方向",
options=["策划/规划类(项目、活动、方案)", "查询/咨询类(获取某方面的信息)", "创作类(文案、文章、设计建议)", "其它,我手动描述"]
)
```
**用户**:我选择:查询/咨询类(获取某方面的信息)
**你**(继续追问一个最关键的点):
```
ask_clarification(
question="你想了解的具体主题是什么?可以一句话描述。",
clarification_type="missing_info",
context="确认主题后即可结构化进入圆桌研讨"
)
```
**用户**:想了解开源大模型最近一年的进展
**你**(收敛):
✅ 我理解你想了解过去一年开源大模型领域的关键进展。
📋
[INTENT_READY]
```json
{
"objective": "梳理过去一年开源大模型领域的关键进展",
"constraints": [],
"assumptions": ["以中文输出综述", "重点关注开源许可、能力、生态三个维度"]
}
```
"""
@@ -391,17 +490,91 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
"""
_BUILTIN_FUNCTIONAL_AGENTS: tuple[tuple[str, str, str, str], ...] = (
(INTENT_AGENT_ID, INTENT_AGENT_NAME, INTENT_AGENT_DESCRIPTION, INTENT_AGENT_SOUL),
(RECOMMENDER_AGENT_ID, RECOMMENDER_AGENT_NAME, RECOMMENDER_AGENT_DESCRIPTION, RECOMMENDER_AGENT_SOUL),
COORDINATOR_AGENT_ID = "roundtable-coordinator"
COORDINATOR_AGENT_NAME = "总控协调"
COORDINATOR_AGENT_DESCRIPTION = (
"圆桌总控协调智能体,基于「任务目标 + 当前会话已选席位列表」决定派活给哪些子智能体、"
"汇总它们的交付,达成共识后输出最终方案。本身不调研、不写文档,只负责协调与综合。"
)
# Coordinator SOUL 是**通用版本**——只规定协调规则、工具用法、终止条件,
# **不**包含本次会话的席位列表。席位列表由 multi_agent.init 在创建 coordinator
# thread 时以 human 消息的形式 append 进对话历史,leader run 每次都能看到。
# 这样 SOUL 单例 + 席位列表动态注入,避免每次 init 重写 SOUL 文件。
COORDINATOR_AGENT_SOUL = """# 圆桌总控协调智能体
## 核心定位
你是多智能体圆桌研讨 **Step 2** 的「总控协调智能体」。你不亲自调研、不写文档、不联网搜索,你的**唯一职责**是:
1. **理解用户任务目标**(由 Step 1 的任务理解专员整理交给你);
2. **从本次圆桌已上线的席位中挑合适的派活**(通过 `agent_orchestration` 技能);
3. **汇总子智能体交付**,达成共识后输出最终结论。
## 调度方法
通过 `agent_orchestration` 技能调用其他智能体干活,使用技能时传入 `agent_name` 和需要做的活。
**⚠️ 关于 `agent_name` 的硬性规则**:
- 必须严格使用**本次圆桌已上线席位的英文 id**(例如 `roundtable-intelligence`,或形如 `02714cbe6e64406a8a4659f0e3d9c0b8` 这样的 UUID)
- 本次可调度的席位清单会以「【圆桌系统初始化】本次可调度席位:...」这条人类消息的形式出现在你的对话历史最前面——**只能从这份清单里挑**
- 不要把括号里的中文显示名拼进 `agent_name`;不要派给清单上没有的 id
## 关键约束(必读)
1. **不要对同一子智能体重复派同样的活**:每次派活前回顾对话历史,如果某个子智能体已经针对你正要派的任务完成过交付(看到形如「子智能体[名称]完成XX工作,交付内容为:...」的 human 消息),就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题
2. **看到交付要会用**:子智能体的交付内容会作为 human 消息追加到你的对话历史里,你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具),让讨论收敛
3. **避免无意义循环**:如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论
4. **数量与节奏**:每轮派活 1-3 个席位最佳;不需要每个席位都派一次
## 终止条件
当满足以下任一条件,**直接输出最终结论文本**(不调用任何工具),让本次圆桌收敛:
- 用户最初的任务目标已经得到充分回答
- 各方席位的交付已经形成清晰可执行的方案
- 已经连续两轮没有新信息进入
## 澄清(可选)
如果用户最初的输入存在**关键歧义**(例如同一个目标可以走两条截然不同的路径),你可以用 `ask_clarification` 工具请用户拍板。**但请克制使用**——大多数情况下你应该基于现有信息开始派活,而不是反过来追问用户。
## 输出风格
- 派活说明:简短(1-2 句话)告诉子智能体你期望它做什么、关注哪些点
- 最终综合:结构化,明确各席位的核心结论、冲突点、最终建议
- 适度 emoji:📋 汇总、⚡ 派活、✅ 完成
## 禁止事项
- ❌ 亲自调研、写文档、写代码、联网搜索(这些是子智能体的活)
- ❌ 派活给本次清单**没有**的 agent_name
- ❌ 把括号里的中文显示名拼进 `agent_name`
- ❌ 已经达成共识后仍然继续派活
"""
# (id, name, description, soul, skills) — skills 为 None 表示不写 skills 字段。
_BUILTIN_FUNCTIONAL_AGENTS: tuple[tuple[str, str, str, str, list[str] | None], ...] = (
(INTENT_AGENT_ID, INTENT_AGENT_NAME, INTENT_AGENT_DESCRIPTION, INTENT_AGENT_SOUL, None),
(RECOMMENDER_AGENT_ID, RECOMMENDER_AGENT_NAME, RECOMMENDER_AGENT_DESCRIPTION, RECOMMENDER_AGENT_SOUL, None),
(
COORDINATOR_AGENT_ID,
COORDINATOR_AGENT_NAME,
COORDINATOR_AGENT_DESCRIPTION,
COORDINATOR_AGENT_SOUL,
["agent_orchestration"],
),
)
def _ensure_one(agent_id: str, name: str, description: str, soul: str) -> bool:
def _ensure_one(
agent_id: str,
name: str,
description: str,
soul: str,
skills: list[str] | None = None,
) -> bool:
"""Create the on-disk files for one functional agent if missing.
Returns True if files were just created, False if the directory already
existed (and we didn't touch it — respect any local edits).
``skills`` 写入 config.yaml 顶层 ``skills`` 字段,与 routers/agents.py 的
``_write_agent_config`` 输出形状对齐;coordinator 需要 ``agent_orchestration``
才能派活,intent / recommender 不需要任何 skill(传 None 即可)。
"""
agent_dir = get_paths().agent_dir(agent_id)
if agent_dir.exists():
@@ -411,8 +584,13 @@ def _ensure_one(agent_id: str, name: str, description: str, soul: str) -> bool:
# newline="\n" 强制 LF;否则 Windows 下 open() 把 \n 翻成 \r\n,跨 OS 部署时与
# Linux 部署机字节级不一致(功能上 YAML/Markdown 都兼容,但 git diff / 哈希对不齐)。
config_payload: dict[str, object] = {"id": agent_id, "name": name, "description": description}
if skills:
# 与 routers/agents.py::_write_agent_config 行为一致:None 不写、空列表也不写,
# 仅当确有 skill 名时才追加顶层 skills 字段。
config_payload["skills"] = list(skills)
config_yaml = yaml.dump(
{"id": agent_id, "name": name, "description": description},
config_payload,
default_flow_style=False,
allow_unicode=True,
)
@@ -435,9 +613,9 @@ def ensure_roundtable_functional_agents() -> list[str]:
actionable error if the agent really cannot be resolved.
"""
created: list[str] = []
for agent_id, name, description, soul in _BUILTIN_FUNCTIONAL_AGENTS:
for agent_id, name, description, soul, skills in _BUILTIN_FUNCTIONAL_AGENTS:
try:
if _ensure_one(agent_id, name, description, soul):
if _ensure_one(agent_id, name, description, soul, skills):
created.append(agent_id)
except Exception:
logger.exception("Failed to seed roundtable functional agent '%s'", agent_id)
@@ -2,23 +2,31 @@
from __future__ import annotations
import asyncio
import logging
import shutil
import time
from datetime import datetime
from typing import Any
from uuid import uuid4
import yaml
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from pydantic import BaseModel, Field
from app.gateway.deps import get_agent_store, get_tag_store
from app.gateway.auth.password import verify_password_async
from app.gateway.deps import get_agent_store, get_config, get_tag_store
from deerflow.agents.lead_agent.prompt import refresh_skills_system_prompt_cache_async
from deerflow.config.agents_api_config import get_agents_api_config
from deerflow.config.agents_config import list_custom_agents, load_agent_config, load_agent_soul, validate_agent_id
from deerflow.config.app_config import AppConfig
from deerflow.config.paths import get_paths
from deerflow.config.system_settings import load_system_settings
from deerflow.persistence.agents import AgentStore
from deerflow.persistence.tags import TagStore
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.skills.cascade import sync_agent_skills_to_available
from deerflow.skills.storage import get_or_new_skill_storage
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api", tags=["agents"])
@@ -32,6 +40,7 @@ class AgentResponse(BaseModel):
description: str = Field(default="", description="Agent description")
user_id: str | None = Field(default=None, description="Owner user id. Null means built-in.")
published: bool = Field(default=False, description="Whether a user-created agent is public")
is_private: bool = Field(default=False, description="Whether the agent lives in the password-gated private square instead of the public one")
builtin: bool = Field(default=False, description="True when user_id is null")
created_at: datetime | str | None = None
updated_at: datetime | str | None = None
@@ -50,6 +59,11 @@ class AgentResponse(BaseModel):
description="Admin-curated homepage rank. Null = not featured. Lower sorts first.",
)
is_favorite: bool = Field(
default=False,
description="Whether the requesting user has favorited (收藏) this agent.",
)
class AgentsListResponse(BaseModel):
"""Response model for listing all visible agents."""
@@ -68,6 +82,7 @@ class AgentCreateRequest(BaseModel):
description: str = Field(default="", description="Agent description")
soul: str = Field(default="", description="SOUL.md content - agent personality and behavioral guardrails")
published: bool = Field(default=False, description="Whether the new agent should be public")
is_private: bool = Field(default=False, description="When True, the new agent goes into the password-gated private square instead of the public one")
skills: list[str] | None = Field(default=None, description="Skill whitelist persisted to the agent's config.yaml")
model: str | None = Field(default=None, description="Optional model override persisted to config.yaml")
tool_groups: list[str] | None = Field(default=None, description="Optional tool group whitelist persisted to config.yaml")
@@ -80,6 +95,7 @@ class AgentUpdateRequest(BaseModel):
description: str | None = Field(default=None, description="Updated description")
soul: str | None = Field(default=None, description="Updated SOUL.md content")
published: bool | None = Field(default=None, description="Updated publication state")
is_private: bool | None = Field(default=None, description="Move the agent into / out of the private square (true / false)")
skills: list[str] | None = Field(default=None, description="Updated skill whitelist (config.yaml)")
model: str | None = Field(default=None, description="Updated model override (config.yaml)")
tool_groups: list[str] | None = Field(default=None, description="Updated tool group whitelist (config.yaml)")
@@ -192,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():
@@ -210,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.
@@ -247,20 +306,36 @@ 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"],
description=record.get("description") or "",
user_id=record.get("user_id"),
published=bool(record.get("published", False)),
is_private=bool(record.get("is_private", False)),
builtin=record.get("user_id") is None,
created_at=record.get("created_at"),
updated_at=record.get("updated_at"),
@@ -272,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:
@@ -291,17 +385,21 @@ async def _get_editable_or_raise(
) -> tuple[dict[str, Any], str]:
"""Return (record, edit_mode) where edit_mode is 'owner' or 'admin'.
'admin' is granted only when the agent is built-in (user_id IS NULL) and the
caller is a system admin. Custom agents owned by other users still 403.
Admins may edit any agent — built-in (``user_id IS NULL``) or owned by
another user — so the admin console can manage the full agent catalog.
Non-admin callers are still restricted to agents they own.
"""
owned = await store.get_owned(agent_id, user_id)
if owned is not None:
return owned, "owner"
if is_admin:
record = await store.get_any(agent_id)
if record is None:
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
return record, "admin"
visible = await store.get_visible(agent_id, user_id)
if visible is None:
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
if is_admin and visible.get("user_id") is None:
return visible, "admin"
raise HTTPException(status_code=403, detail="You can only modify agents that you created")
@@ -315,20 +413,47 @@ async def list_agents(
request: Request,
search: str | None = Query(default=None, description="Case-insensitive fuzzy match on agent name"),
tag_ids: list[str] | None = Query(default=None, description="Keep agents carrying any of these tags (OR)"),
admin: bool = Query(
default=False,
description=(
"Admin-only override. When ``true`` and the caller is a system admin, "
"every agent is returned (built-in + every user's agents + private-square "
"entries) regardless of the usual ownership/visibility filter."
),
),
) -> AgentsListResponse:
_require_agents_api_enabled()
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)
agents = await store.list_visible(user_id)
responses = [_record_to_response(a, include_soul=True) for a in agents]
if admin:
if not _is_admin_user(request):
raise HTTPException(status_code=403, detail="Only system admins can list all agents")
agents = await store.list_all()
else:
agents = await store.list_visible(user_id)
# Load lightweight extras (skills / model / tool_groups from config.yaml)
# so the gallery cards can render skill badges. SOUL.md is still skipped
# (only the edit dialog reads it). The per-agent config.yaml reads run in
# worker threads concurrently, so the event loop stays responsive even
# with many agents.
responses = list(
await asyncio.gather(
*(
_record_to_response_async(a, include_soul=False, include_extras=True)
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)
for resp in responses:
resp.tags = assignments.get(resp.id, [])
resp.is_favorite = resp.id in favorite_ids
if search:
needle = search.strip().lower()
responses = [r for r in responses if needle in r.name.lower()]
@@ -368,11 +493,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:
@@ -392,12 +524,18 @@ 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")
return _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
except HTTPException:
raise
except Exception as e:
@@ -446,10 +584,11 @@ async def create_agent_endpoint(request: AgentCreateRequest, http_request: Reque
"description": request.description or "",
"user_id": _current_user_id(http_request),
"published": request.published,
"is_private": request.is_private,
}
)
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
@@ -486,6 +625,15 @@ async def update_agent(agent_id: str, request: AgentUpdateRequest, http_request:
update_data["description"] = request.description or ""
if "published" in fields_set:
update_data["published"] = bool(request.published)
if "is_private" in fields_set:
update_data["is_private"] = bool(request.is_private)
# Mutual exclusion: an agent moved into the private square cannot
# also be publicly published. Force ``published=false`` whenever the
# same request sets ``is_private=true``, even if the client forgot
# to clear it. (When the request only sets one of the two, the
# other side is left untouched.)
if update_data.get("is_private") is True:
update_data["published"] = False
next_name = update_data.get("name", current["name"])
next_description = update_data.get("description", current.get("description") or "")
@@ -511,14 +659,17 @@ async def update_agent(agent_id: str, request: AgentUpdateRequest, http_request:
if not update_data:
record = current
elif edit_mode == "admin":
record = await store.update_builtin(agent_id, update_data)
# Admins may edit built-in (``user_id IS NULL``) and any user-owned
# agent — ``update_any`` skips the ownership check that ``update``
# and ``update_builtin`` enforce.
record = await store.update_any(agent_id, update_data)
else:
record = await store.update(agent_id, user_id, update_data)
if record is None:
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
@@ -598,7 +749,7 @@ async def delete_agent(agent_id: str, request: Request) -> None:
)
if edit_mode == "admin":
deleted = await store.delete_builtin(agent_id)
deleted = await store.delete_any(agent_id)
else:
deleted = await store.delete(agent_id, user_id)
if not deleted:
@@ -611,6 +762,10 @@ async def delete_agent(agent_id: str, request: Request) -> None:
await get_tag_store(request).unassign_all("agent", agent_id)
except Exception: # noqa: BLE001 — tag cleanup must not block deletion
pass
try:
await store.delete_favorites_for_agent(agent_id)
except Exception: # noqa: BLE001 — favorite cleanup must not block deletion
pass
logger.info("Deleted agent '%s' from %s", agent_id, agent_dir)
except HTTPException:
raise
@@ -619,6 +774,53 @@ async def delete_agent(agent_id: str, request: Request) -> None:
raise HTTPException(status_code=500, detail=f"Failed to delete agent: {str(e)}") from e
@router.post(
"/agents/{agent_id}/favorite",
status_code=204,
summary="Favorite Agent",
description="Bookmark a visible agent into the current user's 我的 tab. Idempotent.",
)
async def favorite_agent(agent_id: str, request: Request) -> None:
_require_agents_api_enabled()
agent_id = _validate_storage_id(agent_id)
try:
store = get_agent_store(request)
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")
await store.add_favorite(user_id, agent_id)
logger.info("User '%s' favorited agent '%s'", user_id, agent_id)
except HTTPException:
raise
except Exception as e:
logger.error("Failed to favorite agent '%s': %s", agent_id, e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to favorite agent: {str(e)}") from e
@router.delete(
"/agents/{agent_id}/favorite",
status_code=204,
summary="Unfavorite Agent",
description="Remove an agent from the current user's favorites. No-op when not favorited.",
)
async def unfavorite_agent(agent_id: str, request: Request) -> None:
_require_agents_api_enabled()
agent_id = _validate_storage_id(agent_id)
try:
store = get_agent_store(request)
user_id = _current_user_id(request)
await store.remove_favorite(user_id, agent_id)
logger.info("User '%s' unfavorited agent '%s'", user_id, agent_id)
except HTTPException:
raise
except Exception as e:
logger.error("Failed to unfavorite agent '%s': %s", agent_id, e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to unfavorite agent: {str(e)}") from e
@router.put(
"/agents/{agent_id}/featured",
response_model=AgentResponse,
@@ -642,7 +844,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")
@@ -654,7 +856,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:
@@ -662,6 +864,107 @@ async def set_agent_featured(
raise HTTPException(status_code=500, detail=f"Failed to set featured: {str(e)}") from e
class PrivateSquareListRequest(BaseModel):
"""Request body for listing all private-square agents."""
password: str = Field(..., description="Plain-text private-square password. Verified against the admin-configured hash before the listing is returned.")
class AgentPrivateUpdateRequest(BaseModel):
"""Admin-only request body for moving an agent into / out of the private square."""
is_private: bool = Field(..., description="Set true to move the agent into the private square, false to move it back into the public square.")
@router.post(
"/agents/private-square/list",
response_model=AgentsListResponse,
summary="List Private-Square Agents",
description=(
"Verify the admin-configured private-square password and return every "
"agent flagged ``is_private=true``. Returns 404 when no password is set, "
"403 when the supplied password does not match, and the full agent list "
"otherwise — regardless of ownership, since the password is the gate."
),
)
async def list_private_square(body: PrivateSquareListRequest, request: Request) -> AgentsListResponse:
_require_agents_api_enabled()
settings = load_system_settings().private_square
if not settings.password_hash:
raise HTTPException(status_code=404, detail="Private square password is not configured")
if not body.password or not await verify_password_async(body.password, settings.password_hash):
raise HTTPException(status_code=403, detail="Incorrect private square password")
try:
store = get_agent_store(request)
tag_store: TagStore = get_tag_store(request)
await _sync_legacy_agents(store)
records = await store.list_private()
responses = [_record_to_response(r, include_soul=True) for r in records]
assignments = await tag_store.list_assignments("agent", [r.id for r in responses])
for resp in responses:
resp.tags = assignments.get(resp.id, [])
return AgentsListResponse(agents=responses)
except HTTPException:
raise
except Exception as e:
logger.error("Failed to list private agents: %s", e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to list private agents: {str(e)}") from e
@router.put(
"/agents/{agent_id}/private",
response_model=AgentResponse,
summary="Move Agent Between Squares (admin)",
description=(
"Admin-only. Move any agent (built-in or user-owned) into or out of the "
"password-gated private square. Owners can already toggle the flag on "
"their own agents via the regular ``PUT /agents/{id}`` endpoint; this "
"endpoint exists so admins can re-categorize other users' agents too."
),
)
async def set_agent_private(
agent_id: str,
body: AgentPrivateUpdateRequest,
request: Request,
) -> AgentResponse:
_require_agents_api_enabled()
agent_id = _validate_storage_id(agent_id)
if not _is_admin_user(request):
raise HTTPException(status_code=403, detail="Only system admins can re-categorize other users' agents")
try:
store = get_agent_store(request)
await _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")
update: dict[str, Any] = {"is_private": bool(body.is_private)}
# Mutual exclusion — see ``update_agent`` above. Moving an agent into
# the private square also drops the public ``published`` flag so we
# never end up with an agent that's "in both squares".
if update["is_private"]:
update["published"] = False
if record.get("user_id") is None:
updated = await store.update_builtin(agent_id, update)
else:
updated = await store.update(agent_id, str(record["user_id"]), update)
if updated is None:
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
logger.info("Admin set agent '%s' is_private=%s", agent_id, body.is_private)
return _record_to_response(updated, include_soul=True)
except HTTPException:
raise
except Exception as e:
logger.error("Failed to set is_private for agent '%s': %s", agent_id, e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to update agent: {str(e)}") from e
class AgentFeaturedReorderRequest(BaseModel):
"""Request body for admin to bulk-reorder the homepage featured list."""
@@ -705,3 +1008,78 @@ async def reorder_featured_agents(
except Exception as e:
logger.error("Failed to reorder featured agents: %s", e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to reorder featured agents: {str(e)}") from e
class AgentSkillRemovalEntry(BaseModel):
agent_id: str = Field(..., description="Agent storage id whose config.yaml was rewritten")
agent_name: str | None = Field(default=None, description="Display name from the DB, if any")
user_id: str | None = Field(default=None, description="Owner user id (null = built-in)")
removed_skills: list[str] = Field(..., description="Skill names that no longer exist on disk and were stripped")
class AgentSkillsSyncResponse(BaseModel):
checked: int = Field(..., description="Number of agent config.yaml files scanned")
modified: int = Field(..., description="Number of agents whose skills list was rewritten")
removals: list[AgentSkillRemovalEntry] = Field(default_factory=list, description="Per-agent removal details")
@router.post(
"/agents/admin/sync-skills",
response_model=AgentSkillsSyncResponse,
summary="Sync Agent Skills (admin)",
description=(
"Admin-only. Scan every agent's ``config.yaml`` and strip skill names that "
"no longer exist on disk. Repairs historical drift where a skill was "
"removed without the delete/unpublish cascade firing (e.g. deleted from "
"the filesystem manually, or removed in an older version of the code)."
),
)
async def admin_sync_agent_skills(
request: Request,
config: AppConfig = Depends(get_config),
) -> AgentSkillsSyncResponse:
_require_agents_api_enabled()
if not _is_admin_user(request):
raise HTTPException(status_code=403, detail="Only system admins can run agent-skills sync")
try:
storage = get_or_new_skill_storage(app_config=config)
all_skills = storage.load_skills(enabled_only=False)
available = {skill.name for skill in all_skills}
result = await asyncio.to_thread(sync_agent_skills_to_available, available)
# Enrich removals with display name + owner from the DB so the admin
# UI can show "我的智能体 X 移除了技能 Y" instead of just the id.
store = get_agent_store(request)
enriched: list[AgentSkillRemovalEntry] = []
for entry in result.get("removals", []):
agent_id = entry["agent_id"]
record = await store.get_any(agent_id)
enriched.append(
AgentSkillRemovalEntry(
agent_id=agent_id,
agent_name=(record.get("name") if record else None),
user_id=(record.get("user_id") if record else None),
removed_skills=entry["removed_skills"],
)
)
if result.get("modified"):
await refresh_skills_system_prompt_cache_async()
logger.info(
"Admin agent-skills sync: checked=%d modified=%d",
result["checked"],
result["modified"],
)
return AgentSkillsSyncResponse(
checked=result["checked"],
modified=result["modified"],
removals=enriched,
)
except HTTPException:
raise
except Exception as e:
logger.error("Failed to sync agent skills: %s", e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to sync agent skills: {str(e)}") from 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,
@@ -93,6 +94,173 @@ _INTENT_READY_PATTERN = re.compile(
)
def _normalize_question(text: str) -> str:
"""归一化澄清问题文本,用于"和过往是否重复"的判断。
LLM 复述时常见模式:
- 完全一样的字符串
- 仅前后空白、标点、emoji 不同(如多/少一个问号)
- 仅大小写差异(中文场景几乎不见,但兼容英文/拼音)
所以归一化时统一:小写 + 去除非字母/数字/中文字符 + 折叠空白。
Levenshtein/语义相似度等模糊匹配先不做,等观察到误判再迭代。
"""
if not isinstance(text, str):
return ""
lowered = text.strip().lower()
# 只保留中英文字符与数字,去掉 ?。!?,。"":/、() 等标点 + emoji + 空白。
return re.sub(r"[^0-9a-z一-鿿]+", "", lowered)
def _normalize_options(raw: object) -> tuple[str, ...]:
"""把 options 拍平成可比较的归一化元组,用于"选项是否和上一轮一致"的判断。"""
if raw is None:
return ()
if isinstance(raw, str):
# 偶有模型把 list 序列化成 JSON 字符串发回来。
try:
raw = json.loads(raw)
except (json.JSONDecodeError, TypeError):
return (_normalize_question(raw),)
if not isinstance(raw, list):
return (_normalize_question(str(raw)),)
out: list[str] = []
for item in raw:
if isinstance(item, dict):
label = item.get("label") or item.get("title") or item.get("name") or item.get("value") or item.get("id") or ""
out.append(_normalize_question(str(label)))
else:
out.append(_normalize_question(str(item)))
# 顺序无关:LLM 偶尔会调换选项顺序,但语义上还是同一组。
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]],
*,
new_question: str,
new_options: object,
new_tool_call_id: str | None = None,
) -> bool:
"""检测本轮的 ask_clarification 是否在过往轮次问过。
判定为重复必须同时满足:
1. 历史里至少有一次 ``ask_clarification`` 的问题文本 / 选项 与本次归一化相同;
2. 用户在那次之后**已经发过消息**(即"已回答过")。
没有这条限制会把"模型在同一 run 内多次输出同一 tool_call_chunks"
误判成重复,导致首次澄清都吞掉。
3. 不是同一个 tool_call(LangGraph 偶有把同一 call 拆成多块返回的情况)。
传回 True → 上层应把 status 从 ``clarification`` 降级为 ``asking``,
避免前端弹同一张卡片(配合 SOUL §2.5 prompt 双层防御)。
"""
norm_new_q = _normalize_question(new_question)
norm_new_opts = _normalize_options(new_options)
if not norm_new_q and not norm_new_opts:
return False
seen_match_index: int | None = None
for idx, msg in enumerate(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":
continue
if new_tool_call_id and tc.get("id") == new_tool_call_id:
# 同一个 tool_call 的不同 chunk,不算重复。
continue
args = tc.get("args") or {}
past_q = _normalize_question(str(args.get("question") or ""))
past_opts = _normalize_options(args.get("options"))
# 问题文本和选项任一明确重叠就视为重复:
# - question 相同(options 可能空)
# - options 相同且非空(question 可能被改写)
q_match = bool(past_q) and past_q == norm_new_q
opt_match = bool(past_opts) and past_opts == norm_new_opts
if q_match or opt_match:
seen_match_index = idx
break
if seen_match_index is not None:
break
if seen_match_index is None:
return False
# 校验用户在该次澄清之后是否已经发过 human 消息(= 回答过)。
for msg in messages[seen_match_index + 1:]:
if not isinstance(msg, dict):
continue
msg_type = (msg.get("type") or "").lower()
if msg_type in ("human", "humanmessage", "user"):
return True
return False
@router.post("/init")
async def init_intent(request: Request, payload: dict = Body(...)) -> dict[str, Any]:
"""创建任务理解会话用的空 thread。
@@ -119,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
@@ -176,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=[],
)
@@ -235,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 = {
@@ -264,6 +475,34 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
# 比 tool_call 旁的短 preamble 更适合作为气泡正文。
follow_up = _find_trailing_ai_text(messages, exclude_text=content)
display_content = follow_up or content or question
# 兜底:模型偶尔会复述上一轮的同一澄清问题(SOUL §2.5 prompt
# 已经强约束,但小模型 / 长上下文仍可能漏跟随)。检测到完全
# 重复时,把 status 从 clarification 降级为 asking,前端只渲染
# 文本气泡不弹卡片,避免用户看到"同样选项又出现一次"。
if _is_repeated_clarification(
messages,
new_question=question,
new_options=cargs.get("options"),
new_tool_call_id=clarification_call.get("id"),
):
logger.warning(
"intent stream: dropping duplicate ask_clarification "
"(question=%r) — already asked & answered earlier in thread",
question[:80],
)
fallback_msg = (
display_content
or "(系统识别到智能体重复了上一问题,已自动跳过。请补充其他信息继续推进。)"
)
final_frame: dict[str, Any] = {
"status": "asking",
"content": fallback_msg,
"summary": None,
}
yield f"data: {json.dumps(final_frame, ensure_ascii=False)}\n\n"
return
final_frame: dict[str, Any] = {
"status": "clarification",
"content": display_content,
@@ -182,10 +182,13 @@ async def update_mcp_configuration(request: McpConfigUpdateRequest, http_request
current_config = get_extensions_config()
# Convert request to dict format for JSON serialization
config_data = {
"mcpServers": {name: server.model_dump() for name, server in request.mcp_servers.items()},
"skills": {name: {"enabled": skill.enabled} for name, skill in current_config.skills.items()},
}
config_data = {
"mcpServers": {name: server.model_dump(by_alias=True) for name, server in request.mcp_servers.items()},
"skills": {
name: skill.model_dump(by_alias=True, exclude_none=True)
for name, skill in current_config.skills.items()
},
}
# Write the configuration to file
with open(config_path, "w", encoding="utf-8") as f:
@@ -53,18 +53,32 @@ import os
import re
import time
import traceback
import uuid
from typing import Any, AsyncIterator
import httpx
from app.gateway.routers._roundtable_seed import (
COORDINATOR_AGENT_ID,
ensure_roundtable_functional_agents,
)
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 +299,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 +350,110 @@ 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
# NOTE: 历史上这里曾有 _create_coordinator_direct(),每次 init 都创建一个带时间戳的
# roundtable-coordinator-<ts36>。2026-05 改造为单例(见 _roundtable_seed.py 与本
# 模块 init_multi_agent 的 docstring),不再需要"每次创建"的直调辅助;改造后由
# ensure_roundtable_functional_agents() 在路由入口落地静态 SOUL,席位列表通过
# _append_thread_message 注入到 coordinator thread 的对话历史中。
async def _append_thread_message(
client: httpx.AsyncClient,
base: str,
@@ -456,6 +587,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 +596,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 +616,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,
@@ -508,137 +651,165 @@ async def init_multi_agent(request: Request, payload: dict = Body(...)) -> Strea
{
"agent_names": ["roundtable-intelligence", ...], // 用户选的席位 id
"main_agent_name": "roundtable-coordinator-xxx", // 当次总控 id,前端生成
"model": "<可选>" // 可选,写入 coordinator agent;不传由 lead_agent 回退到 config.yaml models[0]
"main_agent_name": "<可选,已废弃>", // 历史兼容字段,新代码忽略,统一用 COORDINATOR_AGENT_ID
"model": "<可选>" // 可选;不传由 lead_agent 回退到 config.yaml models[0]
}
SSE 事件(每条 ``data: {...}\\n\\n``)::
seat_ready — 某一席位 thread 已创建(``asyncio.as_completed`` 顺序,先完先发)
coordinator_ready — 总控 agent + thread 已就绪
coordinator_ready — 总控 thread 就绪 + 已注入本次席位列表
init_done — ``thread_ids`` 全量映射,前端存 ``threadIds`` 并开始编排
error — 失败详情(同步校验失败则直接 HTTP 4xx,不进 SSE)
协调智能体 SOUL 在运行时根据各席位 ``GET /api/agents/{id}`` 的 name/description 拼接,
强调 ``agent_name`` 必须用英文 id(``roundtable-intelligence``),禁止带中文括号后缀。
协调智能体改造说明
----------------
协调智能体现在是 **单例**(id 固定为 ``roundtable-coordinator``,SOUL 见
``_roundtable_seed.COORDINATOR_AGENT_SOUL``),由 ``ensure_roundtable_functional_agents()``
在首次调用时自动落地。每次 init **不再创建新的 coordinator agent**(避免
``.deer-flow/agents/`` 累积 ``roundtable-coordinator-<ts36>`` 残留)。
本次圆桌的「可调度席位列表」由 init 在创建好 coordinator thread 之后,以一条
human 消息追加到 thread 历史里,leader run 每轮都能从对话历史里读到。这样
SOUL 长期稳定,席位列表随会话动态注入。前端为兼容旧 client 仍可传
``main_agent_name`` 字段,但后端会忽略并使用 ``COORDINATOR_AGENT_ID``。
"""
# 调用前自检:与 intent/recommend 一样,缺失时就地补建,避免 /run/stream 报 500。
ensure_roundtable_functional_agents()
agent_names_in = payload.get("agent_names") or []
agent_names: list[str] = [str(name).lower() for name in agent_names_in]
main_agent_name: str = str(payload.get("main_agent_name") or "main-agent").lower()
model: str | None = payload.get("model")
# main_agent_name 是历史字段,2026-05 改造后忽略,统一用单例 id。保留接收但不再用前端值,
# 避免下游再次创建带时间戳的临时 coordinator。
main_agent_name = COORDINATOR_AGENT_ID
if not agent_names:
raise HTTPException(status_code=400, detail="agent_names is required")
for name in agent_names:
_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:
# 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()
# 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 ""),
)
tasks = [asyncio.create_task(_prepare_sub_agent(sid)) for sid in agent_names]
prepared: list[tuple[str, str, str, str]] = []
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
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()
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}
seat_lines: list[str] = [
f"- agent_name: {sid}({display_name}):{description}"
for sid, _, display_name, description in prepared
]
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
]
# 协调智能体改造为**单例**(id = roundtable-coordinator):
# - SOUL 已经由 ensure_roundtable_functional_agents() 在路由入口落地,
# 通用协调规则常驻磁盘,无需每次重写。
# - 本次"可调度席位清单"以 human 消息形式 append 到 coordinator
# thread 历史里,leader run 每轮都能在对话历史最前面看到。
# 这样席位列表随会话动态变化,但磁盘上不再累积 roundtable-coordinator-<ts36>。
#
# 注意:model 字段在 lead_agent run 时通过 context.model_name 注入,
# 不需要写回到 agent config.yaml(否则会污染单例的配置)。
seat_introduction = (
"【圆桌系统初始化】本次研讨可调度的席位清单如下,请严格按 agent_name "
"通过 agent_orchestration 技能派活,不要派给不在清单内的 agent:\n"
+ "\n".join(seat_lines)
)
# 动态 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)
)
coord_thread = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
agent_threads[main_agent_name] = coord_thread
coord_thread = await _create_thread(client, base, headers)
agent_threads[main_agent_name] = coord_thread
ok, result = await _create_agent(
client,
base,
headers,
name=main_agent_name,
description="你是总控控制协调智能体,负责协调其他智能体干活",
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
# 把席位列表注入 coordinator thread 的对话历史。HTTP loopback 调用
# 走 /api/threads/{tid}/state(读+写),开销 ~10-30ms,在 init 整体
# 几百 ms 的尺度上微不足道,但比起手撕 langgraph checkpoint 内部结构
# 风险低、与 _broadcast / _append_thread_message 一套语义,长期可维护。
try:
base = _loopback_base(request)
headers = _auth_headers(request)
async with _make_loopback_client(request) as seed_client:
await _append_thread_message(
seed_client,
base,
headers,
coord_thread,
seat_introduction,
)
except Exception:
# 注入失败不阻断 init:leader run 仍能跑,只是看不到席位清单,
# 可能派活给错的 id。会被 _strip_display_suffix / _validate_agent_id
# 兜底,前端能看到错误帧;运维通过 backend 日志找到根因再修复。
logger.exception("[init-direct] failed to seed coordinator thread with seat list")
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}"})
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}"})
return StreamingResponse(
stream_generator(),
@@ -764,25 +935,30 @@ def _find_trailing_ai_text(messages: list[dict[str, Any]], *, exclude_text: str
def _find_dispatch_message(messages: list[dict[str, Any]]) -> dict[str, Any]:
"""Locate the latest AIMessage that carries dispatch tool calls.
"""Locate the latest AIMessage from THIS turn.
The very last message can be a skill-stop ToolMessage injected by
``SkillStopMiddleware`` when it intercepts the dispatch, or a clarification
ToolMessage. Either way, the actionable tool_calls live on the AIMessage
*before* it, so we walk backwards until we hit one with ``tool_calls``.
Returns the latest AIMessage when none of them carry tool_calls so the
caller can still extract its text content.
Trailing ToolMessages (skill_stop / clarification ack injected by
``SkillStopMiddleware`` or ``ClarificationMiddleware``) are skipped because
the actionable tool_calls live on the AIMessage that emitted them. We stop
at the **first AIMessage encountered** — its ``tool_calls`` (or lack
thereof) is the authoritative signal for this turn.
历史踩坑:之前实现是"一直往前翻直到撞到带 tool_calls 的 AI",看似优雅,
实际上让 leader 永远收敛不了:本轮 LLM 不论怎么共识,只要历史上派过活,
上一轮的派活 AI 就会被翻出来再用一次,前端会把同样的 task 又发给子智能体,
陷入死循环。现在停在最新一条 AI 上,把判定交给调用方:`tool_calls=[]` →
共识分支,有 tool_calls → 派活分支。
"""
last_ai: dict[str, Any] = {}
for msg in reversed(messages):
if not isinstance(msg, dict):
continue
if (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
if not last_ai:
last_ai = msg
if msg.get("tool_calls"):
return msg
return last_ai
msg_type = (msg.get("type") or "").lower()
if msg_type in ("tool", "toolmessage"):
continue # 跳过末尾 skill_stop / clarification ack
if msg_type in ("ai", "aimessage", "aimessagechunk"):
return msg
break # 撞到 Human/其他类型,本轮没 AI(异常路径,留空让上游报错)
return {}
async def _leader_run(
@@ -820,6 +996,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] = []
@@ -37,6 +37,10 @@ class ScheduledTaskCreateRequest(BaseModel):
timezone: str | None = None
execution_agent_name: str | None = None
execution_context: dict[str, Any] | None = None
require_markdown: bool = Field(
default=False,
description="Require the result to be Markdown and exceed the minimum length, else mark the run failed.",
)
class ScheduledTaskUpdateRequest(BaseModel):
@@ -48,6 +52,7 @@ class ScheduledTaskUpdateRequest(BaseModel):
enabled: bool | None = None
execution_agent_name: str | None = None
execution_context: dict[str, Any] | None = None
require_markdown: bool | None = None
class ScheduledTaskDeliveryProfileUpdateRequest(BaseModel):
@@ -555,6 +560,7 @@ async def create_scheduled_task(body: ScheduledTaskCreateRequest, request: Reque
timezone=body.timezone,
execution_agent_name=body.execution_agent_name,
execution_context=body.execution_context,
require_markdown=body.require_markdown,
)
except DuplicateTaskNameError:
raise HTTPException(status_code=409, detail="已存在同名定时任务,请换一个名称")
File diff suppressed because it is too large. Load diff
@@ -3,10 +3,12 @@
from __future__ import annotations
import logging
from typing import Literal
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field
from app.gateway.auth.password import hash_password_async
from app.gateway.deps import get_optional_user_from_request
from deerflow.config.system_settings import (
PromptPrefixSettings,
@@ -31,6 +33,30 @@ class PromptPrefixUpdateRequest(BaseModel):
prompt_prefix: str | None = None
class PrivateSquareStatusResponse(BaseModel):
"""Status of the private-square feature, readable by any logged-in user.
The password hash itself is **never** returned — the response only signals
whether the feature is enabled and whether a password has been set.
"""
enabled: bool = Field(..., description="Whether the private square is exposed to users")
password_set: bool = Field(..., description="Whether a password is currently configured")
class PrivateSquareUpdateRequest(BaseModel):
enabled: bool | None = Field(default=None, description="Toggle the private-square feature")
password: str | None = Field(default=None, description="New password in plain text. Pass an empty string to clear.")
class CitationDisplayResponse(BaseModel):
cleanup_words: list[str] = Field(default_factory=list, description="Global citation cleanup words")
class CitationDisplayUpdateRequest(BaseModel):
cleanup_words: list[str] | None = None
def _to_response(settings: PromptPrefixSettings) -> PromptPrefixResponse:
return PromptPrefixResponse(
enabled=settings.enabled,
@@ -61,5 +87,116 @@ async def update_prompt_prefix(body: PromptPrefixUpdateRequest, request: Request
current = load_system_settings()
fields = body.model_dump(exclude_unset=True)
updated_prefix = current.prompt_prefix.model_copy(update=fields)
saved = save_system_settings(SystemSettings(prompt_prefix=updated_prefix))
saved = save_system_settings(current.model_copy(update={"prompt_prefix": updated_prefix}))
return _to_response(saved.prompt_prefix)
@router.get("/citation-display", response_model=CitationDisplayResponse)
async def get_citation_display() -> CitationDisplayResponse:
"""Return global citation display cleanup settings."""
settings = load_system_settings().citation_display
return CitationDisplayResponse(cleanup_words=settings.cleanup_words)
@router.put("/citation-display", response_model=CitationDisplayResponse)
async def update_citation_display(
body: CitationDisplayUpdateRequest,
request: Request,
) -> CitationDisplayResponse:
"""Update global citation display cleanup settings. Admin only."""
await _require_admin(request)
current = load_system_settings()
words = [
str(item).strip()
for item in (body.cleanup_words or [])
if str(item).strip()
]
updated = current.citation_display.model_copy(update={"cleanup_words": words})
saved = save_system_settings(
current.model_copy(update={"citation_display": updated})
)
return CitationDisplayResponse(cleanup_words=saved.citation_display.cleanup_words)
@router.get("/private-square", response_model=PrivateSquareStatusResponse)
async def get_private_square_status() -> PrivateSquareStatusResponse:
"""Return whether the private square is enabled and a password is set."""
settings = load_system_settings().private_square
return PrivateSquareStatusResponse(
enabled=settings.enabled,
password_set=bool(settings.password_hash),
)
@router.put("/private-square", response_model=PrivateSquareStatusResponse)
async def update_private_square(body: PrivateSquareUpdateRequest, request: Request) -> PrivateSquareStatusResponse:
"""Update the private-square configuration. Admin only.
Setting ``password`` to a non-empty string rehashes and stores it.
Passing an empty string clears the password (effectively locking the
private square so no one can read it).
"""
await _require_admin(request)
current = load_system_settings()
updated = current.private_square
fields = body.model_dump(exclude_unset=True)
update: dict[str, object] = {}
if "enabled" in fields and fields["enabled"] is not None:
update["enabled"] = bool(fields["enabled"])
if "password" in fields and fields["password"] is not None:
plain = str(fields["password"])
if plain == "":
update["password_hash"] = ""
else:
update["password_hash"] = await hash_password_async(plain)
if update:
updated = updated.model_copy(update=update)
saved = save_system_settings(current.model_copy(update={"private_square": updated}))
else:
saved = current
return PrivateSquareStatusResponse(
enabled=saved.private_square.enabled,
password_set=bool(saved.private_square.password_hash),
)
class AppearanceResponse(BaseModel):
"""Effective system-wide appearance defaults."""
default_theme: Literal["light", "dark"] = Field(..., description="System default theme — light or dark")
class AppearanceUpdateRequest(BaseModel):
default_theme: Literal["light", "dark"] | None = Field(default=None)
@router.get("/appearance", response_model=AppearanceResponse)
async def get_appearance_settings() -> AppearanceResponse:
"""Return the global default theme. Any logged-in user may read this."""
settings = load_system_settings().appearance
return AppearanceResponse(default_theme=settings.default_theme)
@router.put("/appearance", response_model=AppearanceResponse)
async def update_appearance_settings(
body: AppearanceUpdateRequest, request: Request
) -> AppearanceResponse:
"""Update the global appearance defaults. Admin only."""
await _require_admin(request)
current = load_system_settings()
fields = body.model_dump(exclude_unset=True)
update: dict[str, object] = {}
if "default_theme" in fields and fields["default_theme"] is not None:
update["default_theme"] = fields["default_theme"]
if update:
updated_appearance = current.appearance.model_copy(update=update)
saved = save_system_settings(
current.model_copy(update={"appearance": updated_appearance})
)
else:
saved = current
return AppearanceResponse(default_theme=saved.appearance.default_theme)
@@ -12,24 +12,38 @@ matching the LangGraph Platform wire format expected by the
from __future__ import annotations
import logging
import uuid
from typing import Any
import logging
import json
import os
from pathlib import Path
import uuid
from typing import Any
from fastapi import APIRouter, HTTPException, Request
from langgraph.checkpoint.base import empty_checkpoint
from pydantic import BaseModel, Field, field_validator
from fastapi import APIRouter, HTTPException, Request
from langgraph.checkpoint.base import empty_checkpoint
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import select
from app.gateway.authz import require_permission
from app.gateway.deps import get_checkpointer
from app.gateway.utils import sanitize_log_param
from deerflow.config.extensions_config import SkillDisplayConfig
from deerflow.config.paths import Paths, get_paths
from deerflow.persistence.engine import get_engine, get_session_factory
from deerflow.persistence.skills.model import SkillDisplayAdapterRow
from deerflow.runtime import serialize_channel_values
from deerflow.runtime.references import (
adapt_reference_batches_with_displays,
build_reference_batches as build_runtime_reference_batches,
build_reference_debug_payload,
build_reference_debug_summary,
)
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.utils.time import coerce_iso, now_iso
from app.gateway.authz import require_permission
from app.gateway.deps import get_checkpointer
from app.gateway.utils import sanitize_log_param
from deerflow.config.paths import Paths, get_paths
from deerflow.runtime import serialize_channel_values
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.utils.time import coerce_iso, now_iso
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/threads", tags=["threads"])
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/threads", tags=["threads"])
_RAG_DEBUG_LOGGER_NAME = "rag_citation_debug"
# Metadata keys that the server controls; clients are not allowed to set
@@ -41,16 +55,86 @@ router = APIRouter(prefix="/api/threads", tags=["threads"])
_SERVER_RESERVED_METADATA_KEYS: frozenset[str] = frozenset({"owner_id", "user_id"})
def _strip_reserved_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]:
"""Return ``metadata`` with server-controlled keys removed."""
if not metadata:
return metadata or {}
return {k: v for k, v in metadata.items() if k not in _SERVER_RESERVED_METADATA_KEYS}
# ---------------------------------------------------------------------------
# Response / request models
# ---------------------------------------------------------------------------
def _strip_reserved_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]:
"""Return ``metadata`` with server-controlled keys removed."""
if not metadata:
return metadata or {}
return {k: v for k, v in metadata.items() if k not in _SERVER_RESERVED_METADATA_KEYS}
def _get_rag_debug_logger() -> logging.Logger:
debug_logger = logging.getLogger(_RAG_DEBUG_LOGGER_NAME)
if getattr(debug_logger, "_rag_file_configured", False):
return debug_logger
log_path = Path(os.environ.get("RAG_CITATION_DEBUG_LOG", "logs/rag_citation_debug.log"))
if not log_path.is_absolute():
log_path = Path.cwd() / log_path
log_path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.FileHandler(log_path, encoding="utf-8")
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
debug_logger.addHandler(handler)
debug_logger.setLevel(logging.INFO)
debug_logger.propagate = False
setattr(debug_logger, "_rag_file_configured", True)
setattr(debug_logger, "_rag_file_path", str(log_path))
return debug_logger
def _rag_debug_file_path() -> str:
return str(getattr(_get_rag_debug_logger(), "_rag_file_path", "logs/rag_citation_debug.log"))
def _log_rag_debug(message: str, *args: Any) -> None:
logger.info(message, *args)
_get_rag_debug_logger().info(message, *args)
def _json_text_loads(value: str | None) -> dict[str, Any] | None:
if not value:
return None
try:
parsed = json.loads(value)
return parsed if isinstance(parsed, dict) else None
except Exception:
logger.debug("Failed to parse persisted skill display JSON", exc_info=True)
return None
async def _load_persisted_skill_displays() -> dict[str, SkillDisplayConfig]:
engine = get_engine()
sf = get_session_factory()
if engine is None or sf is None:
return {}
try:
async with engine.begin() as conn:
await conn.run_sync(SkillDisplayAdapterRow.__table__.create, checkfirst=True)
async with sf() as session:
result = await session.execute(select(SkillDisplayAdapterRow))
displays: dict[str, SkillDisplayConfig] = {}
for row in result.scalars():
payload = _json_text_loads(row.display_config)
if payload is None:
continue
try:
display = SkillDisplayConfig.model_validate(payload)
except Exception:
logger.debug(
"Failed to validate persisted skill display for %s",
row.skill_name,
exc_info=True,
)
continue
if display.enabled and display.mode != "none":
displays[row.skill_name] = display
return displays
except Exception:
logger.debug("Failed to load persisted skill display adapters", exc_info=True)
return {}
# ---------------------------------------------------------------------------
# Response / request models
# ---------------------------------------------------------------------------
class ThreadDeleteResponse(BaseModel):
@@ -132,11 +216,45 @@ class HistoryEntry(BaseModel):
next: list[str] = Field(default_factory=list)
class ThreadHistoryRequest(BaseModel):
class ThreadHistoryRequest(BaseModel):
"""Request body for checkpoint history."""
limit: int = Field(default=10, ge=1, le=100, description="Maximum entries")
before: str | None = Field(default=None, description="Cursor for pagination")
before: str | None = Field(default=None, description="Cursor for pagination")
class ReferenceSourceResponse(BaseModel):
index: int
id: str | None = None
skillName: str = "backend-reference"
mode: str = "citation"
title: str
type: str = "result"
snippet: str = ""
source: str | None = None
author: str | None = None
time: str | None = None
url: str | None = None
content: str | None = None
score: float | None = None
tableColumns: list[dict[str, Any]] = Field(default_factory=list)
raw: dict[str, Any] = Field(default_factory=dict)
class ReferenceBatchResponse(BaseModel):
id: str
assistant_message_id: str
sources: list[ReferenceSourceResponse] = Field(default_factory=list)
class ReferenceBatchesResponse(BaseModel):
batches: list[ReferenceBatchResponse] = Field(default_factory=list)
class ReferenceDebugLogRequest(BaseModel):
event: str
thread_id: str | None = None
payload: dict[str, Any] = Field(default_factory=dict)
# ---------------------------------------------------------------------------
@@ -163,7 +281,7 @@ def _delete_thread_data(thread_id: str, paths: Paths | None = None, *, user_id:
return ThreadDeleteResponse(success=True, message=f"Deleted local thread data for {thread_id}")
def _derive_thread_status(checkpoint_tuple) -> str:
def _derive_thread_status(checkpoint_tuple) -> str:
"""Derive thread status from checkpoint metadata."""
if checkpoint_tuple is None:
return "idle"
@@ -179,7 +297,251 @@ def _derive_thread_status(checkpoint_tuple) -> str:
if tasks:
return "interrupted"
return "idle"
return "idle"
def _message_type(message: Any) -> str:
if isinstance(message, dict):
return str(message.get("type") or message.get("role") or "")
return str(getattr(message, "type", "") or "")
def _message_id(message: Any) -> str | None:
if isinstance(message, dict):
value = message.get("id")
else:
value = getattr(message, "id", None)
return str(value) if value else None
def _message_content(message: Any) -> Any:
if isinstance(message, dict):
return message.get("content")
return getattr(message, "content", None)
def _message_name(message: Any) -> str:
if isinstance(message, dict):
return str(message.get("name") or (message.get("additional_kwargs") or {}).get("name") or "tool")
return str(getattr(message, "name", None) or "tool")
def _has_tool_calls(message: Any) -> bool:
if isinstance(message, dict):
direct = message.get("tool_calls")
chunks = message.get("tool_call_chunks")
extra = (message.get("additional_kwargs") or {}).get("tool_calls")
else:
direct = getattr(message, "tool_calls", None)
chunks = getattr(message, "tool_call_chunks", None)
extra = (getattr(message, "additional_kwargs", None) or {}).get("tool_calls")
return bool(direct or chunks or extra)
def _has_answer_content(message: Any) -> bool:
content = _message_content(message)
if isinstance(content, str):
return bool(content.strip())
if isinstance(content, list):
for part in content:
if isinstance(part, dict) and part.get("type") == "text" and str(part.get("text") or "").strip():
return True
return False
def _extract_tool_payload(message: Any) -> Any:
content = _message_content(message)
if isinstance(content, dict):
return _unwrap_tool_payload(content)
if isinstance(content, list):
texts: list[str] = []
for part in content:
if not isinstance(part, dict):
continue
if isinstance(part.get("json"), dict):
return _unwrap_tool_payload(part["json"])
if part.get("type") == "json" and isinstance(part.get("json"), dict):
return _unwrap_tool_payload(part["json"])
if part.get("type") == "text" and isinstance(part.get("text"), str):
texts.append(part["text"])
unwrapped = _unwrap_tool_payload(part)
if unwrapped is not None:
return unwrapped
content = "\n".join(texts)
if isinstance(content, str):
return _parse_json_like(content)
return None
def _unwrap_tool_payload(value: dict[str, Any]) -> Any:
if any(isinstance(value.get(key), list) for key in ("results", "data", "items", "hits", "documents", "records", "sources")):
return value
for key in ("result", "output", "stdout", "content", "text", "payload", "response"):
nested = value.get(key)
if isinstance(nested, str):
parsed = _parse_json_like(nested)
if parsed is not None:
return parsed
if isinstance(nested, dict):
unwrapped = _unwrap_tool_payload(nested)
if unwrapped is not None:
return unwrapped
return value
def _parse_json_like(text: str) -> Any:
trimmed = text.strip()
if not trimmed:
return None
try:
return json.loads(trimmed)
except Exception:
pass
first_obj = trimmed.find("{")
last_obj = trimmed.rfind("}")
if 0 <= first_obj < last_obj:
try:
return json.loads(trimmed[first_obj : last_obj + 1])
except Exception:
return None
return None
def _first_list(payload: Any) -> list[Any]:
if isinstance(payload, list):
return payload
if not isinstance(payload, dict):
return []
for key in ("results", "data", "items", "hits", "documents", "records", "entries", "matches", "chunks", "sources"):
value = payload.get(key)
if isinstance(value, list):
return value
for key in ("result", "payload", "response", "output", "data"):
nested = payload.get(key)
found = _first_list(nested)
if found:
return found
return []
def _nested_records(raw: Any) -> list[dict[str, Any]]:
if not isinstance(raw, dict):
return []
layers = [raw]
for key in ("metadata", "meta", "attrs", "attributes", "properties", "extra"):
nested = raw.get(key)
if isinstance(nested, dict):
layers.append(nested)
return layers
def _pick(layers: list[dict[str, Any]], keys: tuple[str, ...]) -> str | None:
for layer in layers:
for key in keys:
value = layer.get(key)
if isinstance(value, (str, int, float)) and str(value).strip():
return str(value).strip()
return None
def _normalize_reference_item(raw: Any, fallback_index: int) -> dict[str, Any] | None:
if not isinstance(raw, dict):
return None
layers = _nested_records(raw)
title = _pick(layers, ("title", "m_title", "name", "question", "keyword", "query", "text"))
content = _pick(layers, ("content", "page_content", "summary", "snippet", "description", "content_preview", "abstract", "body"))
url = _pick(layers, ("url", "link", "href", "source_url", "web_url"))
source = _pick(layers, ("source", "source1", "site", "provider", "from"))
time = _pick(layers, ("time", "date", "published_date", "publishedDate", "m_publish", "publish_time", "created_at"))
item_id = _pick(layers, ("recUuid", "rec_uuid", "uuid", "docId", "doc_id", "id", "recordId", "taskId"))
score_raw = _pick(layers, ("score", "relevance", "similarity"))
score = None
if score_raw is not None:
try:
score = float(score_raw)
except Exception:
score = None
if not title:
title = source or f"结果 {fallback_index}"
if not content:
content = ""
if not title and not content and not url:
return None
return {
"id": item_id or url,
"title": title,
"type": "result",
"snippet": content,
"content": content,
"source": source,
"time": time,
"url": url,
"score": score,
"raw": raw,
}
def _reference_key(source: dict[str, Any]) -> str:
raw = source.get("raw") if isinstance(source.get("raw"), dict) else {}
item_id = (
source.get("id")
or source.get("url")
or raw.get("recUuid")
or raw.get("uuid")
or raw.get("docId")
or raw.get("id")
)
if item_id:
return str(item_id).casefold().strip()
return "|".join(
str(source.get(key) or "").casefold().strip()
for key in ("title", "source", "time", "content", "snippet")
)
def _build_reference_batches(messages: list[Any]) -> list[ReferenceBatchResponse]:
batches: list[ReferenceBatchResponse] = []
for index, message in enumerate(messages):
if _message_type(message) != "human":
continue
end = len(messages)
for cursor in range(index + 1, len(messages)):
if _message_type(messages[cursor]) == "human":
end = cursor
break
assistant_message: Any | None = None
for cursor in range(end - 1, index, -1):
candidate = messages[cursor]
if (
_message_type(candidate) == "ai"
and _message_id(candidate)
and _has_answer_content(candidate)
and not _has_tool_calls(candidate)
):
assistant_message = candidate
break
if assistant_message is None:
continue
seen: set[str] = set()
sources: list[ReferenceSourceResponse] = []
for tool_message in messages[index + 1 : end]:
if _message_type(tool_message) != "tool":
continue
payload = _extract_tool_payload(tool_message)
for raw_item in _first_list(payload):
normalized = _normalize_reference_item(raw_item, len(sources) + 1)
if not normalized:
continue
key = _reference_key(normalized)
if key and key in seen:
continue
if key:
seen.add(key)
sources.append(ReferenceSourceResponse(index=len(sources) + 1, **normalized))
if sources:
mid = _message_id(assistant_message) or str(index)
batches.append(ReferenceBatchResponse(id=mid, assistant_message_id=mid, sources=sources))
return batches
# ---------------------------------------------------------------------------
@@ -406,9 +768,9 @@ async def get_thread(thread_id: str, request: Request) -> ThreadResponse:
# ---------------------------------------------------------------------------
@router.get("/{thread_id}/state", response_model=ThreadStateResponse)
@require_permission("threads", "read", owner_check=True)
async def get_thread_state(thread_id: str, request: Request) -> ThreadStateResponse:
@router.get("/{thread_id}/state", response_model=ThreadStateResponse)
@require_permission("threads", "read", owner_check=True)
async def get_thread_state(thread_id: str, request: Request) -> ThreadStateResponse:
"""Get the latest state snapshot for a thread.
Channel values are serialized to ensure LangChain message objects
@@ -455,12 +817,109 @@ async def get_thread_state(thread_id: str, request: Request) -> ThreadStateRespo
parent_checkpoint_id=parent_checkpoint_id,
created_at=coerce_iso(metadata.get("created_at", "")),
tasks=tasks,
)
@router.post("/{thread_id}/state", response_model=ThreadStateResponse)
@require_permission("threads", "write", owner_check=True, require_existing=True)
async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, request: Request) -> ThreadStateResponse:
)
@router.get("/{thread_id}/reference-batches", response_model=ReferenceBatchesResponse)
@require_permission("threads", "read", owner_check=True)
async def get_thread_reference_batches(thread_id: str, request: Request) -> ReferenceBatchesResponse:
"""Return backend-built citation batches for completed assistant turns."""
checkpointer = get_checkpointer(request)
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
try:
checkpoint_tuple = await checkpointer.aget_tuple(config)
except Exception:
logger.exception("Failed to get reference batches for thread %s", sanitize_log_param(thread_id))
raise HTTPException(status_code=500, detail="Failed to get reference batches")
if checkpoint_tuple is None:
raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
values = serialize_channel_values(checkpoint.get("channel_values", {}) or {})
messages = values.get("messages")
if not isinstance(messages, list):
_log_rag_debug(
"backend reference-batches thread=%s messages=missing batches=0 log_file=%s",
sanitize_log_param(thread_id),
_rag_debug_file_path(),
)
return ReferenceBatchesResponse(batches=[])
batches_before_display_adapt = build_runtime_reference_batches(messages)
raw_batches = batches_before_display_adapt
persisted_displays = await _load_persisted_skill_displays()
if persisted_displays:
raw_batches = adapt_reference_batches_with_displays(raw_batches, persisted_displays)
debug_summary = build_reference_debug_summary(messages, raw_batches)
_log_rag_debug(
"backend reference-batches thread=%s messages=%s turns=%s tool_messages=%s raw_items=%s "
"normalized_items=%s kept_items=%s batches=%s batch_sources=%s batch_ids=%s "
"display_adapters=%s turns_detail=%s log_file=%s",
sanitize_log_param(thread_id),
debug_summary["messages"],
debug_summary["turns"],
debug_summary["tool_messages"],
debug_summary["raw_items"],
debug_summary["normalized_items"],
debug_summary["kept_items"],
len(raw_batches),
",".join(str(batch.get("source_count", 0)) for batch in debug_summary["batches"]),
",".join(str(batch.get("id") or "") for batch in debug_summary["batches"]),
len(persisted_displays),
json.dumps(debug_summary["turns_detail"], ensure_ascii=False),
_rag_debug_file_path(),
)
_log_rag_debug(
"backend reference-batches-payload thread=%s %s",
sanitize_log_param(thread_id),
json.dumps(
{
"display_adapter_names": sorted(persisted_displays.keys()),
"before_display_adapt": build_reference_debug_payload(
messages=messages,
batches=batches_before_display_adapt,
),
"after_display_adapt": build_reference_debug_payload(
messages=messages,
batches=raw_batches,
),
},
ensure_ascii=False,
default=str,
),
)
return ReferenceBatchesResponse(
batches=[
ReferenceBatchResponse(
id=str(batch.get("id") or ""),
assistant_message_id=str(batch.get("assistant_message_id") or batch.get("id") or ""),
sources=[
ReferenceSourceResponse(**source)
for source in batch.get("sources", [])
if isinstance(source, dict)
],
)
for batch in raw_batches
]
)
@router.post("/reference-debug-log")
async def write_reference_debug_log(body: ReferenceDebugLogRequest) -> dict[str, str | bool]:
"""Append frontend citation diagnostics to the backend RAG debug log."""
_get_rag_debug_logger().info(
"frontend %s thread=%s payload=%s log_file=%s",
body.event,
sanitize_log_param(body.thread_id or ""),
json.dumps(body.payload, ensure_ascii=False, default=str),
_rag_debug_file_path(),
)
return {"ok": True, "log_file": _rag_debug_file_path()}
@router.post("/{thread_id}/state", response_model=ThreadStateResponse)
@require_permission("threads", "write", owner_check=True, require_existing=True)
async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, request: Request) -> ThreadStateResponse:
"""Update thread state (e.g. for human-in-the-loop resume or title rename).
Writes a new checkpoint that merges *body.values* into the latest
@@ -0,0 +1,108 @@
"""CRUD API for a user's personal custom prompts (常用提示词).
Prompts are bound to the requesting user (not stored in the browser), so they
follow the account across devices/sessions.
"""
from __future__ import annotations
import logging
from datetime import datetime
from uuid import uuid4
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field
from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/user-prompts", tags=["user-prompts"])
class UserPromptResponse(BaseModel):
id: str
title: str = ""
content: str
sort_order: int = 0
created_at: datetime | str | None = None
updated_at: datetime | str | None = None
class UserPromptsListResponse(BaseModel):
prompts: list[UserPromptResponse]
class UserPromptCreateRequest(BaseModel):
title: str = Field(default="", max_length=255)
content: str = Field(..., min_length=1)
sort_order: int = Field(default=0)
class UserPromptUpdateRequest(BaseModel):
title: str | None = Field(default=None, max_length=255)
content: str | None = Field(default=None, min_length=1)
sort_order: int | None = None
def _current_user_id(request: Request) -> str:
user = getattr(request.state, "user", None)
if user is not None:
return str(user.id)
return get_effective_user_id()
def _get_store(request: Request):
store = getattr(request.app.state, "user_prompt_store", None)
if store is None:
raise HTTPException(status_code=503, detail="User prompt store not available")
return store
@router.get("", response_model=UserPromptsListResponse)
async def list_user_prompts(request: Request) -> UserPromptsListResponse:
store = _get_store(request)
user_id = _current_user_id(request)
rows = await store.list_for_user(user_id)
return UserPromptsListResponse(prompts=[UserPromptResponse(**r) for r in rows])
@router.post("", response_model=UserPromptResponse, status_code=201)
async def create_user_prompt(request: Request, body: UserPromptCreateRequest) -> UserPromptResponse:
store = _get_store(request)
user_id = _current_user_id(request)
row = await store.create(
user_id,
{
"id": uuid4().hex,
"title": (body.title or "").strip(),
"content": body.content,
"sort_order": body.sort_order,
},
)
return UserPromptResponse(**row)
@router.put("/{prompt_id}", response_model=UserPromptResponse)
async def update_user_prompt(request: Request, prompt_id: str, body: UserPromptUpdateRequest) -> UserPromptResponse:
store = _get_store(request)
user_id = _current_user_id(request)
data: dict = {}
if body.title is not None:
data["title"] = body.title.strip()
if body.content is not None:
data["content"] = body.content
if body.sort_order is not None:
data["sort_order"] = body.sort_order
row = await store.update(prompt_id, user_id, data)
if row is None:
raise HTTPException(status_code=404, detail="Prompt not found")
return UserPromptResponse(**row)
@router.delete("/{prompt_id}", status_code=204)
async def delete_user_prompt(request: Request, prompt_id: str) -> None:
store = _get_store(request)
user_id = _current_user_id(request)
deleted = await store.delete(prompt_id, user_id)
if not deleted:
raise HTTPException(status_code=404, detail="Prompt not found")
@@ -11,6 +11,7 @@ import asyncio
import json
import logging
import re
import time
from collections.abc import Mapping
from typing import Any
@@ -102,6 +103,14 @@ _DEFAULT_ASSISTANT_ID = "lead_agent"
# These are NOT stored in the agents DB and must bypass the custom-agent access check.
_BUILTIN_GRAPH_IDS: frozenset[str] = frozenset({"lead_agent", "canvas_agent", "ai_writing"})
_RAG_CITATION_PROMPT = """当前对话已开启参考文献模式。
如果本轮调用了搜索、知识库、检索类技能或工具,请基于本轮所有检索结果整体回答。
如果你已经知道需要多个检索词,请优先在同一步里并行调用检索工具,等这些检索都返回后再回答,不要查一个检索词就先写一版答案。
你必须把同一轮内多次检索得到的资料视为同一个 referenceBatch,按系统稍后提供的合并批次从 [1] 开始连续编号,不要让每个检索词各自从 [1] 重新编号。
回答正文中的关键结论必须标注对应编号,例如 [1]、[2],多个来源可写为 [1][3][5]。
不要引用不存在的编号,不要使用技能内部自带的编号,不要把参考文献完整列表堆在正文末尾。
如果没有任何可用检索结果,请直接说明未检索到可靠参考资料。"""
def _prepend_prompt_prefix(graph_input: dict[str, Any], prefix: str) -> dict[str, Any]:
"""Return ``graph_input`` with the prefix *tagged* onto its first HumanMessage.
@@ -140,6 +149,43 @@ def _prompt_prefix_opted_in(body: Any) -> bool:
return False
def _rag_mode_requested(body: Any) -> bool:
context = getattr(body, "context", None)
if isinstance(context, Mapping) and context.get("rag_mode_enabled"):
return True
metadata = getattr(body, "metadata", None)
if isinstance(metadata, Mapping) and (
metadata.get("rag_mode_enabled") or metadata.get("rag_mode")
):
return True
return False
def _thread_rag_mode_enabled(record: Any) -> bool:
if not isinstance(record, Mapping):
return False
metadata = record.get("metadata")
return isinstance(metadata, Mapping) and bool(
metadata.get("rag_mode_enabled") or metadata.get("rag_mode")
)
def _append_invisible_prompt(graph_input: dict[str, Any], prompt: str) -> dict[str, Any]:
messages = graph_input.get("messages")
if not isinstance(messages, list) or not messages:
return graph_input
new_messages = list(messages)
for idx, msg in enumerate(new_messages):
if not isinstance(msg, HumanMessage):
continue
extra = dict(getattr(msg, "additional_kwargs", None) or {})
existing = str(extra.get("prompt_prefix") or "").strip()
extra["prompt_prefix"] = f"{existing}\n\n{prompt}".strip() if existing else prompt
new_messages[idx] = msg.model_copy(update={"additional_kwargs": extra})
break
return {**graph_input, "messages": new_messages}
# Whitelist of run-context keys that the langgraph-compat layer forwards from
# ``body.context`` into the run config. ``config["context"]`` exists in
# LangGraph >=0.6, but these values must be written to both ``configurable``
@@ -153,6 +199,7 @@ _CONTEXT_CONFIGURABLE_KEYS: frozenset[str] = frozenset(
"mode",
"thinking_enabled",
"reasoning_effort",
"rag_mode_enabled",
"writing_mode",
"writing_artifact_path",
"canvas_artifact_title",
@@ -316,8 +363,17 @@ def _collect_requested_agent_ids(body: Any) -> set[str]:
async def _sync_legacy_agent_records(store) -> None:
"""Import existing folder-backed agents as built-in records."""
import sys
"""Import existing folder-backed agents as built-in records.
Heavy: ``list_custom_agents()`` scans ``.deer-flow/agents/`` and reads
``config.yaml`` + ``SOUL.md`` per directory, then ``store.ensure_builtin``
does one DB round-trip per agent. On a machine with N agent dirs this is
O(N) file IO + O(N) DB calls — historically the worst per-request hotspot
in ``start_run`` (observed 30s+ on machines with 100+ residual coordinator
shells + slow MySQL). **Never call this directly from request handlers** —
use :func:`_maybe_sync_legacy_agent_records` which applies the same TTL
gating as ``routers/agents._maybe_sync_legacy_agents``.
"""
from deerflow.config.agents_config import list_custom_agents
print(f"[_sync_legacy_agent_records] store type: {type(store).__name__}, id: {id(store)}", file=sys.stderr)
@@ -338,6 +394,69 @@ async def _sync_legacy_agent_records(store) -> None:
logger.info("_sync_legacy_agent_records: done")
# Throttle per-request legacy-sync. Matches ``routers/agents._LEGACY_SYNC_TTL_SECONDS``
# semantics — the real sync runs once at startup (lifespan); this is a safety
# net for environments that bypass the lifespan hook or for freshly dropped
# agent directories. Lifespan registers itself via the same state by calling
# the agents router's ``_mark_legacy_sync_fresh()`` after its boot-time sync,
# so this module's TTL state is independent (a separate gate kept here so we
# don't import from ``routers/agents`` and create a circular import).
_LEGACY_SYNC_TTL_SECONDS = 3600.0
_legacy_sync_state: dict[str, Any] = {"last_at": 0.0, "lock": None}
# Anything slower than this triggers the WARNING-level breakdown in
# ``start_run``. 500ms gives plenty of headroom for cold caches + a few DB
# round-trips on healthy systems while still catching the pathological
# multi-second cases that brought us here.
_START_RUN_SLOW_THRESHOLD_SECONDS = 0.5
def _mark_legacy_sync_fresh() -> None:
"""Tell this module's TTL gate that a sync just completed elsewhere.
Called by the Gateway lifespan hook after it runs the boot-time legacy
agent sync (``routers/agents._sync_legacy_agents``). Without this, the
first request after startup would race past the TTL gate here and trigger
a redundant rescan even though the data is already fresh.
"""
_legacy_sync_state["last_at"] = time.monotonic()
async def _maybe_sync_legacy_agent_records(store) -> None:
"""TTL-gated wrapper around :func:`_sync_legacy_agent_records`.
Steady-state behavior: returns in microseconds inside the TTL window — no
filesystem walk, no DB round-trips. Outside the TTL window we serialize
callers under an :class:`asyncio.Lock` so a burst of concurrent
``/runs/stream`` requests fans into a single rescan instead of N parallel
fan-outs. Pattern mirrors ``routers/agents._maybe_sync_legacy_agents``.
"""
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
t_sync_start = time.perf_counter()
await _sync_legacy_agent_records(store)
elapsed = time.perf_counter() - t_sync_start
_legacy_sync_state["last_at"] = time.monotonic()
# When this fires it's a "first request of the hour" warmup — log it
# so operators can see the spike isn't an isolated bug. Anything
# over 2s usually means a lot of agent directories + slow IO/DB.
logger.info(
"[start-run-timing] legacy-agent sync warmed cache in %.3fs (next sync in %.0fs)",
elapsed,
_LEGACY_SYNC_TTL_SECONDS,
)
async def _enforce_agent_access(body: Any, request: Request) -> None:
agent_ids = _collect_requested_agent_ids(body)
if not agent_ids:
@@ -348,9 +467,7 @@ async def _enforce_agent_access(body: Any, request: Request) -> None:
from deerflow.runtime.user_context import get_effective_user_id
store = get_agent_store(request)
import sys
print(f"[_enforce_agent_access] store type: {type(store).__name__}, id: {id(store)}", file=sys.stderr)
await _sync_legacy_agent_records(store)
await _maybe_sync_legacy_agent_records(store)
user_id = get_effective_user_id()
logger.info(f"_enforce_agent_access: user_id={user_id}, agent_ids={agent_ids}")
@@ -396,14 +513,27 @@ async def start_run(
Target thread.
request : Request
FastAPI request — used to retrieve singletons from ``app.state``.
Timing observability
--------------------
Each phase records ``perf_counter`` deltas. When the total wallclock from
function entry to ``run_agent`` task creation exceeds
``_START_RUN_SLOW_THRESHOLD_SECONDS`` we emit a ``[start-run-timing]``
breakdown at WARNING level. Stays at DEBUG when fast, so steady-state
logs are not polluted. Investigate spikes by grep'ing
``[start-run-timing]`` together with ``[upstream-timing]`` (multi-agent
router) and ``[leader-timing] / [special-timing]`` (圆桌 leader/special).
"""
t_entry = time.perf_counter()
bridge = get_stream_bridge(request)
run_mgr = get_run_manager(request)
run_ctx = get_run_context(request)
disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_
t_pre_enforce = time.perf_counter()
await _enforce_agent_access(body, request)
t_post_enforce = time.perf_counter()
try:
record = await run_mgr.create_or_reject(
@@ -418,20 +548,34 @@ async def start_run(
raise HTTPException(status_code=409, detail=str(exc)) from exc
except UnsupportedStrategyError as exc:
raise HTTPException(status_code=501, detail=str(exc)) from exc
t_post_create = time.perf_counter()
# Upsert thread metadata so the thread appears in /threads/search,
# even for threads that were never explicitly created via POST /threads
# (e.g. stateless runs).
requested_rag_mode = _rag_mode_requested(body)
effective_rag_mode = requested_rag_mode
try:
existing = await run_ctx.thread_store.get(thread_id)
effective_rag_mode = requested_rag_mode or _thread_rag_mode_enabled(existing)
if existing is None:
metadata = dict(body.metadata or {})
if effective_rag_mode:
metadata["rag_mode_enabled"] = True
metadata["rag_mode"] = True
await run_ctx.thread_store.create(
thread_id,
assistant_id=body.assistant_id,
metadata=body.metadata,
metadata=metadata,
)
else:
await run_ctx.thread_store.update_status(thread_id, "running")
if requested_rag_mode and not _thread_rag_mode_enabled(existing):
await run_ctx.thread_store.update_metadata(
thread_id,
{"rag_mode_enabled": True, "rag_mode": True},
)
effective_rag_mode = True
# SDK 的 ``client.threads.create()`` 不传 assistant_id —— 先建出来空的 thread,
# 真正提交 run 时才知道用哪个 assistant。这里回填一次,避免历史列表 / 按
# assistant 筛选时拿不到关联。
@@ -439,6 +583,7 @@ async def start_run(
await run_ctx.thread_store.update_assistant_id(thread_id, body.assistant_id)
except Exception:
logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
t_post_thread_upsert = time.perf_counter()
agent_factory = resolve_agent_factory(body.assistant_id)
graph_input = normalize_input(body.input)
@@ -455,6 +600,8 @@ async def start_run(
prefix_settings = load_system_settings().prompt_prefix
if prefix_settings.enabled and prefix_settings.prompt_prefix.strip():
graph_input = _prepend_prompt_prefix(graph_input, prefix_settings.prompt_prefix.strip())
if effective_rag_mode:
graph_input = _append_invisible_prompt(graph_input, _RAG_CITATION_PROMPT)
config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id)
# Merge DeerFlow-specific context overrides into both ``configurable`` and ``context``.
@@ -462,6 +609,8 @@ async def start_run(
# that carries agent configuration (model_name, thinking_enabled, etc.).
# Only agent-relevant keys are forwarded; unknown keys (e.g. thread_id) are ignored.
body_context = dict(getattr(body, "context", None) or {})
if effective_rag_mode:
body_context["rag_mode_enabled"] = True
# Promote whitelisted top-level body fields into the context map so
# callers can pass them either nested under ``context`` or as flat
# top-level fields (``excluded_tools``, ``skill_stop_names``, …).
@@ -473,6 +622,7 @@ async def start_run(
stream_modes = normalize_stream_modes(body.stream_mode)
t_pre_task = time.perf_counter()
task = asyncio.create_task(
run_agent(
bridge,
@@ -497,6 +647,33 @@ async def start_run(
# title from the checkpoint and calls thread_store.update_display_name
# after the run completes.
total = t_pre_task - t_entry
if total >= _START_RUN_SLOW_THRESHOLD_SECONDS:
# Slow path: emit a phase breakdown so on-call can identify the
# culprit without re-running with extra instrumentation. The most
# common cause is a stale ``_maybe_sync_legacy_agent_records`` TTL
# window firing on a machine with hundreds of agent directories.
logger.warning(
"[start-run-timing] thread=%s total=%.3fs enforce_access=%.3fs "
"run_create=%.3fs thread_upsert=%.3fs build_payload=%.3fs",
sanitize_log_param(thread_id),
total,
t_post_enforce - t_pre_enforce,
t_post_create - t_post_enforce,
t_post_thread_upsert - t_post_create,
t_pre_task - t_post_thread_upsert,
)
else:
logger.debug(
"[start-run-timing] thread=%s total=%.3fs enforce=%.3fs create=%.3fs upsert=%.3fs build=%.3fs",
sanitize_log_param(thread_id),
total,
t_post_enforce - t_pre_enforce,
t_post_create - t_post_enforce,
t_post_thread_upsert - t_post_create,
t_pre_task - t_post_thread_upsert,
)
return record