代码修改

This commit is contained in:
董宏凯 committed 2026-05-26 14:18:00 +08:00
1 parent d1e6cd5a8f
commit daba5c0599
8 files changed
+174 -34

No files matched your search

@@ -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"}
@@ -114,8 +114,9 @@ async def init_intent(request: Request, payload: dict = Body(...)) -> dict[str,
base = _loopback_base(request) base = _loopback_base(request)
headers = _auth_headers(request) headers = _auth_headers(request)
# ASGITransport:进程内直调本机 ASGI app,**不走 TCP / 不经过网络栈**, # TCP loopback 到 127.0.0.1:<gateway port>(详见 multi_agent._make_loopback_client)。
# 因此既无须 trust_env=False 防代理,也不存在 loopback 40s 超时这种坑。 # 不要在这里替换成 httpx.ASGITransport——曾试过,会整段 buffer SSE 流导致前端
# 看不到逐 token 增量。
try: try:
async with _make_loopback_client(request) 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)
@@ -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,12 +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 已改用 ASGITransport(进程内直调),不再经网络;ConnectError 几乎全部来自上游 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 ASGI loopback 失败:这种情况罕见,通常意味着 ASGI app " "本地 Gateway loopback 连接失败:确认后端进程正在监听配置的端口。"
"未正确注入到 request.app,或路由层抛出未捕获异常。请查后端日志栈。" "端口配置优先级(高→低):env DEER_FLOW_GATEWAY_PORT > config.yaml 的 "
"gateway.port > 默认 8001。若部署机改了端口,在 config.yaml 加 "
"`gateway:\\n port: <端口>` 即可。"
) )
else: else:
code = "upstream_connect_failed" code = "upstream_connect_failed"
@@ -160,42 +165,116 @@ def _diagnose_exception(exc: BaseException, *, context: str = "") -> dict[str, A
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Loopback 辅助函数(Gateway → Gateway,同进程) # Loopback 辅助函数(Gateway → Gateway,同进程)
# #
# 通过 httpx ASGITransport 直接调用本进程的 ASGI app,**不走 TCP / 不经过网络栈**: # 圆桌三路由(intent / recommend / multi_agent)需要从一个 FastAPI handler 里
# - 彻底规避「开发机 HTTP_PROXY 把 loopback 也走代理」导致的 40s 超时 # 调用本机 Gateway 的另一个 endpoint(`/api/threads`、`/api/agents`、
# - 不再需要 trust_env=False / 真实端口 / 真实 host # `/api/threads/{id}/runs/stream`)。这里用真 TCP loopback 到本机的 127.0.0.1
# - 路由分发、鉴权 middleware、Pydantic 校验、SSE 流式响应一切照旧 # + Gateway 监听端口。
# #
# 历史背景:早期实现走 `httpx.AsyncClient(trust_env=False)` + `127.0.0.1:<port>`, # ⚠️ 历史踩坑(三个) ⚠️
# 见 git log 与 multi-agent-backend-dev.md §9.1。改用 ASGITransport 后已不需要 # 1. 直接走系统代理:开发机若设了 HTTP_PROXY / HTTPS_PROXY,httpx 默认会把
# 防代理 workaround,但 _loopback_base 仍返回一个占位 base_url 以保持原有 helper # 127.0.0.1 也走代理 → 上游代理不可达,40s 超时(典型日志 headers_in=42.160s)。
# 函数签名(_create_thread / _stream_upstream 等)零改动。 # → 修法: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 是逐个到达而不是一次喷出。
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# ASGITransport 不在意 host,仅作 URL 拼接占位。 # 写死的最终 fallback。**不要**为了改端口而修改这个常量——优先用 config.yaml /
_LOOPBACK_BASE_URL = "http://loopback" # 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:
"""返回 loopback base URL 占位串。 """返回 `http://127.0.0.1:<port>`,供 loopback 调用拼 URL。
ASGI transport 模式下不再用真实 scheme+host,但保留此函数以兼容 强制 127.0.0.1 是为了避免 Nginx 反代场景下读到经过 Host 改写的 netloc。
`_create_thread / _stream_upstream` 等内部 helper 的 `base` 参数。 端口按 `_resolve_gateway_port` 的四档 fallback 解析。
""" """
return _LOOPBACK_BASE_URL return f"http://127.0.0.1:{_resolve_gateway_port(request)}"
def _make_loopback_client(request: Request) -> httpx.AsyncClient: def _make_loopback_client(request: Request) -> httpx.AsyncClient:
"""构造一个走 ASGITransport 的 httpx.AsyncClient,**不走网络栈**。 """构造同进程 loopback 用的 httpx.AsyncClient(真 TCP)。
所有圆桌路由(intent / recommend / multi_agent)的 loopback 调用都走这里, 所有圆桌路由(intent / recommend / multi_agent)的 loopback 调用必须走这里,
替代原先的 `httpx.AsyncClient(timeout=None, trust_env=False)` + 127.0.0.1。 保持配置一致。
- `trust_env=False`:**关键**,忽略 HTTP_PROXY / HTTPS_PROXY,防代理拦截 127.0.0.1
- `timeout=None`:SSE 长连接需要,LangGraph runs/stream 可能跑数十秒
- `request` 参数当前未被使用,保留是为了:(a) 未来若需要从 request 取部署
上下文(如多租户端口隔离)可以无侵入扩展;(b) 与调用方现有签名兼容。
""" """
transport = httpx.ASGITransport(app=request.app)
return httpx.AsyncClient( return httpx.AsyncClient(
transport=transport,
base_url=_LOOPBACK_BASE_URL,
timeout=None, timeout=None,
trust_env=False,
) )
@@ -462,8 +541,8 @@ 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]:
# ASGITransport:进程内直调本机 ASGI app,**不走 TCP / 不经过网络栈**, # TCP loopback 到 127.0.0.1:<gateway port>。client 配置见 _make_loopback_client:
# 因此既无须 trust_env=False 防代理,也不存在 loopback 40s 超时这种坑。 # trust_env=False 防 HTTP_PROXY 拦截,timeout=None 容纳 SSE 长连接。
async with _make_loopback_client(request) as client: async with _make_loopback_client(request) 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
@@ -581,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 已走 ASGITransport,不会再有代理超时。 日志 ``[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"
@@ -997,8 +1078,8 @@ 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]:
# ASGITransport:进程内直调本机 ASGI app,**不走 TCP / 不经过网络栈**, # TCP loopback 到 127.0.0.1:<gateway port>。client 配置见 _make_loopback_client:
# 因此既无须 trust_env=False 防代理,也不存在 loopback 40s 超时这种坑。 # trust_env=False 防 HTTP_PROXY 拦截,timeout=None 容纳 SSE 长连接。
async with _make_loopback_client(request) as client: async with _make_loopback_client(request) as client:
try: try:
if agent_type == "leader": if agent_type == "leader":
@@ -199,8 +199,9 @@ async def stream_recommend(request: Request, payload: dict = Body(...)) -> Strea
headers = _auth_headers(request) headers = _auth_headers(request)
async def stream_generator() -> AsyncIterator[str]: async def stream_generator() -> AsyncIterator[str]:
# ASGITransport:进程内直调本机 ASGI app,**不走 TCP / 不经过网络栈**, # TCP loopback 到 127.0.0.1:<gateway port>(详见 multi_agent._make_loopback_client)。
# 因此既无须 trust_env=False 防代理,也不存在 loopback 40s 超时这种坑。 # 不要在这里替换成 httpx.ASGITransport——曾试过,会整段 buffer SSE 流导致前端
# 看不到逐 token 增量。
async with _make_loopback_client(request) as client: async with _make_loopback_client(request) as client:
try: try:
# 每次推荐独立 thread;prompt 已包含完整 intent + 候选列表,无需多轮。 # 每次推荐独立 thread;prompt 已包含完整 intent + 候选列表,无需多轮。
@@ -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 对齐)。"
),
)