Merge branch 'dhk-1' into 'master'

Dhk 1

See merge request knowledge-base-bj/deerflow-server!57
This commit is contained in:
董宏凯 committed 2026-05-26 06:19:16 +00:00
commit 6a9d8e903e
11 files changed
+647 -54

No files matched your search

+18 -11
View File
@@ -10,8 +10,8 @@
## 0. 速读 ## 0. 速读
- 三个圆桌 Router(`intent / recommend / multi_agent`)**全部走 HTTP loopback**:在同进程内通过 `httpx.AsyncClient` 调本机 `127.0.0.1` 的 `/api/threads`、`/api/agents`、`/api/threads/{id}/runs/stream`。 - 三个圆桌 Router(`intent / recommend / multi_agent`)**通过 httpx ASGITransport 进程内直调本机 ASGI app**,复用既有 `/api/threads`、`/api/agents`、`/api/threads/{id}/runs/stream` 路由——**不走 TCP / 不经过网络栈**。所有 client 都由 `multi_agent._make_loopback_client(request)` 工厂统一构造。
- **所有 loopback `httpx.AsyncClient` 必须 `trust_env=False`**。开发机若设了 `HTTP_PROXY` / `HTTPS_PROXY`,httpx 默认会把 loopback 也走代理 → 不可达 → 单次请求多出 40 秒超时(典型症状日志:`headers_in=42.160s`)。 - 历史包袱:早期实现走 `httpx.AsyncClient(trust_env=False)` + `127.0.0.1:<port>`,遇到开发机 `HTTP_PROXY` / `HTTPS_PROXY` 时仍可能把回环走代理(症状 `headers_in=42.160s`)。**改用 ASGITransport 后此问题不再存在**,相关 `trust_env=False` workaround 全部移除。
- **席位 agent 持久注册**(磁盘 `.deer-flow/agents/` + `_sync_legacy_agents`),**协调智能体每次新建**(命名 `roundtable-coordinator-{timestamp36}`,DB 累积,需后续 GC)。 - **席位 agent 持久注册**(磁盘 `.deer-flow/agents/` + `_sync_legacy_agents`),**协调智能体每次新建**(命名 `roundtable-coordinator-{timestamp36}`,DB 累积,需后续 GC)。
- **`.deer-flow/` 整个目录被 git ignore**(仓库根 `.gitignore` 第 64 / 74 行),`git clone` 后内置 agent 目录在部署机上**不会存在**——任何环境都必须把 `roundtable-intent` + `roundtable-recommender` 两个**功能型 agent** 持久化进发布物,否则 `/api/intent/stream` 报 500。**互联网 demo 还需要 6 个示例研讨席位**(`roundtable-intelligence` 等),**内网部署用客户自有 agent 替代**这些席位,无需打包。详见 §11.7。 - **`.deer-flow/` 整个目录被 git ignore**(仓库根 `.gitignore` 第 64 / 74 行),`git clone` 后内置 agent 目录在部署机上**不会存在**——任何环境都必须把 `roundtable-intent` + `roundtable-recommender` 两个**功能型 agent** 持久化进发布物,否则 `/api/intent/stream` 报 500。**互联网 demo 还需要 6 个示例研讨席位**(`roundtable-intelligence` 等),**内网部署用客户自有 agent 替代**这些席位,无需打包。详见 §11.7。
- **`init` 是 SSE 而非一次性 JSON**:用 `asyncio.as_completed` 按真实完成顺序推送 `seat_ready`,前端因此能逐个点亮席位。 - **`init` 是 SSE 而非一次性 JSON**:用 `asyncio.as_completed` 按真实完成顺序推送 `seat_ready`,前端因此能逐个点亮席位。
@@ -59,7 +59,7 @@ Step 3 最终方案展示
| 2 | 2 | `multi_agent.py` | 协调智能体(每次新建)+ 用户选定的 `roundtable-*` 席位 | | 2 | 2 | `multi_agent.py` | 协调智能体(每次新建)+ 用户选定的 `roundtable-*` 席位 |
| 3 | 3 | `artifacts.py`(通用) | 无新 agent | | 3 | 3 | `artifacts.py`(通用) | 无新 agent |
所有接口需 `Authorization: Bearer <JWT>`。Loopback 调用统一 `trust_env=False`,避免本机 `HTTP_PROXY` 把 `127.0.0.1` 指到不可达代理(症状:`headers_in=40s+`,前端表现为「Step 1 init 卡顿 / Step 2 init 永远不发 seat_ready」)。 所有接口需 `Authorization: Bearer <JWT>`。Loopback 调用一律由 `_make_loopback_client(request)` 构造(`httpx.ASGITransport(app=request.app)`,base_url 占位 `http://loopback`),**不经 TCP**,因此不再受本机 `HTTP_PROXY` 干扰。
--- ---
@@ -499,7 +499,7 @@ async def _broadcast(client, base, headers, message, agent_threads, skip):
(None, captured) — 流结束,captured 是所有 data: 行的字符串列表 (None, captured) — 流结束,captured 是所有 data: 行的字符串列表
``` ```
日志 `[upstream-timing] <agent> headers_in=<>` / `first_line=<>`:排查首 token 慢、代理导致 loopback 超时(见 `trust_env=False`)。 日志 `[upstream-timing] <agent> headers_in=<>` / `first_line=<>`:排查上游首字节延迟;改 ASGITransport 后通常 `headers_in` < 10ms,长尾大概率出在路由 handler 内部(不再是网络/代理问题)。
`_parse_last_messages(captured)` 从倒数往前扫,**跳过 trailing `data: null` end 帧**,找第一条解析为 dict 且含 `messages` key 的 frame;取 `messages` 列表。无可解析帧抛 `502`。 `_parse_last_messages(captured)` 从倒数往前扫,**跳过 trailing `data: null` end 帧**,找第一条解析为 dict 且含 `messages` key 的 frame;取 `messages` 列表。无可解析帧抛 `502`。
@@ -593,14 +593,21 @@ Step 3 **没有** 单独的 `multi_agent` 路由;使用既有 artifacts 能力
## 9. 关键设计决策 ## 9. 关键设计决策
### 9.1 HTTP loopback + `trust_env=False` ### 9.1 ASGITransport 进程内 loopback(替代旧的 HTTP loopback)
三个 router 均通过 `httpx` 调本机 Gateway(`127.0.0.1`),复用鉴权与 LangGraph 路由,代价是额外 HTTP 开销;init 已用 `asyncio.as_completed` 并行化席位准备。 三个 router 通过 `httpx.ASGITransport(app=request.app)` 直调本机 ASGI app,**不走 TCP / 不开 socket / 不经网络栈**,但仍然完整复用 FastAPI 路由分发、鉴权 middleware、Pydantic 校验、SSE 流式响应。代价仅是路由层的少量 dispatch 开销(< 1ms),相对 LLM run 本身的数百 ms~数十秒可忽略。
**每个 `httpx.AsyncClient(timeout=None, trust_env=False)` 都是必须的**: 统一入口:`multi_agent._make_loopback_client(request)`,返回:
- `timeout=None` — SSE 长连接需要 ```python
- `trust_env=False` — 否则 httpx 会读取 `HTTP_PROXY` / `HTTPS_PROXY` 环境变量,把 loopback `127.0.0.1` 也走代理。开发机常配的「校园网代理 / 公司代理」会把回环地址丢弃,导致单次请求 40+ 秒超时(日志典型症状:`headers_in=42.160s`) httpx.AsyncClient(
transport=httpx.ASGITransport(app=request.app),
base_url="http://loopback", # 占位 host,ASGITransport 不在意
timeout=None, # SSE 长连接仍需 None
)
```
**已废止的 `trust_env=False` workaround**:旧版本走 `127.0.0.1:<port>` 真实 HTTP,需要 `trust_env=False` 阻止 httpx 读 `HTTP_PROXY` 把回环也走代理(症状日志 `headers_in=42.160s`)。改 ASGITransport 后这个问题在物理上已不存在——根本没走网络栈。
### 9.2 协调智能体每次新建 ### 9.2 协调智能体每次新建
@@ -1105,7 +1112,7 @@ curl -X POST http://127.0.0.1:8001/api/recommend/stream \
| Step 1 一直有澄清、无法进入 Step 2 | 未产出 `[INTENT_READY]` 或终态被 `clarification` 覆盖 | 看 SOUL、最后一帧 `status`;确认 intent 先扫 READY(§4.3 顺序) | | Step 1 一直有澄清、无法进入 Step 2 | 未产出 `[INTENT_READY]` 或终态被 `clarification` 覆盖 | 看 SOUL、最后一帧 `status`;确认 intent 先扫 READY(§4.3 顺序) |
| 选项全是多选 | 旧前端默认 `allow_multiple !== false` | 确认 `allow_multiple` 字段与 `resolve_allow_multiple`(缺省 false) | | 选项全是多选 | 旧前端默认 `allow_multiple !== false` | 确认 `allow_multiple` 字段与 `resolve_allow_multiple`(缺省 false) |
| Step 2 init SSE `error` | 协调 agent 创建失败 / agent_orchestration 缺失 / `_validate_agent_id` 命中 | Network + 后端 `multi_agent` 日志 | | Step 2 init SSE `error` | 协调 agent 创建失败 / agent_orchestration 缺失 / `_validate_agent_id` 命中 | Network + 后端 `multi_agent` 日志 |
| Step 2 init 完全卡住 ~40 秒 | `httpx` 走了 `HTTP_PROXY` | 确认 `trust_env=False` 仍在;grep `headers_in=` 看时长 | | Step 2 init 完全卡住(历史症状,ASGITransport 改造后已不应再出现) | 旧版 `httpx` 走 `HTTP_PROXY` 拦回环 → 40s 超时 | 自改 ASGITransport 起此现象物理上不可能复现;若仍卡,看后端日志栈、检查路由 handler 是否阻塞 |
| leader 立即 `status: []` | 未调用 orchestration 技能 / SkillStopMiddleware 没拦到 / SOUL 没声明派活 | 换模型或加强 SOUL;检查 `skill_stop_names` 是否包含 `agent_orchestration` | | leader 立即 `status: []` | 未调用 orchestration 技能 / SkillStopMiddleware 没拦到 / SOUL 没声明派活 | 换模型或加强 SOUL;检查 `skill_stop_names` 是否包含 `agent_orchestration` |
| dispatch 失败但 leader 文本看着正常 | 模型把 `roundtable-x(中文)` 整段塞进 `agent_name` | `_strip_display_suffix` 会兜底;若仍报 400,强化 SOUL 的「严格英文 id」 | | dispatch 失败但 leader 文本看着正常 | 模型把 `roundtable-x(中文)` 整段塞进 `agent_name` | `_strip_display_suffix` 会兜底;若仍报 400,强化 SOUL 的「严格英文 id」 |
| 推荐 picks 为空 | 未输出 `[RECOMMEND_READY]` 或 picks 不在候选池 | 看 recommend 终态帧;检查 `candidate_ids` 集合 | | 推荐 picks 为空 | 未输出 `[RECOMMEND_READY]` 或 picks 不在候选池 | 看 recommend 终态帧;检查 `candidate_ids` 集合 |
@@ -1122,7 +1129,7 @@ grep "dispatch failed" .runtime/logs/*.log
grep "upstream-timing" .runtime/logs/*.log grep "upstream-timing" .runtime/logs/*.log
grep "leader-timing" .runtime/logs/*.log grep "leader-timing" .runtime/logs/*.log
grep "special-timing" .runtime/logs/*.log grep "special-timing" .runtime/logs/*.log
grep "headers_in=" .runtime/logs/*.log # 走代理症状 grep "headers_in=" .runtime/logs/*.log # 上游首字节延迟;ASGITransport 改造后通常 < 10ms
``` ```
--- ---
@@ -124,17 +124,17 @@ frontend-web/src/roundtable-planning/
[进入 Step 3] [进入 Step 3]
HighFidelityReport HighFidelityReport
consensusText ← lastLeaderContent consensusText ← lastLeaderContent
seatDeliveries ← step2RoundtableDialogues.filter(d => d.confidence === "子智能体交付") seatDeliveries ← step2RoundtableDialogues.filter(d => d.confidence === "子智能体交付")
Object.entries(threadIds).filter(name !== coordinatorName).map → listSessionArtifacts(threadId) Object.entries(threadIds).filter(name !== coordinatorName).map → listSessionArtifacts(threadId)
文本类文件 → ArtifactPreviewModal;其它 → window.open(downloadUrl) 文本类文件 → ArtifactPreviewModal;其它 → window.open(downloadUrl)
``` ```
--- ---
## 4. API 层 ## 4. API 层
### 4.1 `api/intent.ts` — Step 1 ### 4.1 `api/intent.ts` — Step 1
| 函数 | 说明 | | 函数 | 说明 |
|------|------| |------|------|
@@ -119,7 +119,7 @@ export function WorkspaceNavChatList({
</Link> </Link>
</SidebarMenuButton> </SidebarMenuButton>
</SidebarMenuItem> </SidebarMenuItem>
{/* <SidebarMenuItem> <SidebarMenuItem>
<SidebarMenuButton <SidebarMenuButton
isActive={pathname.startsWith("/page/workspace/roundtable/planning")} isActive={pathname.startsWith("/page/workspace/roundtable/planning")}
asChild asChild
@@ -130,7 +130,7 @@ export function WorkspaceNavChatList({
<span>多智能体规划</span> <span>多智能体规划</span>
</Link> </Link>
</SidebarMenuButton> </SidebarMenuButton>
</SidebarMenuItem> */} </SidebarMenuItem>
<SidebarMenuItem> <SidebarMenuItem>
<SidebarMenuButton <SidebarMenuButton
isActive={pathname === "/page/workspace/scheduled/tasks"} isActive={pathname === "/page/workspace/scheduled/tasks"}
@@ -0,0 +1,411 @@
"""圆桌功能型 agent 的「调用前自检 + 缺失自动补建」工具。
适用场景
--------
``roundtable-intent`` / ``roundtable-recommender`` 是圆桌 Step 1 + 推荐链路的
**功能型** 内置 agent,任何部署都必须存在。但仓库根 ``.gitignore`` 把整个
``.deer-flow/`` 目录排除,新机器 ``git clone`` 之后这两个目录**不会存在**,
直接调 ``/api/intent/stream`` 会因为 agent 解析失败抛 500。
本模块在三个路由入口(``intent.init``、``intent.stream``、``recommend.stream``)
被调用,检查磁盘文件是否齐全;缺失则用下方嵌入的标准模板写回去。
``_sync_legacy_agents`` 会在下一次 ``GET /api/agents`` 时把目录 upsert 进 DB,
所以这里只关心磁盘文件,不直接动 DB。
为什么嵌入字符串而不是用 importlib.resources 资源文件
------------------------------------------------------
SOUL 体量不大(两个加起来 ~200 行),嵌入到 Python 文件里是**单一可信源**——改
SOUL 走代码 review、commit 走 git 历史,且不会因为 wheel 打包漏配
``package-data`` 导致部署机拿不到。
"""
from __future__ import annotations
import logging
from typing import Iterable
import yaml
from deerflow.config.paths import get_paths
logger = logging.getLogger(__name__)
INTENT_AGENT_ID = "roundtable-intent"
INTENT_AGENT_NAME = "任务理解"
INTENT_AGENT_DESCRIPTION = (
"任务理解智能体,通过 2-5 轮简短追问把用户模糊需求转化成结构化的"
"「任务目标+约束+关键假设」,为多智能体圆桌研讨准备输入"
)
# 与 .deer-flow/agents/roundtable-intent/SOUL.md 保持完全一致;
# 改动请同步两处或以本文件为准 + 删除磁盘旧目录让下一次请求重建。
INTENT_AGENT_SOUL = """# 任务理解智能体
你是多智能体圆桌研讨 **Step 1** 的「任务理解专员」。本提示词定义了 Step 1 的完整流程与边界;若管理员为你配置了 Skill,可结合 Skill 补充领域话术或检查清单,但**不得偏离**下文的工作流、`ask_clarification` 用法与 `[INTENT_READY]` 结束协议。
---
## 1. 你的目标
通过 **2–5 轮**简短对话,把用户模糊的需求整理成结构化输入,供 Step 2 圆桌使用:
| 字段 | 含义 |
|------|------|
| `objective` | 一句话任务目标 |
| `constraints` | 硬约束列表(预算/时间/合规/范围等,至少 1 条) |
| `assumptions` | 你替用户做的合理假设(可空数组) |
**优先级**:目标 > 硬约束 > 软偏好。已经清楚的信息不要再问。
---
## 2. 工作流(必须遵守)
```
用户描述任务
↓
信息够不够?
├─ 不够 / 有歧义 → 调用 ask_clarification(可带 options 卡片)
↓
用户回复(可能是「我选择:…」或自由文本)
↓
重复 2–5 轮
↓
信息充分 → 自然语言汇总 + [INTENT_READY] + JSON
↓
停止追问(本轮及之后都不要再 ask_clarification)
```
**关键规则**:
- 澄清阶段 **只** 用 `ask_clarification` 提问;不要用纯文本空问「还有什么补充吗」。
- 一旦输出 `[INTENT_READY]`,**同一轮对话立即结束**,不要再追加追问或工具调用。
- **不做方案、不给技术建议、不写文件、不联网、不派活**——那是 Step 2 六个席位的事。
---
## 3. 唯一工具:`ask_clarification`
当缺少信息、存在歧义、需要用户拍板时,**必须**调用此工具(不要只在正文里提问)。
### 3.1 参数说明
| 参数 | 必填 | 说明 |
|------|------|------|
| `question` | ✅ | 一个聚焦的问题,一次只问一个决策点 |
| `clarification_type` | ✅ | 见下表 |
| `context` | 推荐 | 为什么需要用户确认(1 句话) |
| `options` | 有离散选项时必填 | 字符串列表,如 `["≤500万", "500–1000万", "暂不确定"]` |
| `allow_multiple` | 可选 | 默认 **不传 / False = 单选**;仅当用户**确实可同时选多项**时传 `True` |
### 3.2 `clarification_type` 怎么选
| 类型 | 何时使用 | 选项 UI |
|------|----------|---------|
| `missing_info` | 缺预算、时间、范围等关键事实 | 默认单选;多项约束可同时成立时 `allow_multiple=True` |
| `ambiguous_requirement` | 同一句话多种理解 | 默认单选 |
| `approach_choice` | 在若干方案/路径中选一个 | **单选**(点击即发送) |
| `risk_confirmation` | 需要用户明确确认风险 | **单选** |
| `suggestion` | 你有建议,需用户批准 | **单选** |
### 3.3 选项卡片(前端行为)
- 你提供 `options` 后,前端会渲染**可点击卡片**。
- **默认单选**:用户点一项即发送,无需再点「发送」。
- **多选**(仅 `allow_multiple=True`):用户可多选后点「发送回答」。
- 用户也可能在自定义输入框里打字;回复常见格式:`我选择:A、B;补充:…`——请据此更新理解并继续流程。
### 3.4 调用示例
**缺信息 + 单选预算档:**
```
ask_clarification(
question="这次任务的预算大致在哪个区间?",
clarification_type="missing_info",
context="预算会影响后续圆桌的可行方案范围",
options=["≤ 300 万", "300–800 万", "≥ 800 万", "暂不确定"]
)
```
**选部署方式(单选):**
```
ask_clarification(
question="优先采用哪种部署方式?",
clarification_type="approach_choice",
context="不同部署方式对工期和合规要求差异很大",
options=["仅内网私有化", "混合云", "全公有云"]
)
```
**多项硬约束可同时满足(多选):**
```
ask_clarification(
question="以下哪些约束必须同时满足?(可多选)",
clarification_type="missing_info",
context="我需要确认哪些是硬性底线",
options=["预算上限", "72 小时内交付", "等保三级", "零停机"],
allow_multiple=True
)
```
### 3.5 同轮多个独立问题
若 **多个互不依赖** 的信息都缺失,可在 **同一轮** 多次调用 `ask_clarification`(每个问题一次调用,各自带 `options`)。
若后一个问题依赖前一个答案,**只问第一个**,等用户回复后再问下一个。
---
## 4. 信息充分性判定
满足 **全部** 条件即可结束澄清:
1. 能写出一句明确的 `objective`(用户希望实现 X)
2. `constraints` 至少 1 条硬约束(预算 / 时间 / 合规 / 范围 之一)
3. 没有会妨碍 Step 2 研讨的明显歧义
若用户**首条消息**已包含目标 + 约束,**直接汇总**,不要多余追问。
---
## 5. 结束协议:`[INTENT_READY]`
信息充分时,回复结构必须是:
1. **自然语言**向用户复述「我理解的任务是 …」(📋 汇总,让用户 review)
2. **紧接着**输出下方固定标记 + JSON(后端靠此解锁「进入 Step 2」)
格式要求(**严格**):
```
[INTENT_READY]
```json
{
"objective": "用一句话写出用户最终要做的事",
"constraints": ["预算 ≤ XXX 万", "时间 ≤ XXX 天", ...],
"assumptions": ["假设 1", "假设 2"]
}
```
```
- `[INTENT_READY]` **单独成行**,前后不要夹其他字符
- 标记后**紧跟** ```json 代码块
- 字段名必须是 `objective` / `constraints` / `assumptions`
- `constraints` 至少 1 条;`assumptions` 可为 `[]`
**输出 `[INTENT_READY]` 后禁止再调用 `ask_clarification` 或继续追问。**
---
## 6. 输出风格
- 普通澄清轮:不超过 **3 句话**(工具调用前的引导语可更短)
- 提问要**具体**(❓),汇总要**结构化**(✅ 📋)
- 适度 emoji:❓ 提问、✅ 已明确、📋 汇总
---
## 7. 禁止事项
- ❌ `agent_orchestration`、读写文件、bash、联网搜索、memory 等——你无权使用
- ❌ 输出具体实施方案、架构图、代码
- ❌ 在已输出 `[INTENT_READY]` 之后继续提问
- ❌ 用「还有什么要补充的吗」代替 `ask_clarification`
---
## 8. 完整对话示例
**用户**:帮我规划一个生产系统安全改造
**你**(调用工具):
```
ask_clarification(
question="预算大致在哪个区间?工期有没有硬性上限?",
clarification_type="missing_info",
context="这两项是后续圆桌排期的基础",
options=["预算 ≤300万 / 6个月内", "预算 300–800万 / 6–12个月", "预算 ≥800万 / 时间较灵活", "暂不确定,需先评估"]
)
```
**用户**:我选择:预算 300–800万 / 6–12个月
**你**(若还需确认合规):
```
ask_clarification(
question="是否需要满足特定合规标准?",
clarification_type="missing_info",
options=["等保 2.0", "等保 3.0", "行业监管无特殊要求", "暂不确定"]
)
```
**用户**:我选择:等保 2.0
**你**(汇总,不再调用工具):
✅ 我理解的任务是:在 300–800 万预算、6–12 个月工期内,对生产系统做安全改造并满足等保 2.0。
📋
[INTENT_READY]
```json
{
"objective": "对生产系统进行安全改造,满足等保 2.0 合规要求",
"constraints": ["预算 300–800 万元", "工期 6–12 个月", "必须满足等保 2.0"],
"assumptions": ["允许在预定窗口内安排必要停机", "现有系统具备基础数字化能力"]
}
```
"""
RECOMMENDER_AGENT_ID = "roundtable-recommender"
RECOMMENDER_AGENT_NAME = "智能体推荐"
RECOMMENDER_AGENT_DESCRIPTION = (
"智能体推荐专员,基于用户的任务目标与约束,从候选智能体池中挑选最相关的 "
"2-6 个智能体并给出推荐理由,为多智能体圆桌研讨准备最终参会名单"
)
RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
## 角色定位
你是多智能体圆桌研讨的"智能体推荐专员"。任务理解专员已经把用户需求整理成「任务目标 + 约束 + 关键假设」交给你,你的工作是从一个**候选智能体池**里,挑出**最适合**参加这次圆桌研讨的 2-6 个智能体,并为每一位写一句推荐理由。
## 核心职责
- 仔细阅读用户的任务目标与约束
- 浏览候选池中每个智能体的 `name` 与 `description`
- 选出与本次任务**最相关**的 2-6 个智能体(数量动态,不必凑齐 6 个)
- 为每个推荐的智能体写一句**具体、可验证**的推荐理由——必须能回答"它能为这次任务做什么"
- 不推荐重复职能的智能体(如果两个智能体能力高度重叠,只选一个)
- 不推荐与任务明显无关的智能体(宁缺毋滥)
## 数量约束
- **下限**:至少推荐 2 个(单个智能体谈不上"圆桌")
- **上限**:最多推荐 6 个(超过 6 个会让协调智能体派活混乱、上下文膨胀)
- 候选池不足 2 个时,有几个推几个
## 输出协议
**你必须严格按下述格式输出**(后端用正则识别):
1. 先用自然语言简短说明你的推荐思路(1-3 句话,让用户能 review)
2. 然后输出标记 `[RECOMMEND_READY]` 单独成行
3. 紧接一个 ```json 代码块,内容为:
```json
{
"picks": [
{"agent_id": "<候选池中的智能体 id>", "reason": "<一句推荐理由>"},
...
]
}
```
格式注意:
- `[RECOMMEND_READY]` 标记必须**单独成行**,前后无其他字符
- 标记紧接着一个 ```json 代码块
- `picks` 数组长度严格在 [2, 6] 之间(候选池不足时按实际数量,但不要少于 2 除非候选池本身就 < 2)
- `agent_id` 必须**严格**来自候选池(不要造词)
- `reason` 必须**具体**,不要"很合适"、"很强"这类空话
## 输出风格
- 思路部分简短:用一句话概括你为什么挑这些(例如"按情报→设计→风险→执行的研讨节奏挑选了 4 位")
- 推荐理由要**具体到任务**,例如"负责检索同类合规改造案例,直接对应本次等保 2.0 的对标需求"
- 适度用 emoji:📋 表示开始汇总
## 边界
- 不调用任何工具(不联网、不执行命令、不读写文件)
- 不评价候选智能体的优劣(只判断"是否适合本次任务")
- 不发起对话、不反问用户——你只产出推荐结果
- 候选池里没有的智能体绝不出现在 `picks` 中
## 示例
**用户输入**(由后端拼装):
```
任务目标:对生产系统进行安全改造,满足等保 2.0 合规要求
约束:预算 ≤ 500 万元;工期 ≤ 6 个月;必须满足等保 2.0
关键假设:现有系统具备基础数字化能力;允许预定窗口停机
候选智能体池:
- roundtable-intelligence (情报收集):检索同类历史案例、行业最佳实践与外部情报
- roundtable-environment (环境评估):评估硬件容量与系统量化承载力
- roundtable-solution-design (方案设计):产出主备方案与关键路径
- roundtable-risk-review (风险审查):合规红线检查与收敛建议
- roundtable-execution-plan (执行规划):里程碑与资源排布
- roundtable-summary (总结陈述):归纳冲突标记与盖印
- my-cost-calc (成本测算):基于预算和资源单价计算总成本
```
**你的输出**:
按"情报→设计→风险→执行"的研讨节奏,挑选 5 位最贴合本次等保 2.0 改造任务的智能体,并加入成本测算覆盖预算硬约束。📋
[RECOMMEND_READY]
```json
{
"picks": [
{"agent_id": "roundtable-intelligence", "reason": "检索同类等保 2.0 改造案例与监管口径,直接支撑本次合规对标"},
{"agent_id": "roundtable-solution-design", "reason": "在 6 个月工期约束下产出主备改造方案与关键路径"},
{"agent_id": "roundtable-risk-review", "reason": "对照等保 2.0 合规红线审查方案,避免后期返工"},
{"agent_id": "roundtable-execution-plan", "reason": "把改造拆成可在 6 个月内交付的里程碑与资源排布"},
{"agent_id": "my-cost-calc", "reason": "校验整体方案落在 500 万元预算硬约束内"}
]
}
```
"""
_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),
)
def _ensure_one(agent_id: str, name: str, description: str, soul: str) -> 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).
"""
agent_dir = get_paths().agent_dir(agent_id)
if agent_dir.exists():
return False
agent_dir.mkdir(parents=True, exist_ok=True)
# newline="\n" 强制 LF;否则 Windows 下 open() 把 \n 翻成 \r\n,跨 OS 部署时与
# Linux 部署机字节级不一致(功能上 YAML/Markdown 都兼容,但 git diff / 哈希对不齐)。
config_yaml = yaml.dump(
{"id": agent_id, "name": name, "description": description},
default_flow_style=False,
allow_unicode=True,
)
(agent_dir / "config.yaml").write_bytes(config_yaml.encode("utf-8"))
(agent_dir / "SOUL.md").write_bytes(soul.encode("utf-8"))
return True
def ensure_roundtable_functional_agents() -> list[str]:
"""Seed `roundtable-intent` + `roundtable-recommender` if their directories
are missing. Idempotent — fast no-op when both exist.
Called from the entry of intent/init, intent/stream, and recommend/stream
so the routes self-heal on fresh internal deployments where `.deer-flow/`
is gitignored and the agent directories were never shipped.
Returns the list of agent ids that were just created (empty when nothing
needed seeding). Errors are logged but never raised — a partial failure
must not block the request; the downstream route will surface its own
actionable error if the agent really cannot be resolved.
"""
created: list[str] = []
for agent_id, name, description, soul in _BUILTIN_FUNCTIONAL_AGENTS:
try:
if _ensure_one(agent_id, name, description, soul):
created.append(agent_id)
except Exception:
logger.exception("Failed to seed roundtable functional agent '%s'", agent_id)
if created:
logger.info("Seeded missing roundtable functional agents: %s", ", ".join(created))
return created
def required_agent_ids() -> Iterable[str]:
"""Stable ids of agents this seeder owns — exposed for tests/diagnostics."""
return tuple(a[0] for a in _BUILTIN_FUNCTIONAL_AGENTS)
@@ -63,6 +63,7 @@ from fastapi.responses import StreamingResponse
# Reuse the building blocks already in multi_agent.py — keeps both routers # Reuse the building blocks already in multi_agent.py — keeps both routers
# identical in connection handling, payload shape, and error mapping. # 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 ( from app.gateway.routers.multi_agent import (
_auth_headers, _auth_headers,
_create_thread, _create_thread,
@@ -71,6 +72,7 @@ from app.gateway.routers.multi_agent import (
_find_trailing_ai_text, _find_trailing_ai_text,
_flatten_content, _flatten_content,
_loopback_base, _loopback_base,
_make_loopback_client,
_parse_last_messages, _parse_last_messages,
_run_payload, _run_payload,
_stream_upstream, _stream_upstream,
@@ -106,15 +108,17 @@ async def init_intent(request: Request, payload: dict = Body(...)) -> dict[str,
与 ``POST /api/multi-agent/init`` 的区别:不创建协调智能体、不建多席位 与 ``POST /api/multi-agent/init`` 的区别:不创建协调智能体、不建多席位
thread 映射;agent 本身已通过磁盘 ``.deer-flow/agents/`` 同步注册。 thread 映射;agent 本身已通过磁盘 ``.deer-flow/agents/`` 同步注册。
""" """
# 调用前自检:内网部署 .deer-flow/ 被 gitignore,首次启动时本 agent 目录可能不存在,
# 缺失则就地用嵌入模板补建,避免下一步 /stream 报 500。
ensure_roundtable_functional_agents()
base = _loopback_base(request) base = _loopback_base(request)
headers = _auth_headers(request) headers = _auth_headers(request)
# trust_env=False: this client only talks to the same-process gateway at # TCP loopback 到 127.0.0.1:<gateway port>(详见 multi_agent._make_loopback_client)。
# 127.0.0.1, so honoring HTTP_PROXY / HTTPS_PROXY env vars (set system-wide # 不要在这里替换成 httpx.ASGITransport——曾试过,会整段 buffer SSE 流导致前端
# on the dev machine) makes httpx route loopback calls through an upstream # 看不到逐 token 增量。
# proxy that can't reach 127.0.0.1, costing 40+ seconds of timeout per
# request before falling back. Symptom in logs: `headers_in=42.160s`.
try: try:
async with httpx.AsyncClient(timeout=None, trust_env=False) as client: async with _make_loopback_client(request) as client:
thread_id = await _create_thread(client, base, headers) thread_id = await _create_thread(client, base, headers)
except HTTPException: except HTTPException:
# 上游已用结构化方式失败,原样传给前端不需要二次包装。 # 上游已用结构化方式失败,原样传给前端不需要二次包装。
@@ -161,15 +165,16 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
if not message: if not message:
raise HTTPException(status_code=400, detail="message is required") raise HTTPException(status_code=400, detail="message is required")
# 调用前自检:防御性二次检查——若管理员在 init 之后手动删了目录,这里也能自愈。
ensure_roundtable_functional_agents()
base = _loopback_base(request) base = _loopback_base(request)
headers = _auth_headers(request) headers = _auth_headers(request)
async def stream_generator() -> AsyncIterator[str]: async def stream_generator() -> AsyncIterator[str]:
# trust_env=False: same-process loopback to 127.0.0.1 — honoring # ASGITransport:进程内直调本机 ASGI app,**不走 TCP / 不经过网络栈**,
# HTTP_PROXY / HTTPS_PROXY env vars (commonly set on dev machines) # 因此既无须 trust_env=False 防代理,也不存在 loopback 40s 超时这种坑。
# causes httpx to route loopback calls through an unreachable upstream async with _make_loopback_client(request) as client:
# proxy and wait ~40 seconds for timeout before falling back.
async with httpx.AsyncClient(timeout=None, trust_env=False) as client:
try: try:
body = _run_payload( body = _run_payload(
agent_name=INTENT_AGENT_ID, agent_name=INTENT_AGENT_ID,
@@ -49,12 +49,14 @@ from __future__ import annotations
import asyncio import asyncio
import json import json
import logging import logging
import os
import re import re
import time import time
import traceback import traceback
from typing import Any, AsyncIterator from typing import Any, AsyncIterator
import httpx import httpx
from deerflow.config import get_app_config
from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
from fastapi import APIRouter, Body, HTTPException, Request from fastapi import APIRouter, Body, HTTPException, Request
from fastapi.responses import StreamingResponse from fastapi.responses import StreamingResponse
@@ -97,13 +99,15 @@ def _diagnose_exception(exc: BaseException, *, context: str = "") -> dict[str, A
if isinstance(root, httpx.ConnectError): if isinstance(root, httpx.ConnectError):
target = str(root).strip().lower() target = str(root).strip().lower()
# loopback 失败 = Gateway 自己没起 / 代理拦回环;outbound 失败 = LLM 提供商不可达。 # loopback 走真 TCP 到 127.0.0.1:<gateway port>;失败 = Gateway 进程没起 / 端口被占。
# outbound 失败(其它 host)= LLM 提供商不可达。
if "127.0.0.1" in target or "localhost" in target or "loopback" in target: if "127.0.0.1" in target or "localhost" in target or "loopback" in target:
code = "loopback_connect_failed" code = "loopback_connect_failed"
hint = ( hint = (
"本地 Gateway loopback 连接失败:确认后端进程在 8001 端口监听;" "本地 Gateway loopback 连接失败:确认后端进程正在监听配置的端口。"
"另检查 HTTP_PROXY / HTTPS_PROXY 是否把 127.0.0.1 也走了代理" "端口配置优先级(高→低):env DEER_FLOW_GATEWAY_PORT > config.yaml 的 "
"(代码已显式 trust_env=False,理论上不该走代理)。" "gateway.port > 默认 8001。若部署机改了端口,在 config.yaml 加 "
"`gateway:\\n port: <端口>` 即可。"
) )
else: else:
code = "upstream_connect_failed" code = "upstream_connect_failed"
@@ -160,12 +164,118 @@ def _diagnose_exception(exc: BaseException, *, context: str = "") -> dict[str, A
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Loopback 辅助函数(Gateway → Gateway,同进程) # Loopback 辅助函数(Gateway → Gateway,同进程)
#
# 圆桌三路由(intent / recommend / multi_agent)需要从一个 FastAPI handler 里
# 调用本机 Gateway 的另一个 endpoint(`/api/threads`、`/api/agents`、
# `/api/threads/{id}/runs/stream`)。这里用真 TCP loopback 到本机的 127.0.0.1
# + Gateway 监听端口。
#
# ⚠️ 历史踩坑(三个) ⚠️
# 1. 直接走系统代理:开发机若设了 HTTP_PROXY / HTTPS_PROXY,httpx 默认会把
# 127.0.0.1 也走代理 → 上游代理不可达,40s 超时(典型日志 headers_in=42.160s)。
# → 修法:client 用 `trust_env=False`,强制忽略代理 env vars。
#
# 2. 用 request.url.netloc 拼 base URL:Nginx 反代部署时,客户端打到 nginx 的
# public host(如 example.com),nginx 把请求转给 gateway 时 Host header
# 通常是 `gateway:8001` 或 public host;后者 httpx 会真的去 DNS 解析,
# 在 gateway 容器内可能不可达 → 连接失败。
# → 修法:**写死 127.0.0.1**,完全无视 request.url.netloc。
#
# 3. 一度尝试用 httpx.ASGITransport(app=request.app) 进程内直调,看似一劳永逸,
# 但 httpx 0.28 的 ASGITransport 对 streaming response **整段 buffer**,
# 一直收齐所有 body chunks 才一次性把响应给消费方 → SSE 完全失效,前端表现
# 为"调度阶段很长 + 流式结果一次性出现"。已弃用。
# → 详见:git log 6e3a07b(那次改 ASGI 引入)及随后回退。
#
# 端口策略(优先级,从高到低):
# 1. 环境变量 DEER_FLOW_GATEWAY_PORT
# ── 部署侧紧急覆盖通道,适合 docker-compose / k8s 注入或临时调试
# 2. config.yaml 的 `gateway.port`
# ── 部署侧**推荐**的配置方式,改完不需要重启二进制(get_app_config 有 mtime
# 热加载,但仍建议跑一次 reload 以确保所有 worker 拿到新值)
# 3. request.url.port
# ── 浏览器直连 gateway:8001 时拿得到;走 Nginx 80/443 反代时会落到下一档
# 4. 默认 8001(与 `make dev` / `make gateway` / `scripts/start-all.sh` 写死的一致)
#
# 端口在哪里改(给运维 / 部署侧的提示):
# - yaml 配置:打开 config.yaml,加 / 改:
# gateway:
# port: 9999
# - 环境变量:
# export DEER_FLOW_GATEWAY_PORT=9999
#
# 修改这两个函数时务必跑一次 SSE 流式 smoke 验证(见 multi-agent-backend-dev.md
# §13 测试建议),确认 chunks 是逐个到达而不是一次喷出。
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# 写死的最终 fallback。**不要**为了改端口而修改这个常量——优先用 config.yaml /
# env 这两层(它们在部署侧不需要改源码即可生效)。这个值只在 yaml 缺失 + env
# 没设 + 请求里也拿不到合法 port 时才会用到。
_DEFAULT_GATEWAY_PORT = 8001
def _resolve_gateway_port(request: Request) -> int:
"""按四档 fallback 解析 Gateway 监听端口。详见模块顶部「端口策略」。"""
# 第 1 档:环境变量。允许非法值(garbage),只警告并继续下一档,不抛异常。
env_port = os.environ.get("DEER_FLOW_GATEWAY_PORT", "").strip()
if env_port:
try:
port = int(env_port)
if 1 <= port <= 65535:
return port
raise ValueError("out of range")
except ValueError:
logger.warning(
"DEER_FLOW_GATEWAY_PORT=%r 不是合法端口,fallback 到 config.yaml / request / %d",
env_port,
_DEFAULT_GATEWAY_PORT,
)
# 第 2 档:config.yaml 的 gateway.port。GatewayConfig.port 的 pydantic 校验
# 已经保证它在 1-65535 范围内,无需再次校验。读 config 失败时降级,不阻断业务。
try:
configured = get_app_config().gateway.port
if configured:
return configured
except Exception as exc:
# 极端场景:config.yaml 不存在或解析失败。这种情况下 gateway 本身也起不来,
# loopback 端口"猜错"也已经无所谓。仅记 debug 日志方便排查。
logger.debug("read gateway.port from app config failed: %s", exc)
# 第 3 档:从当前请求拿 port。浏览器直连 gateway 时是真实监听端口;
# 走 Nginx 反代时通常是 80/443(public port),那就跳过,落到最终默认。
inbound_port = request.url.port
if inbound_port is not None and inbound_port not in (80, 443):
return inbound_port
# 第 4 档:写死默认。
return _DEFAULT_GATEWAY_PORT
def _loopback_base(request: Request) -> str: def _loopback_base(request: Request) -> str:
"""当前请求的 scheme+host,用于 httpx 打回本机 Gateway。""" """返回 `http://127.0.0.1:<port>`,供 loopback 调用拼 URL。
return f"{request.url.scheme}://{request.url.netloc}"
强制 127.0.0.1 是为了避免 Nginx 反代场景下读到经过 Host 改写的 netloc。
端口按 `_resolve_gateway_port` 的四档 fallback 解析。
"""
return f"http://127.0.0.1:{_resolve_gateway_port(request)}"
def _make_loopback_client(request: Request) -> httpx.AsyncClient:
"""构造同进程 loopback 用的 httpx.AsyncClient(真 TCP)。
所有圆桌路由(intent / recommend / multi_agent)的 loopback 调用必须走这里,
保持配置一致。
- `trust_env=False`:**关键**,忽略 HTTP_PROXY / HTTPS_PROXY,防代理拦截 127.0.0.1
- `timeout=None`:SSE 长连接需要,LangGraph runs/stream 可能跑数十秒
- `request` 参数当前未被使用,保留是为了:(a) 未来若需要从 request 取部署
上下文(如多租户端口隔离)可以无侵入扩展;(b) 与调用方现有签名兼容。
"""
return httpx.AsyncClient(
timeout=None,
trust_env=False,
)
def _auth_headers(request: Request) -> dict[str, str]: def _auth_headers(request: Request) -> dict[str, str]:
@@ -431,11 +541,9 @@ async def init_multi_agent(request: Request, payload: dict = Body(...)) -> Strea
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n" return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
async def stream_generator() -> AsyncIterator[str]: async def stream_generator() -> AsyncIterator[str]:
# trust_env=False: same-process loopback to 127.0.0.1 — honoring # TCP loopback 到 127.0.0.1:<gateway port>。client 配置见 _make_loopback_client:
# HTTP_PROXY / HTTPS_PROXY env vars (commonly set on dev machines) # trust_env=False 防 HTTP_PROXY 拦截,timeout=None 容纳 SSE 长连接。
# causes httpx to route loopback calls through an unreachable upstream async with _make_loopback_client(request) as client:
# proxy and wait ~40 seconds for timeout before falling back.
async with httpx.AsyncClient(timeout=None, trust_env=False) as client:
try: try:
# Per-seat prepare task — same body as before, but we no longer # Per-seat prepare task — same body as before, but we no longer
# await them with asyncio.gather. Instead, asyncio.as_completed # await them with asyncio.gather. Instead, asyncio.as_completed
@@ -552,7 +660,9 @@ async def _stream_upstream(
- ``(None, captured)`` — 流结束,``captured`` 为所有 ``data:`` 行文本列表, - ``(None, captured)`` — 流结束,``captured`` 为所有 ``data:`` 行文本列表,
供 ``_parse_last_messages`` 从倒数 values 快照提取 ``messages``。 供 ``_parse_last_messages`` 从倒数 values 快照提取 ``messages``。
日志 ``[upstream-timing]``:排查首 token 慢、代理导致 loopback 超时(见 trust_env=False)。 日志 ``[upstream-timing]``:排查首 token 慢;headers_in 长尾若 >> 网络往返(loopback
本机往返通常 <10ms),怀疑 HTTP_PROXY 拦了回环——检查 client 是否走 _make_loopback_client
(trust_env=False)。
""" """
url = f"{base}/api/threads/{thread_id}/runs/stream" url = f"{base}/api/threads/{thread_id}/runs/stream"
@@ -968,11 +1078,9 @@ async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) ->
headers = _auth_headers(request) headers = _auth_headers(request)
async def stream_generator() -> AsyncIterator[str]: async def stream_generator() -> AsyncIterator[str]:
# trust_env=False: same-process loopback to 127.0.0.1 — honoring # TCP loopback 到 127.0.0.1:<gateway port>。client 配置见 _make_loopback_client:
# HTTP_PROXY / HTTPS_PROXY env vars (commonly set on dev machines) # trust_env=False 防 HTTP_PROXY 拦截,timeout=None 容纳 SSE 长连接。
# causes httpx to route loopback calls through an unreachable upstream async with _make_loopback_client(request) as client:
# proxy and wait ~40 seconds for timeout before falling back.
async with httpx.AsyncClient(timeout=None, trust_env=False) as client:
try: try:
if agent_type == "leader": if agent_type == "leader":
gen = _leader_run( gen = _leader_run(
@@ -47,11 +47,13 @@ import httpx
from fastapi import APIRouter, Body, HTTPException, Request from fastapi import APIRouter, Body, HTTPException, Request
from fastapi.responses import StreamingResponse from fastapi.responses import StreamingResponse
from app.gateway.routers._roundtable_seed import ensure_roundtable_functional_agents
from app.gateway.routers.multi_agent import ( from app.gateway.routers.multi_agent import (
_auth_headers, _auth_headers,
_create_thread, _create_thread,
_flatten_content, _flatten_content,
_loopback_base, _loopback_base,
_make_loopback_client,
_parse_last_messages, _parse_last_messages,
_run_payload, _run_payload,
_stream_upstream, _stream_upstream,
@@ -188,16 +190,19 @@ async def stream_recommend(request: Request, payload: dict = Body(...)) -> Strea
if not candidate_ids: if not candidate_ids:
raise HTTPException(status_code=400, detail="candidates must contain at least one valid agent_id") raise HTTPException(status_code=400, detail="candidates must contain at least one valid agent_id")
# 调用前自检:内网部署 .deer-flow/ 被 gitignore,roundtable-recommender 目录可能不存在,
# 缺失则用嵌入模板补建,避免下游 lead_agent 解析失败抛 500。
ensure_roundtable_functional_agents()
prompt = _build_prompt(intent, candidates) prompt = _build_prompt(intent, candidates)
base = _loopback_base(request) base = _loopback_base(request)
headers = _auth_headers(request) headers = _auth_headers(request)
async def stream_generator() -> AsyncIterator[str]: async def stream_generator() -> AsyncIterator[str]:
# trust_env=False: same-process loopback to 127.0.0.1 — honoring # TCP loopback 到 127.0.0.1:<gateway port>(详见 multi_agent._make_loopback_client)。
# HTTP_PROXY / HTTPS_PROXY env vars (commonly set on dev machines) # 不要在这里替换成 httpx.ASGITransport——曾试过,会整段 buffer SSE 流导致前端
# causes httpx to route loopback calls through an unreachable upstream # 看不到逐 token 增量。
# proxy and wait ~40 seconds for timeout before falling back. async with _make_loopback_client(request) as client:
async with httpx.AsyncClient(timeout=None, trust_env=False) as client:
try: try:
# 每次推荐独立 thread;prompt 已包含完整 intent + 候选列表,无需多轮。 # 每次推荐独立 thread;prompt 已包含完整 intent + 候选列表,无需多轮。
thread_id = await _create_thread(client, base, headers) thread_id = await _create_thread(client, base, headers)
@@ -2,6 +2,13 @@
config_version: 8 config_version: 8
log_level: info log_level: info
# Gateway FastAPI 监听端口。圆桌三路由(intent / recommend / multi_agent)的同进程
# loopback 调用会拼 http://127.0.0.1:<port> 打到本机其它 endpoint,所以必须与
# 实际监听端口一致(scripts/start-all.sh、Makefile 默认 8001)。
# 部署机改了端口请同步改这里;也可设环境变量 DEER_FLOW_GATEWAY_PORT 临时覆盖。
gateway:
port: 8001
models: models:
- name: deepseek-chat - name: deepseek-chat
display_name: DeepSeek Chat display_name: DeepSeek Chat
@@ -1,5 +1,6 @@
from .app_config import get_app_config from .app_config import get_app_config
from .extensions_config import ExtensionsConfig, get_extensions_config from .extensions_config import ExtensionsConfig, get_extensions_config
from .gateway_config import GatewayConfig
from .memory_config import MemoryConfig, get_memory_config from .memory_config import MemoryConfig, get_memory_config
from .paths import Paths, get_paths from .paths import Paths, get_paths
from .skill_evolution_config import SkillEvolutionConfig from .skill_evolution_config import SkillEvolutionConfig
@@ -20,6 +21,7 @@ __all__ = [
"SkillsConfig", "SkillsConfig",
"ExtensionsConfig", "ExtensionsConfig",
"get_extensions_config", "get_extensions_config",
"GatewayConfig",
"MemoryConfig", "MemoryConfig",
"get_memory_config", "get_memory_config",
"get_tracing_config", "get_tracing_config",
@@ -13,6 +13,7 @@ from deerflow.config.agents_api_config import AgentsApiConfig, load_agents_api_c
from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
from deerflow.config.database_config import DatabaseConfig from deerflow.config.database_config import DatabaseConfig
from deerflow.config.extensions_config import ExtensionsConfig from deerflow.config.extensions_config import ExtensionsConfig
from deerflow.config.gateway_config import GatewayConfig
from deerflow.config.guardrails_config import GuardrailsConfig, load_guardrails_config_from_dict from deerflow.config.guardrails_config import GuardrailsConfig, load_guardrails_config_from_dict
from deerflow.config.memory_config import MemoryConfig, load_memory_config_from_dict from deerflow.config.memory_config import MemoryConfig, load_memory_config_from_dict
from deerflow.config.model_config import ModelConfig from deerflow.config.model_config import ModelConfig
@@ -122,6 +123,7 @@ class AppConfig(BaseModel):
skill_stop: SkillStopConfig = Field(default_factory=SkillStopConfig, description="Skill stop middleware configuration") skill_stop: SkillStopConfig = Field(default_factory=SkillStopConfig, description="Skill stop middleware configuration")
model_config = ConfigDict(extra="allow") model_config = ConfigDict(extra="allow")
database: DatabaseConfig = Field(default_factory=DatabaseConfig, description="Unified database backend configuration") database: DatabaseConfig = Field(default_factory=DatabaseConfig, description="Unified database backend configuration")
gateway: GatewayConfig = Field(default_factory=GatewayConfig, description="Gateway FastAPI listening port (used by intra-process loopback)")
run_events: RunEventsConfig = Field(default_factory=RunEventsConfig, description="Run event storage configuration") run_events: RunEventsConfig = Field(default_factory=RunEventsConfig, description="Run event storage configuration")
checkpointer: CheckpointerConfig | None = Field(default=None, description="Checkpointer configuration") checkpointer: CheckpointerConfig | None = Field(default=None, description="Checkpointer configuration")
stream_bridge: StreamBridgeConfig | None = Field(default=None, description="Stream bridge configuration") stream_bridge: StreamBridgeConfig | None = Field(default=None, description="Stream bridge configuration")
@@ -0,0 +1,46 @@
"""Gateway 进程的本机绑定配置。
目前唯一字段是 ``port``——Gateway FastAPI 的监听端口。圆桌三路由
(intent / recommend / multi_agent) 内部用 loopback 调本机的其它 endpoint
(``/api/threads``、``/api/agents`` 等),需要知道这个端口才能拼出 base URL。
为什么放在配置里而不是写死
--------------------------
- 部署方常常会改 Gateway 监听端口(端口被占用、合规要求、容器化端口映射),
写死代码意味着每次改端口都得改源码 + 重新构建/分发,**改 config 不需要重启二进制**
- ``DEER_FLOW_GATEWAY_PORT`` 环境变量仍然是更高优先级的紧急覆盖通道,适合
临时调试或者 docker-compose / k8s 注入
修改方式(部署侧)
------------------
在 ``config.yaml`` 顶层加:
.. code-block:: yaml
gateway:
port: 9999 # 默认 8001;部署机改了 gateway 端口时改这里
只想临时改一次(不动 yaml)→ 设环境变量:
.. code-block:: bash
export DEER_FLOW_GATEWAY_PORT=9999
"""
from __future__ import annotations
from pydantic import BaseModel, Field
class GatewayConfig(BaseModel):
"""Gateway 本机 loopback 端口配置。"""
port: int = Field(
default=8001,
ge=1,
le=65535,
description=(
"Gateway FastAPI 监听端口。圆桌内部 loopback 调用会拼到 "
"``http://127.0.0.1:<port>``。默认 8001(与 Makefile/scripts/start-all.sh 对齐)。"
),
)