02｜AI 智能体复杂应用架构设计

1. AI Infra 与系统能力标准

1.1 从“模型服务”扩展为“智能体运行平台”

企业级 AI Infra 通常包括：

• 模型网关：统一不同模型供应商、配额、限流、重试和降级；
• 推理服务：云模型、本地模型、Embedding、Reranker；
• Agent Runtime：状态机、任务队列、调度、检查点和人工介入；
• 知识平台：采集、清洗、索引、权限、更新和评估；
• 工具平台：注册、发现、鉴权、审计和版本管理；
• 可观测平台：Trace、Metric、Log、评估样本和成本账单；
• 安全治理：身份、最小权限、内容安全、数据隔离与密钥管理。

1.2 把非功能指标写成可验证目标

2. 五层通用架构与 Skills 边界

2.1 交互层

负责多端入口、身份、会话、文件上传、流式输出与人工确认。交互层不应直接持有数据库高权限凭据，而是通过受控服务调用。

2.2 编排层

负责工作流、动态路由、并行、循环、重试、检查点与人机协同。稳定步骤应写成确定性代码，只有真正需要语义判断的位置才交给模型。

2.3 模型层

同时管理大模型、小模型、Embedding、Reranker 和内容安全模型。典型路由策略：

• 简单分类交给小模型；
• 复杂规划交给强模型；
• 敏感输出先经过安全模型；
• 失败或低置信度时升级模型或人工。

2.4 知识层

知识层不仅是向量库，还包括原始文件、元数据、权限标签、版本、图谱、对话状态和长期记忆。任何检索结果都应保留来源 ID，以便引用和审计。

2.5 集成层

通过 API、数据库、消息队列、MCP Server 或 A2A Agent 接入外部能力。集成层统一解决超时、重试、熔断、幂等和数据脱敏。

2.6 通用 Skill 与专用 Skill

3. RAG 与上下文工程

上下文工程是为模型在正确时间提供正确的信息。上下文来源包括系统规则、用户目标、历史状态、检索证据、工具结果和输出 Schema。

3.1 一个实用的上下文优先级

1. 安全与系统规则；
2. 当前任务目标和输出约束；
3. 与任务直接相关的最新状态；
4. 检索证据与来源；
5. 必要的历史摘要；
6. 示例与补充背景。

不是信息越多越好。长上下文会增加成本和延迟，也可能让模型被过时信息干扰。常用压缩方法包括：窗口裁剪、结构化摘要、只保留未完成任务、检索式记忆和重复证据去除。

3.2 状态持久化

建议区分：

• thread_state：本次会话的消息、任务和中间结果；
• task_checkpoint：长任务暂停和恢复所需的数据；
• long_term_memory：跨会话共享的稳定事实或经验；
• audit_event：不可变更的行为记录。

4. MCP 与 A2A 协议

4.1 MCP 是什么

Model Context Protocol（MCP）是 AI 应用与外部上下文、工具之间的开放协议。它解决的是“怎样用统一方式发现和调用能力”，不是模型、Agent 框架或业务权限系统。当前课程基线为 2026-07-28 规范 + Python SDK 2.0.0；MCP 版本变化快，复制旧教程前必须先核对版本。

MCP 使用 JSON-RPC 2.0 消息，角色分为：

• Host：面向用户的 AI 应用，负责模型、同意界面、权限和多个连接；
• Client：Host 内与一个 Server 保持连接的协议组件；
• Server：提供上下文和能力，不应自行获得 Host 未授权的数据；
• 不是“模型直连数据库”：模型提出调用意图，Host 仍要校验、征得同意并通过 Client 发出请求。

用户
  ↓ 同意 / 撤销 / 查看结果
Host（模型、策略、UI、审计）
  ├─ Client A ── stdio ───────── 本机文件 MCP Server
  └─ Client B ── Streamable HTTP ─ 企业工单 MCP Server
                                      ├─ Tool：执行受控动作
                                      ├─ Resource：读取上下文
                                      └─ Prompt：取得用户可选择模板

连接要先 initialize，双方协商协议版本与能力，再由 Client 发送 notifications/initialized。之后才进行 tools/list、tools/call、resources/read 等操作；取消、进度和错误也属于协议生命周期，而不是业务函数自己发明的字段。

4.2 Tools、Resources、Prompts 怎么选

Tool 的名称、描述、输入 Schema 都属于不可信元数据；Host 不能因为描述声称“只读”就跳过权限检查。写工具还需要服务端授权、审批、幂等键和审计。Resource 返回的数据同样可能含间接 Prompt Injection，只能作为数据，不能覆盖系统指令。

4.3 传输与部署边界

旧资料中的独立 HTTP+SSE 方案不能直接当作新项目默认模板。新远程项目优先使用 Streamable HTTP；横向扩展时优先无状态 JSON 响应。stdio 不使用 HTTP OAuth 流程，凭据由 Host 以受控环境变量或系统密钥存储注入。

4.4 从零创建一个 MCP Server

本课程已经提供完整工程 repository/mcp-operations-server。核心步骤如下：

1. 用 MCPServer 创建服务并写清 instructions；
2. 将 Python 类型注解转换为 Tool 输入/输出 Schema；
3. 将查询、上下文和模板分别声明为 Tool、Resource、Prompt；
4. 对写操作在服务端检查批准状态和 request_id；
5. 用 Client(mcp) 的内存 Transport 做协议测试；
6. 本机用 Inspector 调试 stdio，远程用 Streamable HTTP；
7. 生产环境接入企业 OAuth、密钥管理、审计和告警。

from mcp.server import MCPServer

mcp = MCPServer("enterprise-operations")

@mcp.tool()
def get_equipment(equipment_id: str) -> dict:
    """读取设备；实现必须在服务端做身份和租户校验。"""
    return {"equipment_id": equipment_id, "status": "warning"}

@mcp.resource("ops://equipment/{equipment_id}")
def equipment_resource(equipment_id: str) -> str:
    return f"equipment={equipment_id}"

app = mcp.streamable_http_app()

4.5 MCP 与 A2A 的边界

两者互补：一个 Agent 可以通过 MCP 使用工具，再通过 A2A 与其他 Agent 协作。低延迟、细粒度、无自主性的能力更适合 Tool；需要独立状态、长任务和协商的服务才适合远程 Agent。

4.6 安全检查清单

• 用户知道连接了哪个 Server、会读取什么、会执行什么，并能撤销授权；
• 禁止 Token passthrough；Server 只接受颁发给自己的 Token；
• Streamable HTTP 校验 Origin、Host、受众和 scope，避免 DNS rebinding 与 confused deputy；
• 本地 Server 使用绝对路径和固定参数，Host 展示命令，防止包名劫持；
• Tool 调用设置超时、请求体上限、并发上限、幂等和结构化错误；
• 日志记录 trace、主体、工具、脱敏参数摘要、结果和耗时，不记录 Token。

5. 可观测与评估体系

OpenTelemetry 将遥测信号划分为 Trace、Metric、Log 等。智能体还要额外记录模型和知识链路指标：

trace_id
  ├─ model_call: model, tokens, latency, prompt_version
  ├─ retrieval: query, index_version, top_k, document_ids
  ├─ tool_call: tool, sanitized_args, status, duration
  ├─ guardrail: policy, decision, reason
  └─ final_answer: citations, evaluator_scores, user_feedback

5.1 三维权衡

• 性能：更强模型和更多检索通常提高质量，但增加延迟；
• 成本：缓存、路由、压缩和批处理可降本，但需防止质量下降；
• 安全：审批和隔离增加步骤，却能显著降低不可逆风险。

5.2 模型路由与熔断示例

完整文件：examples/02_architecture_gateway.py。

from dataclasses import dataclass
import time

@dataclass
class ModelRoute:
    name: str
    max_complexity: int
    cost_weight: float

ROUTES = [
    ModelRoute("small-local", 2, 0.1),
    ModelRoute("general-cloud", 6, 1.0),
    ModelRoute("reasoning-cloud", 10, 3.0),
]

def choose_model(complexity: int, sensitive: bool) -> ModelRoute:
    candidates = [r for r in ROUTES if r.max_complexity >= complexity]
    if sensitive:
        candidates = [r for r in candidates if r.name != "general-cloud"]
    if not candidates:
        raise ValueError("没有满足策略的模型")
    return min(candidates, key=lambda x: x.cost_weight)

class CircuitBreaker:
    def __init__(self, limit: int = 3, cooldown: float = 10):
        self.limit, self.cooldown = limit, cooldown
        self.failures, self.opened_at = 0, 0.0

    def allow(self) -> bool:
        return self.failures < self.limit or time.time() - self.opened_at > self.cooldown

    def record_failure(self) -> None:
        self.failures += 1
        if self.failures == self.limit:
            self.opened_at = time.time()

    def record_success(self) -> None:
        self.failures = 0

6. 应用案例：企业知识助手架构

6.1 需求

员工询问制度问题，系统需要检索最新制度、展示来源、根据部门权限过滤，并在低置信度时转人工。

6.2 设计步骤

1. 交互层完成登录、部门身份和流式答案；
2. 编排层执行“问题分类 → 权限过滤 → 检索 → 重排 → 生成 → 引用检查”；
3. 模型层采用小模型分类、强模型生成；
4. 知识层保存文件版本、失效时间和部门 ACL；
5. 集成层连接 HR 工单系统；
6. 可观测层跟踪每个答案使用的文档版本。

6.3 降级策略

• 向量库不可用：降级到关键词检索；
• 强模型不可用：使用小模型生成简短摘录；
• 无可靠证据：不回答结论，转人工；
• 工单系统不可用：保存待办并通知用户稍后处理。

6.4 项目复现：企业设备 MCP Server

cd repository/mcp-operations-server
python -m venv .venv
# Linux/macOS: source .venv/bin/activate
# PowerShell:   .\.venv\Scripts\Activate.ps1
python -m pip install -e ".[test]"
pytest -q
mcp dev src/mcp_operations_server/server.py

Inspector 中测试 get_equipment、search_runbook、设备 Resource、分诊 Prompt，以及“未批准不得创建工单、相同 request_id 不重复创建”。远程模式运行：

uvicorn mcp_operations_server.server:app --host 127.0.0.1 --port 8000
npx -y @modelcontextprotocol/inspector

连接 http://127.0.0.1:8000/mcp。README.md 给出 Host 配置和知识点映射，DEPLOY.md 给出 Docker、systemd、Nginx、HTTPS、认证、监控与回滚步骤。验收时必须同时展示一次只读成功、一次未批准写入拒绝、一次批准写入和一次幂等重放。

7. 可靠参考资料

• MCP Architecture Overview理解 Host、Client、Server 以及 Tools、Resources、Prompts。
• MCP 2026-07-28 Specification当前协议的权威版本、生命周期、能力与安全要求。
• MCP Python SDK本章工程使用官方 SDK 2.0.0；仓库包含 Server、Client 和协议测试示例。
• MCP Security Best PracticesToken passthrough、confused deputy、DNS rebinding、会话与本地 Server 风险。
• Anthropic：Model Context Protocol 官方视频配合规范理解核心组件、连接方式与生态；协议细节仍以当前规范为准。
• A2A Protocol Specification了解 Agent Card、任务与消息互操作。
• A2A and MCP Comparison官方边界说明。
• OpenTelemetry Observability PrimerTrace、Metric 与 Log 的统一观测基础。
• AgentBench: Evaluating LLMs as Agents （ICLR 2024）Paper and venue are linked above. Code: https://github.com/THUDM/AgentBench.
• τ-bench: A Benchmark for Tool-Agent-User Interaction in Real-World Domains （ICLR 2025）Paper and venue are linked above. Original code: https://github.com/sierra-research/tau-bench.