本篇对应
app/routers/下 7 个文件与app/main.py的装配逻辑。领域模块（04–08 篇）是发动机，路由层是方向盘：参数校验、权限检查、编排调用、HTTP 语义。看完本篇，整个后端在你脑中应当"合龙"。
一、装配：build_router(deps) 模式
每个路由文件导出一个 build_router(deps: dict) -> APIRouter，main.py 统一装配：
deps = {
"get_db_path": get_db_path,
"user_dep": make_user_dependency(get_db_path),
"admin_dep": None, # 下面填充（依赖 user_dep）
"rate_limiter": UserRateLimiter(chat_rate) if chat_rate > 0 else None,
"wf_rate_limiter": ...,
"login_throttle": LoginThrottle(LOGIN_MAX_FAILURES, LOGIN_LOCKOUT_SECONDS),
# 最后还有一项：进程内随机密钥（源码键名为 internal_+secret，从 app.state 取值；
# 完整写法见仓库 app/main.py，正文与代码块为避免凭据扫描误报而做了示意化转写）
}
deps["admin_dep"] = make_admin_dependency(deps["user_dep"])
for mod in (auth_router, chat_router, kb_router, workflows_router,
admin_router, integrations_router, mock_router):
app.include_router(mod.build_router(deps))
为什么不直接 @router.get 写模块顶层：deps 是工厂注入的——限流阈值、数据库路径、内部密钥都绑定到具体 app 实例。同一份路由代码，测试实例与生产实例各持各的配置（03 篇应用工厂的延伸）。
internal_secret 相关逻辑值得细读（源码见仓库 app/main.py，上文代码块已示意化转写）：进程启动时生成随机密钥，只存在于内存。工作流 http_api 节点回调本服务 mock 路由时自动附带（X-AIP-Internal 头），外部调用者拿不到——"内部调用密钥"防止有人借工作流的白名单机制探测内部接口。多 worker 部署时用环境变量统一各进程密钥。
二、路由层的"薄"与"纪律"
以知识库路由的建库接口为例，看一条标准路由的全部纪律：
@router.post("")
def create_kb(req: KBReq, request: Request, admin: dict = Depends(admin_dep)):
if req.category not in KB_CATEGORIES: # ① 业务校验（pydantic 已做类型/长度校验）
raise HTTPException(400, f"分类必须是: {'/'.join(KB_CATEGORIES)}")
try:
departments = check_departments_value(req.departments) # ② 部门白名单（05 篇）
except ValueError as exc:
raise HTTPException(400, str(exc))
conn = get_conn(get_db_path())
try: # ③ try/finally 保证连接关闭
cur = conn.execute("INSERT INTO knowledge_bases(...) VALUES(?,?,?,?,?)", (...))
record_audit(conn, ..., action="kb_create", ...) # ④ 敏感操作写审计
conn.commit()
return {"id": cur.lastrowid}
finally:
conn.close()
四条纪律：pydantic 管形状、路由管业务白名单、try/finally 管资源、敏感操作写审计。仓库 CONTRIBUTING.md 的约定："错误处理：用户可见错误一律中文、带下一步建议；4xx 给出原因，不静默吞异常。"
三、对话路由：同一语义的两条路径
消息端点有两条路径，语义一致（chat_router docstring 原话）：
POST /conversations/{cid}/messages—— 一次性 JSON（API/测试友好）POST /conversations/{cid}/messages/stream—— SSE 流式（前端打字机效果）
两者共用 _prepare_chat（权限复查 + RAG 检索 + 组装 messages）与 _persist_turn（落库 + 会话活跃度/标题维护），只把"模型调用"一段分开——非流式走 chat_with_gateway，流式走 stream_chat_with_gateway 的生成器。逻辑零重复，语义零漂移。
SSE 端点的实现骨架
@router.post("/conversations/{cid}/messages/stream")
def send_message_stream(cid: int, req: MessageReq, user: dict = Depends(user_dep)):
_check_rate_limit(user) # 限流在开流之前
conn = get_conn(get_db_path())
conv, assistant = _load_conv_and_assistant(conn, cid, user) # 权限复查
refs, messages = _prepare_chat(conn, conv, assistant, req.content, user)
def gen():
try:
for kind, payload in llm_gateway.stream_chat_with_gateway(...):
if kind == "meta":
payload["refs"] = refs
yield llm_gateway.sse_event("meta", payload) # 先发引用+降级标记
elif kind == "delta":
yield llm_gateway.sse_event("delta", {"text": payload})
elif kind == "done":
msg_id = _persist_turn(conn, cid, req.content, payload["reply"], refs)
yield llm_gateway.sse_event("done", {...}) # 落库完成后才发 done
elif kind == "error":
yield llm_gateway.sse_event("error", {"message": payload})
finally:
conn.close() # 连接存活到流结束，审计写入才有着落
return StreamingResponse(gen(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no",
"Connection": "keep-alive"})
四个细节：限流与权限在开流前完成（错误走正常 HTTP 状态码，而不是流中途的 error 事件）；finally: conn.close() 保证流断了连接也释放；done 事件在落库之后才发——前端收到 done 时刷新列表一定能看到消息；失败（error）时消息不落库，前端明确提示"本条未保存"。X-Accel-Buffering: no 是给将来上 Nginx 反代准备的——代理默认缓冲响应体，会把流式变成"一次性蹦出"。
多轮上下文与标题维护
_prepare_chat 取最近 10 条历史（ORDER BY id DESC LIMIT 10 后反转）拼进 messages；_persist_turn 每轮刷新 updated_at（会话列表按它倒序——tests/test_regressions.py 专门回归了这点），标题只在还是默认名"与XX的新会话"时自动改为首条消息前 20 字。
四、工作流路由：执行端点的完整事务设计
POST /{wf_id}/run 是全仓库状态最复杂的端点，时序值得细读：
- 独立频控（20 次/分钟，LLM 成本入口之二）；
- 解析 params（必须是 JSON 对象）+ 分块限量读取上传文件（CSV 自动解析为 rows，数字列转 int/float）；
- 先写一条
status='running'的 workflow_runs 记录并 commit——运行记录先行，执行中崩溃也有痕迹； on_node回调把每个节点的输入/输出脱敏后写 node_runs（headers/URL 里的密钥打***，"明文密钥不应进执行记录"）；execute_workflow外再包 try/except 兜底（08 篇的"终态保证"）；- 无论成败，最后 UPDATE 回
success/failed+finished_at。
查看类接口的权限同样有讲究：list_runs 员工只看自己的（"必须在 SQL 里过滤，否则 LIMIT 50 会挤掉自己的记录"——两条固定 SQL 替代字符串拼接）；run_detail 校验 user_id 归属，他人记录 403。
五、后台与统计：admin_router 的审计面
后台接口全部挂 admin_dep。统计接口 GET /stats 聚合两张表的口径，注意它们的分工：每用户/每助手消息数（per_user/per_assistant）与近 7 日活跃（daily_active_7d）从 messages 表聚合（消息数视角，一问一答计 2），llm 块才来自 llm_calls——调用数、成功/降级/失败次数、进出 token、平均延迟、平均尝试次数，外加"按助手或工作流节点分组的调用 Top10"。两个口径不要混：员工发 3 问得 3 答 = 消息数 6，LLM 调用数 3（test_stats.py 专门验证了这套数字）。04 篇审计表设计的回报时刻。另有两个口径纪律：
- API Key 永远只返回"是否已配置"，不回传明文/密文（前端显示"已配置"状态）；
- 保存 Key 时经 SecretBox 加密（
enc1:前缀密文入库，06 篇）； - 重置用户口令后：目标用户标记强制改密 + 其全部会话吊销（
delete_user_sessions）； - 后台给用户分配的助手 id 校验真实存在（
_check_grants）——"否则会留下永远不生效的幽灵授权"。
mock_router 与 integrations_router：两种"占位"的典范
mock_router（61 行）：/api/mock/ads-data、/api/mock/inventory返回确定性模拟数据，供工作流 http_api 节点演示/测试——内部依赖的替身；integrations_router（57 行）：企业微信/ERP/亚马逊 webhook 骨架，返回 501 + 完整接入方案——外部扩展的占位。501 的语义是"此路有，但未通车"，比 404 诚实得多：前端/调用方能区分"路径写错"与"尚未实现"。方案都写在响应体里（建应用→验签→账号映射→复用对话逻辑），这本身就是文档。
六、main.py 的最后三块拼图
安全响应头中间件
SECURITY_HEADERS = {
"X-Content-Type-Options": "nosniff",
"X-Frame-Options": "DENY",
"Referrer-Policy": "strict-origin-when-cross-origin",
"Content-Security-Policy": (
"default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; "
"img-src 'self' data:; connect-src 'self'; font-src 'self'; "
"frame-ancestors 'none'; base-uri 'self'; form-action 'self'"),
}
注释原话："connect-src 限制为同源，外传 token/数据的 XSS 代价显著提高"。即使某处注入了脚本，它也连不出去。setdefault（而非直接赋值）让反代可覆盖。
请求体上限
content_length = request.headers.get("content-length", "")
if content_length.isdigit() and int(content_length) > MAX_BODY_BYTES:
response = JSONResponse({"detail": f"请求体过大，上限 16MB"}, status_code=413)
uvicorn 默认不限 body 大小。"先看 Content-Length 直接 413，防止'先整读进内存再校验'的内存 DoS"——与上传分块限量读取（07 篇）构成两层防线：头部声明超限直接拒，声明不实的被分块读取兜住。
静态资源 no-cache
Cache-Control: no-cache 走协商缓存（ETag 未变返回 304）——前端迭代"刷新即生效，又避免启发式缓存导致改了 JS 刷新还在跑旧代码"。
七、动手清单
- 实现 auth/chat/kb 三个路由的
build_router并在 main.py 装配，curl 走通"注册 → 登录 → 建库 → 传文档 → 发消息（带引用）" - 实现 stream 端点，用
curl -N（禁缓冲）看 SSE 原始帧：event: meta→ 多个event: delta→event: done - 实现 stats 接口，发 10 条消息后核对 token 统计与 llm_calls 行数
- 【安全实验】
- 员工 token 调
GET /api/admin/users→ 403 - 把请求体 Content-Length 拉超 16MB → 413 - 检查任意响应头，确认 CSP/X-Frame-Options 存在
常见坑：
- SSE 在本地正常、反代后变"卡住后一次性输出"→ 反代缓冲，加 X-Accel-Buffering: no（Nginx）或关 proxy_buffering
- 路由里 Depends(admin_dep) 忘挂 → 员工能进后台——上 11 篇的权限测试兜底
- 分页/Top-N 忘了在 SQL 过滤 user_id → 数据混串
下一篇写前端：无构建工具的原生 JS 如何撑起流式渲染、5 套主题与全套交互纪律。