02 · Bridge 核心层

架构剖析第二篇：FastAPI 桥的路由组织、启动副作用顺序、SSE 事件泵、DI 容器、并发控制。对应 ARCHITECTURE.md §3。

§3.1 技术选型

维度	选型	理由
Web 框架	FastAPI	pydantic 类型安全、自动 OpenAPI、ASGI 支持 HTTP/2 + SSE
生产服务器	hypercorn	HTTP/2 多路复用，解决 Chromium 6-socket/host 限制
回退服务器	uvicorn	`FAMILIARS_HTTP_INSECURE=1` 时启用，plain HTTP
证书	自签 `workspace/.bridge.crt`	`src/bridge/certs.py` 自动生成
监听地址	`127.0.0.1:18711`（默认）	不暴露外部端口；端口文件 `workspace/.bridge.port` 是实际信号

§3.2 入口与启动（`src/bridge/run.py`）

python

# src/bridge/run.py 顶部
_REPO_ROOT = str(Path(__file__).resolve().parents[2])
os.environ.setdefault("WORKSPACE_ROUTE", _REPO_ROOT)

启动顺序（python -m src.bridge.run）：

Pin WORKSPACE_ROUTE 环境变量到仓库根
import src.bridge.app（触发 app.py 的模块级副作用，见 §3.4）
hypercorn.serve 启动事件循环 + worker threadpool

daemon 模式（FAMILIARS_DAEMON=1）：

pidfile 锁 workspace/.bridge.pid
stdout/stderr 重定向到 workspace/.bridge.log（rotating）
SIGTERM/SIGINT → 优雅关闭 APScheduler

§3.3 路由组织

src/bridge/routes/ 下共 60 个路由文件（截至当前版本）。每个文件导出一个 APIRouter：

路由文件	主要端点前缀	职责
`agents.py`	`/api/agents`	Agent CRUD、列表、身份、人设演化
`chat.py`	`/api/chat`	SSE 聊天流 `/api/chat/stream`、同步发送
`tree.py`	`/api/tree`	对话树 CRUD、节点操作
`settings.py`	`/api/settings`	全局/Agent 设置
`cron.py`	`/api/cron`	Cron 作业管理
`schedule.py`	`/api/schedule`	日程 CRUD
`todo.py`	`/api/todo`	待办事项
`reminders.py`	`/api/reminders`	提醒（SSE 流 `/api/reminders/stream`）
`skills.py`	`/api/skills`	技能列表/调用
`monitor.py`	`/api/monitor`	社交平台监控
`digital_human.py`	`/api/digital_human`	数字人配置/渲染
`asr.py`	`/api/asr`	语音识别（faster-whisper）
`auth.py`	`/api/auth`	首次启动注册/登录
`usage.py`	`/api/usage`	用量追踪
`status.py`	`/api/status`	健康检查
`reports.py`	`/api/reports`	报告生成/列表
`gift.py`	`/api/gift`	礼物/兑换码
`cg.py`	`/api/cg`	CG 动画工坊
`agent_workshop.py`	`/api/agent_workshop`	Agent 工坊
`skill_forge.py`	`/api/skill_forge`	技能锻造
`skill_proposals.py`	`/api/skill_proposals`	技能提案
`skill_workshop.py`	`/api/skill_workshop`	技能工坊
`feishu.py`	`/api/feishu`	飞书集成
`background.py`	`/api/background`	后台任务
`user.py`	`/api/user`	用户资料
`user_profile.py`	`/api/user/profile`	用户画像
`agent_recipes.py`	`/api/agent_recipes`	配方管理
`broker.py`	`/api/broker`	Broker 层（Hub/匹配/账本）
`broker_workshop.py`	`/api/broker_workshop`	Broker 工坊（安装/管理）
`broker_app.py`	`/api/broker_app`	Broker 应用（iframe UI）
`broker_assets.py`	`/api/broker_assets`	Broker 静态资产
`broker_dev.py`	`/api/broker_dev`	Broker 开发工具
`flow_templates.py`	`/api/flow-templates`	流模板 CRUD + 自定义字段渲染器
`flows.py`	`/api/flows`	流实例（list/get/trace/cancel）
`flow_dev.py`	`/api/flow_dev`	流开发工具
`broadcast.py`	`/api/broadcast`	广播消息
`memory_status.py`	`/api/memory_status`	记忆规模/命中
`memory.py`	`/api/memory`	记忆操作
`onboarding.py`	`/api/onboarding`	新手引导
`phone_call.py`	`/api/phone_call`	语音通话
`quick_ask.py`	`/api/quick_ask`	QuickAsk 弹出
`quick_inputs.py`	`/api/quick_inputs`	快速输入
`draft.py`	`/api/draft`	草稿
`feedback.py`	`/api/feedback`	反馈（含 proactive 评分）
`demo.py`	`/api/demo`	公开 demo
`desktop_notify.py`	`/api/desktop_notify`	桌面通知
`dev.py`	`/api/dev`	开发工具
`recipe_dev.py`	`/api/recipe_dev`	配方开发（编译/干跑/e2e）
`responses.py`	`/api/responses`	审批响应
`setup.py`	`/api/setup`	安装向导
`tool_skills.py`	`/api/tool_skills`	工具技能
`tools_catalog.py`	`/api/tools_catalog`	工具目录
`tools_source.py`	`/api/tools_source`	工具源码
`vram.py`	`/api/vram`	VRAM 状态
`agent_activity.py`	`/api/agent_activity`	Agent 活动
`agent_assets.py`	`/api/agent_assets`	Agent 资产
`guild.py`	`/api/guild`	公会系统
`learn.py`	`/api/learn`	学习模块
`marketplace_artifacts.py`	`/api/marketplace_artifacts`	市场制品
`xun_da.py`	`/api/xun_da`	寻达（XunDa）模块

路由注册（src/bridge/app.py）：

python

from src.bridge.routes import (
    agent_activity, agent_recipes, agent_workshop, agents, agent_assets,
    asr, auth, background, broadcast, broker, broker_app, broker_assets,
    broker_dev, broker_workshop, cg, chat, cron, demo, desktop_notify,
    dev, digital_human, draft, feedback, feishu, flow_dev, flow_templates,
    flows, gift, guild, learn, marketplace_artifacts, memory, memory_status,
    monitor, onboarding, phone_call, quick_ask, quick_inputs, recipe_dev,
    reminders, reports, responses, schedule, settings, setup, skill_forge,
    skill_proposals, skill_workshop, skills, status, todo, tool_skills,
    tools_catalog, tools_source, tree, usage, user, user_profile, vram, xun_da,
)

§3.4 启动副作用顺序（关键！）

src/bridge/app.py 在模块级（import 时）按严格顺序执行：

步骤 1：Pin WORKSPACE_ROUTE

python

os.environ.setdefault("WORKSPACE_ROUTE", str(_Path(__file__).resolve().parents[2]))

步骤 2：CUDA 主线程 Pin

python

try:
    import torch as _torch
    if _torch.cuda.is_available():
        _torch.cuda.init()
        _ = _torch.zeros(1, device="cuda")
        _torch.cuda.synchronize()
except Exception:
    pass

为什么：Windows WDDM 不允许 worker 线程首触 CUDA。如果 torch.zeros 在 asyncio worker thread 执行，整个进程以 0xC0000005（access violation）崩溃。

步骤 3：路径对齐

python

from src.base.paths_align import align_all_paths as _align_all_paths
_align_summary = _align_all_paths()

为什么：DialogueManager._load_existing_dialogues 读取对话文件时，文件中的绝对路径必须正确。如果用户移动了仓库目录，align_all_paths 自动重写 dialogue.json + turn_*/plan.json + step_*.json。

步骤 4：Import + Mount 路由

所有 src.bridge.routes.* 在此处导入并 mount 到 FastAPI app。

步骤 5：CORS + Token 中间件

python

app.add_middleware(CORSMiddleware, allow_origins=[...])

# BridgeTokenMiddleware：检查 X-Bridge-Token 或 ?token=
# 豁免列表见 _TOKEN_EXEMPT_PREFIXES

步骤 6：`on_event("startup")` Hooks

按注册顺序：

broker_workshop bootstrap —— 加载已安装的 broker 应用
anyio 线程限制器提升 —— 防止 asyncio worker pool 饱和
池看门狗 —— 监控 + 自动扩容
浏览器生命周期 —— Chrome CDP 连接管理

步骤 7：模块级收尾

杀孤儿 GPU 进程
预热 embedder / rembg / rewriter / 最近 Agent 的数字人
autorun 检查 + 漂移检测
rehydrate 监控 / cron / 技能调度

§3.5 DI 容器（`src/bridge/deps.py`）

python

# 全局单例缓存
_dm_cache: dict[int, object] = {}   # agent_id → DialogueManager
_daily: Optional[DailyTools] = None
_manager: Optional[object] = None   # AssistantManager singleton
_comm_hub: Optional[CommHub] = None
_broker_hub: Optional[object] = None

提供函数：

函数	返回	说明
`get_auth()`	`AuthManager`	认证管理
`get_usage()`	`UsageTracker`	用量追踪
`get_daily()`	`DailyTools`	日程/待办/提醒工具
`get_assistant_manager()`	`AssistantManager`	Agent 管理单例
`get_dialogue_manager(agent_id)`	`DialogueManager`	按 agent 缓存
`peek_dialogue_manager(agent_id)`	`DialogueManager \| None`	不触发创建
`drop_dialogue_manager(agent_id)`	`None`	从缓存移除
`release_dialogue_manager(agent_id)`	`bool`	移除 + 释放内存
`get_comm_hub()`	`CommHub`	通信中枢
`get_broker_hub()`	`BrokerHub`	Broker 层（懒加载）

Agent 0（manager）特殊处理：

python

if agent_id == 0:
    agent_path = os.path.join(workspace, "workspace", "manager")
    mem_id = "manager"
else:
    agent_path = resolve_agent_dir(agent_id)
    mem_id = f"agent_{agent_id}"

§3.6 SSE 事件泵

`/api/chat/stream`

python

# chat.py 端点
@router.post("/api/chat/stream")
async def chat_stream(request: ChatRequest):
    async def event_generator():
        async for event in conversation_session.stream(...):
            yield f"data: {json.dumps(event)}\n\n"
    return StreamingResponse(event_generator(), media_type="text/event-stream")

SSE 事件类型：

token —— 流式 token
tool_call —— 工具调用开始
tool_result —— 工具结果
emit_event —— 自定义事件（audio/video/工作日志）
done —— 完成
error —— 错误

并发控制

每个对话一个异步任务
HTTP/2 多路复用 → 客户端可并行多个流
事件通过 asyncio.Queue 推给 renderer

§3.7 请求上下文隔离

python

# src/bridge/app.py
def _fresh_request_context() -> "contextvars.Context":
    ctx = contextvars.copy_context()
    def _reset_and_stamp():
        for var in (trace_id_var, agent_id_var, dialogue_id_var,
                    owner_label_var, _current_agent, _current_dialogue,
                    _current_intense, _current_turbo):
            var.set(None)
        bind_trace(trace_id=new_trace_id())
        bind_actor(agent_id=0, dialogue_id=0)
    ctx.run(_reset_and_stamp)
    return ctx

问题：FastAPI 的 BaseHTTPMiddleware 在 request coroutine 上运行，Starlette 复用 worker thread 之间不重置 contextvars。前一个请求的 agent_id/trace_id/recipe_copilot 会泄漏到下一个请求。

解决：copy_context() 快照当前上下文 → POP 所有可观测性/安全变量 → 盖章新值 → 处理器在 ctx.run(...) 内执行。

下一篇：03-Agent 管线 →

02 · Bridge 核心层 ​

§3.1 技术选型 ​

§3.2 入口与启动（src/bridge/run.py） ​

§3.3 路由组织 ​

§3.4 启动副作用顺序（关键！） ​

步骤 1：Pin WORKSPACE_ROUTE ​

步骤 2：CUDA 主线程 Pin ​

步骤 3：路径对齐 ​

步骤 4：Import + Mount 路由 ​

步骤 5：CORS + Token 中间件 ​

步骤 6：on_event("startup") Hooks ​

步骤 7：模块级收尾 ​

§3.5 DI 容器（src/bridge/deps.py） ​

§3.6 SSE 事件泵 ​

/api/chat/stream ​

并发控制 ​

§3.7 请求上下文隔离 ​

02 · Bridge 核心层

§3.1 技术选型

§3.2 入口与启动（`src/bridge/run.py`）

§3.3 路由组织

§3.4 启动副作用顺序（关键！）

步骤 1：Pin WORKSPACE_ROUTE

步骤 2：CUDA 主线程 Pin

步骤 3：路径对齐

步骤 4：Import + Mount 路由

步骤 5：CORS + Token 中间件

步骤 6：`on_event("startup")` Hooks

步骤 7：模块级收尾

§3.5 DI 容器（`src/bridge/deps.py`）

§3.6 SSE 事件泵

`/api/chat/stream`

并发控制

§3.7 请求上下文隔离