第 11 章 Backend 服务参考¶

本章学习目标 - 掌握 Backend 的 API 端点全貌（Chat BI / Admin / Utility） - 理解应用生命周期、中间件栈与优雅关闭 - 了解数据库模型（ORM）组织与异常层次 - 理解生产部署的 Agent Runtime Pool 架构 - 知道去哪里查环境变量与配置（附录 B）

前置知识：第 4-8 章 AI Agent 内核 关联代码：backend/app/main.py、backend/app/api/、backend/app/db/models.py

本章定位

本章是 Backend 的服务面参考（API、生命周期、ORM 模型、部署）。AI Agent 内核（编排、RAG、SQL、记忆、工具）已在第 4-8 章详解，本章不重复。环境变量完整参考见附录 B。

11.1 技术基线¶

组件	版本/规格	用途
Python	3.12+	运行时
FastAPI	0.115+	ASGI Web 框架
LangGraph	≥ 0.4.0	状态图编排（详见第 4 章）
psycopg 3	latest	异步 PG 驱动（raw pool）
SQLAlchemy 2.0	latest	ORM（Admin 资源管理）
Pydantic v2	latest	数据校验 + Settings
structlog	latest	结构化 JSON 日志
slowapi	latest	请求限流

当前版本：4.0.0（见 backend/app/main.py），包名 ttd-backend（uv workspace member）。

11.2 应用生命周期¶

启动序列¶

应用使用 FastAPI 的 lifespan 上下文管理器进行资源初始化和清理：

sequenceDiagram
    autonumber
    box rgb(232,240,254) FastAPI 运行时
        participant App as FastAPI App
        participant PG as PG Pool
        participant MR as Model Registry
        participant CP as Checkpointer
        participant SG as Supervisor Graph
    end

    App->>PG: AsyncConnectionPool(min=5, max=20)
    opt Swarm Mode (TTD_SWARM_MODE=true)
        App->>App: InstanceRegistry + heartbeat + GracefulShutdown
    end
    App->>MR: ModelRegistry() + load_from_db()
    App->>CP: AsyncPostgresSaver.setup()
    App->>SG: build_supervisor_graph(checkpointer, store, registry, pool)
    Note over App: ✓ "TTD Chat BI started"

关闭序列（优雅关闭）¶

Swarm 模式下的优雅关闭是生产稳定性的关键：

Graceful Drain（Swarm）：停止接受新请求，等待 in-flight 完成（max 120s）
停止 heartbeat + 标记实例 shutdown
取消后台清理任务
关闭 Memory Store → Checkpointer → PG Pool → SQLAlchemy Engine

这保证了滚动更新时不会丢失正在处理的查询——HAProxy 健康检查会摘除 draining 实例，新请求路由到健康实例。

中间件栈¶

中间件	功能
CORSMiddleware	跨域（生产收紧到 BETTER_AUTH_URL）
Rate Limiter (slowapi)	按 IP 限流，60 req/min/user
Prometheus Metrics	`/metrics` 端点（`TTD_PROMETHEUS_ENABLED`）

11.3 API 端点¶

所有路由挂载在 /api/v1 前缀。

11.3.1 Chat BI API（主接口）¶

方法	路径	说明
`GET`	`/chats`	列出当前用户所有对话
`POST`	`/chats`	创建新对话
`GET`	`/chats/{id}`	获取对话详情（含消息历史）
`PATCH`	`/chats/{id}`	更新对话（标题/状态）
`DELETE`	`/chats/{id}`	删除对话
`POST`	`/chats/{id}/messages`	发送消息（非流式）
`POST`	`/chats/{id}/messages/stream`	发送消息（SSE 流式）
`POST`	`/messages/{id}/feedback`	提交消息反馈

SSE 流式事件类型（status/sql/result/viz/insights/followup 等 29 种）详见第 4 章与第 12 章。

11.3.2 Admin API¶

分组	前缀	说明
LLM Models	`/admin/llms/*`	LLM 模型 CRUD
Embedding Models	`/admin/embedding-models/*`	Embedding 模型管理
Reranker Models	`/admin/reranker-models/*`	Reranker 模型管理
Knowledge Bases	`/admin/knowledge-bases/*`	知识库 CRUD
Data Sources	`/admin/datasources/*`	数据源管理
Documents / Chunks	`/admin/documents/`、`/admin/chunks/`	文档与 chunk
Graph	`/admin/graph/*`	KG Entity/Relationship
Chat Engines	`/admin/chat-engines/*`	聊天引擎配置
Evaluations	`/admin/evaluations/*`	评估任务管理
Feedbacks	`/admin/feedbacks/*`	反馈审核
Few-Shot Review	`/admin/few-shots/*`	动态 Few-Shot 审批（详见第 7 章）
Semantic Plane	`/admin/semantic-plane/*`	语义层资产浏览
API Keys / Site Settings / Stats / Uploads	`/admin/*`	系统管理

11.3.3 Utility API¶

方法	路径	说明
`GET`	`/health`	健康检查
`GET`	`/metrics`	Prometheus 指标（可选）

11.4 数据库模型¶

ORM 模型定义在 backend/app/db/models.py，Schema：ttd（与 semlayer 语义层分离）。

分类	模型	表名
核心业务	Chat / ChatMessage / Feedback	ttd_chats / ttd_chat_messages / ttd_feedback
模型管理	LLMModel / EmbeddingModel / RerankerModel / ChatEngine	ttd_llm_models / ...
知识库	KnowledgeBase / DataSource / Document / Chunk / KGEntity / KGRelationship	ttd_knowledge_bases / ...
评估	EvaluationDataset / EvaluationTask / EvaluationTaskItem	ttd_evaluation_*
系统	Upload / SiteSettings / ApiKey	ttd_uploads / ...
记忆	MemoryRecord / MemoryEvidence	ttd_memory_records / ...

Alembic 迁移在 backend/alembic/versions/。Better Auth 表（user/session/account/verification/jwks）由 web 端 @better-auth/cli migrate 创建。

11.5 配置参考¶

所有环境变量使用 TTD_ 前缀，通过 app/config.py 的 Pydantic Settings 管理。

环境变量完整参考

环境变量的唯一权威完整列表在附录 B · 环境变量完整参考。本章不重复，只列分类摘要：

分类	示例变量	说明
服务器	`TTD_HOST`、`TTD_PORT`、`TTD_DEBUG`	服务绑定与调试
Runtime Pool	`TTD_SWARM_MODE`、`TTD_INSTANCE_ID`	多实例部署
数据库	`TTD_AURORA_DSN`、`TTD_PG_POOL_MIN/MAX`	PG Supernode 连接
数据面	`TTD_DATA_PLANE_BACKEND`、`TTD_REDSHIFT_*`	双数据面切换
模型	`TTD_DASHSCOPE_API_KEY`、`TTD_SUPERVISOR_MODEL`	LLM/Embedding 配置
认证	`TTD_BETTER_AUTH_URL`、`TTD_JWT_AUDIENCE`	Better Auth JWKS
检索	`TTD_TOP_K_*`、`TTD_SIMILARITY_THRESHOLD`	RAG 参数
执行	`TTD_STATEMENT_TIMEOUT_MS`、`TTD_MAX_ROWS`	SQL 执行限制
记忆	`TTD_SQL_CACHE_SIMILARITY`、`TTD_SHORT_TERM_WINDOW`	记忆系统
修复	`TTD_MAX_REPAIR_RETRIES`、`TTD_MAX_ESTIMATED_ROWS`	护栏阈值
监控	`TTD_PROMETHEUS_ENABLED`、`TTD_AUDIT_LOG_LEVEL`	可观测性

11.6 生产部署 — Agent Runtime Pool¶

生产环境采用 Agent Runtime Pool 模式：多个无状态实例通过 Docker Swarm 编排，HAProxy consistent hashing 实现 session affinity。

C4Deployment
title Agent Runtime Pool — Production Deployment

    Deployment_Node(haproxy_tier, "HAProxy 负载均衡", "HAProxy 2.9 · 2C/4G", "四层 TCP 反向代理，ketama 一致性哈希") {
        Container(haproxy, "HAProxy", "consistent hash · X-Chat-ID", "健康检查 5s fall 3 rise 2，实例增减仅 ~1/N 受影响。性能优化非正确性依赖")
    }

    Deployment_Node(swarm_cluster, "Docker Swarm 集群", "N×16C/32G · Overlay Network", "无状态 Runtime 水平扩展池") {
        Container(runtime_1, "Runtime #1", "FastAPI · LangGraph · 1 worker", "完整 Agent 管线，PG Advisory Lock 跨实例并发")
        Container(runtime_n, "Runtime #N", "FastAPI · LangGraph · 1 worker", "与 #1 完全一致，zero gossip，共享状态通过 PG")
    }

    Deployment_Node(pg_tier, "PG Supernode", "PostgreSQL 18 · 16C/64G · SSD", "共享状态后端：检查点、锁、缓存") {
        ContainerDb(pg_supernode, "PG Supernode", "PostgreSQL 18 · pgvector · AGE", "检查点 + 记忆 + R/V/G 语义资产 + SQL 缓存 + Auth 表")
    }

    Rel(haproxy, runtime_1, "hash(chat_id) 路由", "HTTP · ketama")
    Rel(haproxy, runtime_n, "hash(chat_id) 路由", "HTTP · ketama")
    Rel(runtime_1, pg_supernode, "读写检查点·获取锁·检索·缓存", "asyncpg · TCP")
    Rel(runtime_n, pg_supernode, "读写检查点·获取锁·检索·缓存", "asyncpg · TCP")

    UpdateElementStyle(haproxy, $fontColor="#202124", $bgColor="#e8f0fe", $borderColor="#1a73e8")
    UpdateElementStyle(runtime_1, $fontColor="#202124", $bgColor="#e8f0fe", $borderColor="#1a73e8")
    UpdateElementStyle(runtime_n, $fontColor="#202124", $bgColor="#e8f0fe", $borderColor="#1a73e8")
    UpdateElementStyle(pg_supernode, $fontColor="#202124", $bgColor="#e0f2f1", $borderColor="#0d7377")

    UpdateRelStyle(haproxy, runtime_1, $textColor="#1a73e8", $lineColor="#1a73e8")
    UpdateRelStyle(runtime_1, pg_supernode, $textColor="#0d7377", $lineColor="#0d7377")

    UpdateLayoutConfig($c4ShapeInRow="3", $c4BoundaryInRow="2")

11.6.1 关键设计决策¶

决策	选择	理由
实例间状态共享	PG Supernode（已有）	不引入新中间件（如 Redis）
并发控制	PG Advisory Lock	跨实例互斥，事务结束自动释放
路由策略	Consistent Hash (ketama)	缓存 warmth + 最小迁移（实例增减时只有 1/N session 受影响）
单 worker/instance	Docker Swarm replicas	避免 GIL 竞争、内存膨胀；水平扩展靠实例数
实例间通信	无（zero gossip）	所有共享通过 PG，简化架构

一致性哈希与 Session Affinity

HAProxy 基于 X-Chat-ID header 做 ketama 一致性哈希——同一会话的请求总路由到同一实例。这保证了缓存 warmth（SQL Cache、Metadata Cache 在同一实例命中率高）。

关键：一致性哈希是性能优化，非正确性依赖。即使请求路由到不同实例，因为有 PG Supernode 共享状态（Checkpointer + Advisory Lock），正确性仍保证。一致性哈希只是让缓存更热、延迟更低。

11.6.2 故障转移¶

节点故障：一致性哈希环自动 rehash，仅 ~1/N sessions 受影响
状态恢复：从 PG checkpoint 恢复，无状态丢失
Health check：HAProxy inter 5s, fall 3, rise 2 自动摘除/恢复节点

11.6.3 部署命令¶

task ha:up      # 一键启动 Swarm HA
task ha:down    # 一键拆除
N=5 task deploy:swarm:scale    # 扩缩容
task deploy:swarm:update       # 滚动更新
task deploy:swarm:status       # 查看实例状态

11.7 异常层次与错误处理¶

TTDError                          # 基类
├── ClarificationNeededError      # Follow-up 发现歧义
├── PolicyViolationError          # Guardrails 安全策略违规
├── SQLValidationError            # SQL 校验失败（syntax/policy/EXPLAIN）
├── RetrievalMissError            # R/V/G 三引擎均未命中
└── ModelUnavailableError         # 模型 API 不可达且无 fallback

错误通过 state["errors"] 列表传播，不直接抛异常（符合 LangGraph 的状态增量模式）。技术错误经 Sanitization 转为用户友好中文消息（如 "statement timeout" → "查询执行超时，请尝试缩小查询范围"）。

11.8 请求生命周期¶

一个 Chat BI 请求从客户端到响应的完整调用链：

sequenceDiagram
    autonumber
    box rgb(248,249,250) 外部客户端
        participant C as 客户端
        participant H as HAProxy
    end
    box rgb(232,240,254) FastAPI 运行时
        participant F as FastAPI
        participant MW as 中间件栈
        participant R as Router (chats.py)
        participant LG as LangGraph
        participant S as SSE Stream
    end
    box rgb(224,242,241) 数据面
        participant DP as DataPlane
    end

    C->>H: POST /api/v1/chats/{id}/messages
    H->>H: 一致性哈希 (X-Chat-ID) → 目标实例
    H->>F: 转发请求
    F->>MW: CORS → 速率限制 → JWT 验签 → 请求日志
    MW->>R: 认证后的 request + user
    R->>R: 加载 session + 初始化 AgenticBIState
    R->>LG: graph.astream(initial_state, config)
    loop SSE 事件流
        LG->>LG: 执行节点 (supervisor→QU→router→...)
        LG-->>R: node_complete 事件
        R-->>S: yield SSE event
        S-->>C: 流式推送（29 事件类型）
    end
    LG->>DP: SQL 执行（仅 sql_executor 节点）
    DP-->>LG: 结果集
    LG-->>R: 最终结果 + insights
    R-->>S: yield data + insights + status=completed
    S-->>C: 关闭流

中间件栈（按执行顺序）¶

顺序	中间件	职责	失败行为
1	CORS	允许前端域名跨域	—
2	slowapi 速率限制	`60/minute` per IP（第 14 章）	429 Too Many Requests
3	JWT 验签	JWKS 公钥验签 RS256 JWT	401 Unauthorized
4	请求日志	structlog 记录 request_id + user_id + 耗时	—

依赖注入¶

FastAPI 的依赖注入在 Router 层解析：

# backend/app/api/chats.py
@router.post("/{chat_id}/messages")
@limiter.limit("60/minute")
async def send_message(
    chat_id: str,
    body: MessageCreate,
    request: Request,
    user: AuthUser = Depends(get_current_user),      # JWT → user
    store: AsyncPostgresStore = Depends(get_store),   # LangGraph Store
    session: AsyncSession = Depends(get_db_session),  # ORM session
):

get_current_user 从 request 中提取 JWT → JWKS 验签 → 返回 AuthUser。get_store / get_db_session 从连接池获取。

11.9 并发模型¶

async/await 在 Agent 场景的最佳实践¶

TTD Backend 全程 async/await，但需注意 LangGraph 的并发特性：

维度	实践	原因
节点内 LLM 调用	`await llm.ainvoke()`	LLM API 是 I/O 密集，async 不阻塞事件循环
节点内 DB 查询	`await session.execute()`	asyncpg 异步驱动
并行检索	`asyncio.gather(engine_r(), engine_v(), engine_g())`	四引擎无依赖，并行召回
LangGraph 并发 run	`asyncio.create_task(graph.astream(...))`	每个用户请求一个 task
CPU 密集（嵌入计算）	`asyncio.to_thread(embed_fn)`	嵌入计算阻塞事件循环，丢到线程池

LangGraph 并发 run 的陷阱

多用户并发时，每个请求独立 graph.astream()——它们共享同一个 graph 编译对象（线程安全），但各自有独立的 AgenticBIState 副本。不要在节点函数中使用模块级可变状态（如全局缓存 dict），否则并发请求会相互污染。TTD 的所有状态通过 state 参数传递，或通过 LangGraph Store（线程安全）共享。

并发能力估算¶

资源	限制	单请求占用	理论并发上限
PG 连接池	`TTD_PG_POOL_MAX=20`	1 连接（查询时）	20
LLM API 并发	DashScope 默认 50 QPS	1 QPS（生成时）	50
事件循环	单进程单循环	~1MB 栈	~1000（理论）
实际瓶颈	PG 连接池	—	~20 并发查询

上述并发上限为设计估算。生产扩容通过 Swarm 多实例 + 每实例独立连接池实现水平扩展。

11.10 性能调优¶

连接池配置¶

# backend/app/db/session.py
engine = create_async_engine(
    settings.database_url,
    pool_size=5,           # 持久连接数（常态）
    max_overflow=15,       # 突发时可超额（总上限 20）
    pool_timeout=30,       # 等待连接超时
    pool_recycle=1800,     # 30 分钟回收（防止 PG 断连）
    pool_pre_ping=True,    # 使用前 ping（检测断连）
)

参数	值	调优依据
`pool_size`	5	常态并发查询数（开发环境）
`max_overflow`	15	突发流量缓冲，总上限 20 对应 `TTD_PG_POOL_MAX`
`pool_timeout`	30s	超时后返回 503，避免无限等待
`pool_recycle`	1800s	PG 默认 `idle_timeout` 通常 > 1h，30 分钟回收防边界
`pool_pre_ping`	True	容器网络中连接可能被静默断开

其他生产调优¶

项	配置	效果
uvloop	`uvloop.install()`（main.py）	替代默认 asyncio 事件循环，吞吐提升 2-4x
GC 调优	`PYTHON_GC_THRESHOLD=700,10,10`	减少全代 GC 频率，降低长查询尾延迟
Metadata Cache	TTL 24h	元数据查询从 PG 降到内存，省 ~5ms/请求
SQL Cache	TTL 1h + cosine 0.92	相似问题跳过检索+生成，命中时 ~250ms 返回
orjson	`ORJSONResponse`	替代标准 json，序列化快 2-3x

性能数字为设计目标

上述提升倍数为经验/基准值，未经 TTD 环境实测。系统化性能基线见第 16 章。

11.11 小结¶

本章要点：

Backend 是 FastAPI + LangGraph 服务，API 分 Chat BI（主）/ Admin / Utility 三类。
请求生命周期：HAProxy → 中间件栈（CORS/限流/JWT/日志）→ Router → LangGraph → SSE 流式返回。
并发模型：全程 async/await，四引擎并行检索；实际瓶颈是 PG 连接池（~20 并发），水平扩展靠 Swarm 多实例。
性能调优：连接池（pre_ping + recycle）、uvloop、GC 调优、Metadata/SQL Cache、orjson。
生产部署用 Agent Runtime Pool + HAProxy 一致性哈希（性能优化）+ PG Advisory Lock（正确性保证）。
环境变量完整参考在附录 B（唯一权威）。

下一章：第 12 章 Web 前端——Next.js 架构、Better Auth、SSE 契约与可视化组件。