AI项目实战（3）通用脚手架：搭建一个能跑十个项目的工程底座

老张是团队里能扛活的骨干。第一次接 Agent 项目，他从零搭了一套——配置、日志、重试、API 路由，吭哧吭哧写了一周。第二次接项目，他翻出上次的代码看了又看，觉得几处不顺手，索性又重写了一套。第三次、第四次……等到第十个项目落地时，他面对十个长得像、却又不完全一样的 config.py，欲哭无泪：线上发现一个重试的 bug，得改十个地方；某个项目的日志漏了 trace_id，排查时两眼一抹黑；新同事入职，光是搞清楚"这本书里到底有几种配置写法"就花了三天。

"早知道，我一开始就该把这副底座搭好。"老张盯着满屏的复制粘贴，喃喃自语。

这一章，我们就来帮老张——也帮未来的你——把这件事一次做对。我们要搭的不是某个具体项目，而是一副能跑十个项目的工程脚手架。它像盖楼用的预制构件：先在工厂里标准化生产好，运到工地一拼，楼就立起来了。后面十个项目，都站在它肩膀上，新增成本几乎为零。

全书主线进度：第 1 章看清了全图——八张核心拼图、十条进化路径。从本章开始，我们把这些认知变成能跑的代码。第一步不是直接写 Agent，而是先把工程底座搭好——这个底座会让后面十个项目的新增成本几乎为零。

本章对应代码：backend/core/、backend/main.py

2.1 老张的烦恼

2.1.1 业务背景与痛点

回到老张的十个项目。如果每个都独立搭建——配置、日志、Agent 循环、错误处理、API 路由各写一套——会冒出三类毛病：一是重复代码堆积如山；二是十个项目十种风格，改一个 bug 要满世界找；三是读者翻书时也犯嘀咕：“为什么项目一和项目二的配置方式都不一样？到底听谁的？”

根子上的问题只有一个：没有一副共享的骨架。这一章，我们就把这副骨架立起来。

2.1.2 用户故事

把上面的痛点翻译成四条用户故事，每一条都对应一个真实的"谁、想要什么、图什么"：

2.1.3 功能性需求

• FR-1 配置管理：集中式配置，环境变量优先，.env 兜底，启动时验证必需项
• FR-2 结构化日志：开发环境人类可读格式，生产环境 JSON 行格式，自动注入 trace_id
• FR-3 Agent 基类：封装重试、超时、流控、会话管理、指标收集
• FR-4 项目注册表：导入即注册，后端自动发现并暴露 API
• FR-5 统一 API：健康检查、指标暴露、运行（同步）、流式运行（SSE）
• FR-6 错误兜底：工具调用失败、模型超时、并发满、流控拒绝——全部优雅降级，不崩溃

2.1.4 非功能性需求

• 单次 Agent 调用超时 120 秒自动终止
• 模型调用失败最多重试 3 次（指数退避）
• 每会话每分钟最多 60 次请求
• 全局最多 10 个并发 Agent 执行
• 会话空闲 1 小时自动过期

2.2 画个样子：它该长啥样

2.2.1 架构总览

%%{init: {'theme':'base','flowchart':{'useMaxWidth':true,'htmlLabels':true}}}%%
graph TD
    subgraph Core["core 公共层（全书复用）"]
        Cfg["config.py
配置管理 + 启动验证"]
        Log["logging_conf.py
结构化日志 + trace_id"]
        AC["agent_core.py
BaseProject + 注册表 + 流控 + 指标"]
    end
    subgraph Projects["projects 项目层（第3-12章）"]
        P1["p01 客服"]
        P2["p02 知识库"]
        Pn["... p10"]
    end
    subgraph Serve["服务层"]
        API["main.py
FastAPI + 健康检查 + 指标 + SSE"]
    end
    P1 & P2 & Pn -->|继承 + 注册| AC
    API --> AC
    Cfg & Log --> AC

    classDef coreN fill:#ede7f6,stroke:#5e35b1,stroke-width:2px,color:#311b92
    classDef projN fill:#e0f2f1,stroke:#00897b,stroke-width:2px,color:#004d40
    classDef serveN fill:#e3f2fd,stroke:#1976d2,stroke-width:2px,color:#0d47a1
    class Cfg,Log,AC coreN
    class P1,P2,Pn projN
    class API serveN

核心设计思想：每个项目只需继承 BaseProject、实现 build_agent()，就自动获得重试、超时、流控、会话管理、指标、日志、API 路由。新增项目零样板。

这里有个类比特别贴切。BaseProject 就像公司发给新员工的入职手册：你不用自己琢磨考勤怎么打、报销怎么贴、邮件怎么发——手册里都写好了，照着走就行。新员工（新项目）要做的事只有一件：填好自己的岗位职责（实现 build_agent()），其余的重试、超时、流控、指标、日志、API 路由，公司流程（基类）全包了。

而 ProjectRegistry 就像公司的花名册：新员工入职那天到 HR 报个到（模块被 import），名字自动登记进系统（注册表），前台（main.py）立刻就能查到、能调度，无需手动改任何路由配置。

2.3 拆开看：怎么造出来

2.3.1 技术选型

选型表看着平淡，但每一行背后都有一段"为什么不是它"的故事。挑几个最容易踩坑的，在下文掰开揉碎说。

2.3.2 关键设计决策

决策 1：为什么用 create_agent 而不是 create_react_agent？

这里有个反直觉点，值得讲透。

很多人写 LangChain 习惯了 create_react_agent，书里、博客里到处都是它的身影。可当你翻开官方 v1.0 文档，会看到一行扎眼的标注：langgraph.prebuilt.create_react_agent 已被标记为 deprecated。官方推荐统一使用 langchain.agents.create_agent。

心路历程是这样的：一开始你也许会犹豫——“我用惯了 create_react_agent，网上的例子都是它，换掉会不会出一堆兼容问题？” 但 deprecated 就是 deprecated，今天不换，明天 LangChain 升个版本，你的代码就开始冒警告，后天可能直接报错。与其被动挨打，不如一开始就跟随官方推荐。本书全书统一使用 langchain.agents.create_agent，后面十个项目无一例外。

⚠️ 避坑：在网上搜到老教程用 create_react_agent 时，心里要有个数——那多半是 v1.0 之前的写法，照抄会吃到 deprecation 警告。

决策 2：为什么自己封装重试/超时，而不是依赖 LangChain 内置？

LangChain 确实有内置重试机制，省心。但它的重试策略是全局的——一配全配，没法按项目粒度定制。客服项目希望重试 3 次、知识库项目只想重试 1 次（毕竟每次重试都是要花钱的 Token），内置机制做不到。于是我们自研一层薄薄的封装，换来三个自由度：

• 按项目设置不同的重试次数和退避策略
• 重试过程中记录结构化日志（第几次失败、睡了多久、下次几点再试）
• 最终失败时返回用户友好的兜底文案，而非异常堆栈

💡 顿悟时刻：框架的内置功能不是越省心越好。当你的需求开始"分项目、分场景"时，一层薄薄的自封装，往往比和框架的默认行为较劲要轻松得多。

决策 3：为什么用 SQLite 做短期数据，而不是全内存？

全内存最快，但服务一重启，所有会话记忆灰飞烟灭。SQLite 的好处是"零配置、零运维"——不用装数据库服务，一个文件搞定，适合学习和单机部署。等你的项目真要上生产了，换成 PostgreSQL 只需改一行连接字符串，业务代码一行不用动。

这就叫给未来留一扇门。

2.3.3 模块职责

四个模块、四份职责，泾渭分明。接下来挨个看实现。

2.4 动手写：搭起脚手架

2.4.1 配置管理（config.py）

代码清单 2-1：生产级配置（节选）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env", env_file_encoding="utf-8", extra="ignore"
    )

    # —— 环境 ——
    environment: Literal["dev", "staging", "production"] = "dev"

    # —— 模型 ——
    anthropic_api_key: str = Field(..., min_length=1)
    default_model: str = "claude-sonnet-4-6"
    fast_model: str = "claude-haiku-4-5"
    temperature: float = Field(default=0.0, ge=0.0, le=1.0)

    # —— 模型调用控制 ——
    max_tokens: int = 4096
    request_timeout_seconds: float = 60.0
    max_retries: int = 3
    retry_backoff_base: float = 1.5

    # —— Agent 循环控制 ——
    max_agent_steps: int = 20
    agent_step_timeout_seconds: float = 120.0

    # —— 会话 ——
    max_sessions: int = 10000
    session_ttl_seconds: int = 3600

    # —— 流控 ——
    rate_limit_requests_per_minute: int = 60
    max_concurrent_agent_runs: int = 10

设计要点：

• 用 pydantic.Field 给每个配置项加上 ge/le 约束，防止非法值进入系统
• 用 @field_validator 确保生产环境必须配置 API Key
• 用 @lru_cache 做单例，全局只加载一次

配置项看着枯燥，但每一个都对应一个真实的"防呆"场景。

temperature 加了 ge=0.0, le=1.0，不是多此一举。曾经有人在生产环境把 temperature 配成了 10，Agent 输出直接放飞自我，回答像喝醉了酒。一行约束，把这种事故堵在启动那一刻。

@field_validator 在生产环境强制要求 API Key——宁可启动时就直接报错，也别等线上第一个请求打过来才发现"哎呀忘配了"。

@lru_cache 做单例——配置只读一次 .env，全局共享同一份。既省 IO，又保证大家看到的是同一套参数，不会出现"这个模块用的是旧配置"的诡异 bug。

⚠️ 避坑：配置的"启动时验证"是性价比最高的防线。许多线上事故，根因都是"某个值该配没配、配错了"。用类型 + 约束 + 校验器三道关，能把八成的问题挡在服务跑起来之前。

2.4.2 结构化日志（logging_conf.py）

代码清单 2-2：结构化日志（节选）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

class StructuredFormatter(logging.Formatter):
    """JSON 行格式化器（生产环境）。"""
    def format(self, record):
        payload = {
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S", ...),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        for key, ctx in (("trace_id", trace_id_ctx), ...):
            val = ctx.get()
            if val:
                payload[key] = val
        return json.dumps(payload, ensure_ascii=False)


class HumanFormatter(logging.Formatter):
    """人类可读格式化器（开发环境）。"""
    def format(self, record):
        # 自动注入 trace_id / project_id / thread_id
        ...

设计要点：

• 开发环境用人类可读格式（带时间戳、级别、trace_id），生产环境输出 JSON 行（便于日志聚合到 ELK/Loki）
• 用 contextvars 实现请求级别的上下文自动注入——每条日志自动带上 trace_id，无需手动传参

日志最容易写成"两眼一抹黑"。开发时人读，要好看；生产时机器读，要结构化。一个 Formatter 搞不定两件事，于是我们准备了两位"翻译官"：HumanFormatter 伺候开发者，StructuredFormatter 伺候 ELK/Loki。

但最妙的是 contextvars 这一步。想象一下：一个请求进来，我们给它发一张"工牌"（trace_id），这张工牌挂在 contextvars 上。从此这个请求处理过程中打的每一条日志，都会自动带上这张工牌——不用你手动 logger.info(f"trace_id={xxx} ...") 满世界传参。排查问题时，拿工牌一搜，整条链路的日志齐刷刷全出来了。

💡 顿悟时刻：contextvars 是异步时代传上下文的正解。它和线程局部变量（thread-local）长得像，但在 async/await 下能正确穿透，不会把 A 请求的 trace_id 串到 B 请求里。在线程池里跑同步代码、在事件循环里跑异步代码，它都能对得上号。

2.4.3 Agent 基类（agent_core.py）

代码清单 2-3：BaseProject 核心方法（节选）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

class BaseProject(ABC):
    id: str = ""
    name: str = ""
    description: str = ""
    capabilities: list[str] = []

    def __init__(self):
        self._agent = None
        self._metrics = AgentMetrics()
        self._rate_limiter = None

    @abstractmethod
    def build_agent(self) -> Any:
        """子类实现：构建并返回一个可执行 agent。"""

    def run(self, message: str, thread_id: str = "default") -> str:
        # 1. 流控检查
        if not self.rate_limiter.check(thread_id):
            return "请求过于频繁，请稍后再试。"

        # 2. 并发检查
        if self._metrics.active_runs >= settings.max_concurrent_agent_runs:
            return "系统繁忙，请稍后再试。"

        # 3. 上下文注入（trace_id 自动加入日志）
        with _agent_context(self.id, thread_id):
            self._metrics.start_run()

            # 4. 重试循环（指数退避）
            for attempt in range(settings.max_retries + 1):
                try:
                    result = self.agent.invoke(
                        {"messages": [{"role": "user", "content": message}]},
                        config={"configurable": {"thread_id": thread_id}},
                    )
                    break
                except Exception as e:
                    if attempt < settings.max_retries:
                        time.sleep(settings.retry_backoff_base ** attempt)
                        continue
                    raise

            self._metrics.end_run(duration_ms, tool_calls)
            return self._extract_text(result)

设计要点：

• 流控：per-thread 滑动窗口，超过限制直接拒绝（不调用模型，省 Token）
• 并发保护：超过 max_concurrent_agent_runs 时返回友好提示，防止雪崩
• 重试：指数退避（1.5^n 秒），最多重试 3 次，每次重试前记录 warning 日志
• 兜底：最终失败返回用户友好的中文错误提示，而非框架异常堆栈
• 指标：每次执行自动记录耗时、工具调用次数、错误次数

run() 是整个脚手架的心脏。血在五个腔室里依次流过：流控 → 并发 → 上下文 → 重试 → 指标。我们逐个看。

流控这步最反直觉，值得多聊两句。很多人的第一反应是：“用户请求来了，凭啥拒绝？服务行业不是应该尽量满足吗？” 但想想餐厅叫号：如果不管多少人涌进来都立刻上菜，后厨先崩、食材耗尽，最后所有人都吃不上。流控就是门口那个发号的——超过窗口容量，直接请你稍等，连后厨都不惊动。体现在代码里：rate_limiter.check() 返回 False 时，我们直接返回提示，根本不调用模型。这一步省下的，全是真金白银的 Token。

⚠️ 避坑：限流一定要在"调用模型之前"。如果先调模型再限流，钱已经花了，限流就成了摆设。

并发保护是流控的兄弟：流控管"频率"（每分钟几次），并发管"同时"（此刻几个在跑）。超过 max_concurrent_agent_runs，新请求被温柔劝退，防止雪崩——一个倒、压一片。

重试用指数退避（1.5^n 秒）：第一次失败睡 1.5 秒，第二次睡 2.25 秒，第三次睡更久。为什么不是固定间隔？因为瞬时故障往往需要一点恢复时间，越等越该多等一会儿——就像被拒绝后逐渐拉长再约的间隔，既不骚扰对方，又给自己留机会。

兜底这一步，体现的是对用户的体面：模型真挂了，别把一长串 Python 异常堆栈甩到用户脸上，人家看不懂也不关心。返回一句"系统繁忙，请稍后再试"，体面地收场。

指标是给运维的眼睛：每次执行自动记下耗时、工具调用次数、错误次数。出了问题，仪表盘上一看便知。

金句：好的基类，是把"该有的体面"都替子类想好了——子类只管业务，杂事基类全兜。

三层架构总则：每个项目都必须这样组织

BaseProject 解决了"跨项目复用"的问题，但单个项目内部的代码该怎么组织？如果任由大家自由发挥，三个月后 service.py 就会膨胀成两千行的怪物，里面塞着 Prompt、工具、数据库访问、API 路由……改一行抖三抖。所以本书所有项目，在 BaseProject 之上再立一条铁律：三层架构，确保代码边界清晰、职责单一、便于测试和复用。

%%{init: {'theme':'base','flowchart':{'useMaxWidth':true,'htmlLabels':true}}}%%
graph TD
    subgraph Entry["入口层（Interface Layer）"]
        P["project.py
BaseProject 子类
API 入口 + 注册"]
    end
    subgraph App["应用服务层（Application Layer）"]
        S["service.py
Agent 编排 + 业务流程"]
        G["graph.py
LangGraph StateGraph
复杂流程编排"]
    end
    subgraph Domain["领域与基础设施层（Domain & Infrastructure Layer）"]
        M["models.py
领域模型 + 状态定义"]
        Pr["prompts.py
Prompt 模板 + 系统提示"]
        T["tools.py
工具定义 + API 封装"]
        R["repositories.py
数据访问 + 向量库"]
        Mem["memory.py
短期/长期记忆管理"]
        Mcp["mcp.py
MCP 工具服务器配置"]
    end
    
    P --> S
    P --> G
    S --> M & Pr & T & R & Mem & Mcp
    G --> M & Pr & T & R & Mem & Mcp

    classDef entryN fill:#fff3e0,stroke:#f57c00,stroke-width:2px,color:#e65100
    classDef appN fill:#e8f5e9,stroke:#43a047,stroke-width:2px,color:#1b5e20
    classDef domainN fill:#e3f2fd,stroke:#1976d2,stroke-width:2px,color:#0d47a1
    class P entryN
    class S,G appN
    class M,Pr,T,R,Mem,Mcp domainN

分层职责明细

依赖方向约束

必须严格遵守：上层可以依赖下层，下层绝不可以反向依赖上层。

✅ 允许：

• project.py 导入 service.py
• service.py 导入 models.py、tools.py、prompts.py

❌ 禁止：

• tools.py 导入 service.py
• models.py 引用 project.py
• prompts.py 调用 service.py 中的函数

这条依赖方向，可以用公司组织架构来记：前台（入口层）可以叫部门（应用层）干活，部门可以调用仓库（领域层）里的东西；但仓库里的螺丝钉不能反过来指挥前台，也不能越级去敲 CEO 的门。 一旦下层反向依赖上层，依赖关系就成了死结，改一处牵一片，测试更是无从下手。

💡 顿悟时刻：分层不是为了好看，而是为了"变"。当你要把 SQLite 换成 PostgreSQL，只需改 repositories.py，上层一行不动——这就是分层换来的自由度。

2.4.4 统一 API（main.py）

代码清单 2-4：FastAPI 统一入口（节选）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

app = FastAPI(title="AI Agent 实战 · 统一后端", version="1.0.0")

# 请求日志中间件
@app.middleware("http")
async def _log_requests(request: Request, call_next):
    ...

@app.get("/api/health")
def health():
    """健康检查：返回各项目状态（活跃运行数、错误率等）。"""
    ...

@app.get("/api/metrics")
def metrics():
    """全局指标：总运行次数、总错误数、总工具调用数。"""
    ...

@app.get("/api/projects")
def list_projects():
    """列出所有项目，含实时指标快照。"""
    ...

@app.post("/api/projects/{project_id}/stream")
async def stream_project(project_id: str, req: RunRequest):
    """流式运行（SSE），供前端实时可视化。"""
    ...

main.py 是整栋楼的大堂。不管你是来体检的（健康检查）、来看仪表盘的（指标）、来查花名册的（项目列表），还是来办事的（流式运行），都从这扇门进。一个 FastAPI 实例，把所有项目统一暴露成 API，前端只认这一套接口，不必为每个项目单独对接。

2.5 跑一跑：它真的行吗

脚手架是十个项目的地基，地基不牢，地动山摇。所以下面三类测试不是走形式，而是给地基做承重检测：配置能不能挡住非法值、日志能不能产出合格的 JSON、流控能不能在该拒时拒、该放时放。每一条 assert，都是一道验收线。

2.5.1 配置验证测试

1
2
3
4
5
6
7

def test_config_validation():
    """生产环境缺少 API Key 时应拒绝启动。"""
    from core.config import Settings
    settings = Settings(
        environment="production", anthropic_api_key="")
    with pytest.raises(ValidationError):
        Settings.model_validate(settings.model_dump())

2.5.2 日志格式测试

1
2
3
4
5
6
7
8
9
10
11

def test_structured_log_format():
    """生产环境日志应为 JSON 行。"""
    from core.logging_conf import StructuredFormatter
    import logging
    formatter = StructuredFormatter()
    record = logging.LogRecord(
        "test", logging.INFO, "", 0, "hello", (), None)
    output = formatter.format(record)
    parsed = json.loads(output)
    assert parsed["level"] == "INFO"
    assert parsed["message"] == "hello"

2.5.3 流控测试

1
2
3
4
5
6
7

def test_rate_limiter():
    from core.agent_core import RateLimiter
    rl = RateLimiter(max_requests_per_minute=2)
    assert rl.check("user1")  # 第 1 次放行
    assert rl.check("user1")  # 第 2 次放行
    assert not rl.check("user1")  # 第 3 次拒绝
    assert rl.check("user2")  # 不同用户不受影响

2.6 送上线：让它上班

2.6.1 本地开发

1
2
3
4

cd backend
cp .env.example .env   # 编辑填入 ANTHROPIC_API_KEY
pip install -r requirements.txt
uvicorn main:app --reload --port 8000

2.6.2 Docker 部署

1
2

export ANTHROPIC_API_KEY="sk-ant-..."
docker compose up -d

2.6.3 生产环境 Checklist

本地跑通只是起点，上生产才是真考验。下面这份 Checklist，每一条都对应一个"血泪教训"：有人把 API Key 写进镜像推到公开仓库，有人忘了限流被一个脚本刷爆账单，有人 CORS 开成 * 被人跨站调用……照着勾，把这些坑一个个填平。

• [ ] 设置 ENVIRONMENT=production
• [ ] 配置有效的 ANTHROPIC_API_KEY（通过环境变量，不写进镜像）
• [ ] 限制 CORS_ORIGINS 为具体域名（非 *）
• [ ] 接入日志聚合系统（ELK / Loki）
• [ ] 配置 Prometheus 抓取 /api/metrics
• [ ] 设置合理的 MAX_CONCURRENT_AGENT_RUNS 和 RATE_LIMIT_REQUESTS_PER_MINUTE

2.7 回头看：学到了什么

2.7.1 用到的能力回顾

2.7.2 常见坑

1. 忘记配置 API Key 导致生产环境启动失败——第 2.6.3 节生产环境 Checklist 已覆盖
2. 重试次数设为 0——建议至少保留 1 次重试，模型 API 偶尔会有瞬时故障
3. 流控阈值设太高——默认 60 次/分钟是针对客服场景，高频场景需上调

2.7.3 可扩展方向

• 接入 Redis 做分布式流控（当前是单机内存）
• 接入 LangSmith 做分布式追踪
• 支持多模型供应商（OpenAI/Google），通过配置切换

回头看，老张的十个项目本不必各自为政。一副共享的脚手架，把配置、日志、基类、API、流控、指标一次做对，后面十个项目就站在了同一副骨架上——新增一个项目，继承 BaseProject、实现 build_agent()，收工。

📌 第 2 章（脚手架）完成。 我们搭建了一个能承载十个项目的生产级工程底座——配置、日志、基类、API、流控、指标。下一章，我们用这个底座构建第一个真实项目——智能客服 Agent。

• ← Previous Post
• Next Post →