企业官网建设流程全解析

先泼一盆冷水：把一张 LLM API 的调用封装进while True，那不叫 AI 引擎，那叫脚本。我在把开源书第三章从 50 行最小循环改写成生产级 AI 引擎的过程中，最深的感受是——真正难的从来不是“让模型回一句话”，而是当这句话需要被审计、被重试、被并行调用、被多 Agent 共享、被业务方当成一个稳定服务来依赖时，整个系统该怎么撑住。这一章我写了很多轮，删掉的草稿比留下的还多，因为“工程化”三个字听起来很虚，落地时全是细节。如果你正在做 AI Agent、RAG 服务或者任何带“循环”的 LLM 应用，这篇内容基本就是你从 Demo 走向上线时会撞上的那堵墙。

大脑 —— AI 引擎的工程化：从 50 行最小循环到生产级 AI 引擎（开源书第三章）

1. 50 行最小循环：它是怎么运作的，又是从哪里开始不够用的

1.1 最小循环的“最小”到底指什么

很多教程会让你先跑通一个 ReAct 风格的循环，看起来大概长这样：

messages = [{"role": "system", "content": SYSTEM_PROMPT}] while True: response = client.chat.completions.create( model="gpt-4o", messages=messages, tools=TOOLS, ) message = response.choices[0].message messages.append(message) if not message.tool_calls: break for tool_call in message.tool_calls: result = execute_tool(tool_call.function.name, tool_call.function.arguments) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result, }) print(messages[-1].content)

这段代码的核心逻辑只有三件事：把对话历史累积起来、把模型输出的工具调用发给执行器、把执行结果再喂回去。它能跑，但注意它的前提条件——

1. messages存在内存里，进程一重启，Agent 就失忆。
2. client、TOOLS、SYSTEM_PROMPT全部硬编码在作用域里。
3. 没有重试、没有超时、没有并发控制，模型一旦返回畸形参数，循环直接崩。
4. 过程中没有任何日志，出问题只能靠print复盘。

我不是说这段代码“错”。作为教学演示，它恰好展示了 Agent 循环的骨架。但如果你直接拿它接业务，第一周可能没事，第二周用户开始反馈“回答到一半失败了”“同一个问题每次答案组织方式都变”“工具调用偶尔没生效”，你就知道问题出在哪了——所有状态都在进程里，所有错误都靠运气。

1.2 从“能跑”到“能上线”，缺的不是 AI 能力而是系统思维

我在书里把 50 行循环比作“一颗在培养皿里的心脏”：它能跳动，但它没有被连接到血管、神经和监测设备上。生产级 AI 引擎要做的事情，不是说把模型从 GPT-4o 换成更强的模型，而是围绕这颗心脏搭建一整套生命支持系统。

具体拆开来看，主要有这么几层差距：

这一章接下来的内容，就是逐项把这层差距补起来。但先说清楚：我不会给你一个“终极框架”，因为 AI 引擎的工程化和传统后端最大的不同是——模型行为本身有概率性，所以你设计系统时不能假设“输入相同，输出就相同”。所有组件都要为这一点留出冗余。这是贯穿本章的一条暗线。

2. 记忆与状态持久化：让 AI 引擎不再“每次重开都是新同事”

2.1 短期记忆、长期记忆与业务数据的边界

很多人在 Agent 工程化时把“记忆”简单理解为“把聊天记录存数据库”。真做起来你会发现，存储本身很容易，难的是决定什么该存、什么不该存、存了之后什么时候该被想起。

我习惯把记忆分成三层：

• 短期记忆（会话级）：当前任务上下文，比如用户连续追问时前几轮的问答。它服务于模型对“当前在聊什么”的理解，通常存放在 Redis 或内存里，TTL 设为几十分钟即可。
• 长期记忆（用户级/实体级）：跨会话的关键信息，比如用户偏好、项目背景、历史决策。它需要被检索出来注入到上下文中，因此要用向量库或支持向量检索的数据库。
• 业务事实（系统级）：来自业务数据库的实时数据，比如订单状态、库存数量。Agent 不具备“记忆”这些的能力，应该通过工具调用去查询，而不是试图存进记忆里。

一个典型的错误是：把业务数据一股脑写进长期记忆。我在书里举过一个例子——客服 Agent 把用户的订单信息全部塞进向量库，结果用户改地址之后，Agent 还能检索到旧地址，造成严重事故。记忆的正确用法是只存“抽象出来的结论”或“无法通过工具获取的信息”，一切有权威来源的数据都应通过工具动态获取。

2.2 外置化存储的具体落地方案

以 50 行循环里的messages为例，工程化的第一步是把它从内存变量变成可重建的状态。我在这里给一个最小可行的设计：

# 伪代码：基于 Redis + Postgres 的消息持久化 class ConversationStore: def __init__(self, redis, pg): self.redis = redis # 短期：热数据 self.pg = pg # 长期：冷数据 + 审计 async def append(self, conversation_id: str, message: dict): # 追加消息，裁剪过长的历史 key = f"conv:{conversation_id}:messages" await self.redis.rpush(key, json.dumps(message)) await self.redis.ltrim(key, -MAX_MESSAGES, -1) async def snapshot(self, conversation_id: str) -> list[dict]: # 重建上下文时，先从热存储取，缺失再回源冷存储 ...

这里有两个工程决策值得展开：

1. 为什么不把所有消息都永久存 Redis？成本。Token 量大之后 Redis 内存会迅速膨胀，而且纯列表结构很难做语义检索。所以短期层只放最近 N 轮消息，更早的消息要么压缩成摘要，要么转存到数据仓库做离线分析。

2. 为什么消息要持久化而不仅是缓存？因为 AI 引擎的每一次对话都可能是业务事件的一部分——客服工单、法律咨询、诊断记录。你需要能完整回放“模型当时看到了什么、做了什么决策”，这就是可审计性。生产级系统和 Demo 的显著区别就是：你是否能在事故发生后完整还原现场。

2.3 上下文窗口管理：模型不是垃圾桶，不能什么都装

生产环境里最常见的上下文问题是“越聊越长，最后爆掉 token 上限”。很多人的第一反应是“那就换成上下文更长的模型呗”，这其实是典型的逃避型工程决策。上下文窗口的扩大，确实能推迟问题，但也会带来两个副作用：更贵的成本、更慢的响应、以及注意力分散导致的效果下降。

我在工程化中采用的策略按优先级排序：

1. 对话历史折叠：早期轮次不逐字保留，而是由模型或规则生成一段摘要，保留关键实体、决策结论和待办事项。
2. 消息级 TTL：超过 N 轮的普通闲聊消息直接丢弃，只保留最后几轮和系统关键信息。
3. 向量召回注入：当业务场景需要引用“很久之前说过的话”时，不把全部历史塞进去，而是把历史转成向量，根据当前 query 召回 top-k 条相关片段。
4. 结构化记忆区：在 system prompt 里专门留出一块“已知事实”区域，比如用户偏好：中文/简洁回复；用户公司：XX科技，每轮结束由模型或规则更新这块内容。

我见过很多团队在这上面栽跟头：他们选择把“所有历史”无脑传给 200K 上下文的模型，结果单次调用成本翻了 10 倍，响应延迟从 1 秒涨到 5 秒，而且模型对最新问题的回答质量肉眼可见地下降。上下文不是越大越好，而是越聚焦越好。这也是工程化和调 API 之间最大的思维差异：前者永远在做资源约束下的优化，后者只关心“能不能跑”。

3. 事件通道与工具协议：把模型调用从主流程中剥离出来

3.1 为什么需要事件通道，而不是直接 while 循环里同步调用

当你的 AI 引擎只服务自己一个脚本时，同步调用没问题。但当它要服务多个业务方、多个 Agent、多个并发会话时，同步循环就成了瓶颈。这时代码结构需要从“一个 while 循环”演进为“事件驱动的处理管线”。

我把这个演进理解为三个阶段的递进：

1. 阶段一（单体循环）：50 行最小循环，串行处理一个会话。
2. 阶段二（异步任务化）：每个会话的处理封装成一个 Task，扔进队列，Worker 异步消费。这样同一时刻几十个会话都能推进，但每个会话内部仍然是一个循环。
3. 阶段三（事件驱动）：完整的 AI 引擎不只是一个循环，而是多个协作组件通过事件通信。比如“用户消息到达”触发“意图识别”，进而触发“工具调用”，每次工具返回又触发“模型续写”。

事件通道的引入，解决的不只是并发问题，更重要的是让系统有了扩展点。你可以在“模型准备调用工具”和“工具返回结果”之间插入鉴权、限流、缓存、日志、监控、人工审批等横切逻辑，而这些在同步循环里只能写成一坨 if-else。

3.2 工具协议：比“能调用”更重要的是“能被可靠地调用”

ReAct 循环里的execute_tool太理想化了。生产环境中，工具调用的可靠性至少涉及五个层面：

# 一个生产级的工具执行器，关注点拆分 class ToolExecutor: async def execute(self, tool_call: ToolCall, context: ExecutionContext) -> ToolResult: registry = self.registry.get(tool_call.function.name) # 1. 入参校验：模型可能返回非法 json，也可能缺少必填字段 validated_args = registry.schema.validate(tool_call.function.arguments) # 2. 权限控制：校验该会话是否有权调用此工具、是否有权访问这些资源 await self.authorizer.check(context.principal, registry.name, validated_args) # 3. 超时控制：工具调 API 可能永远不返回 async with timeout(registry.timeout_seconds): # 4. 幂等控制：同一次 tool_call_id 重复执行要能安全跳过 if await self.idempotency.is_processed(tool_call.id): return await self.idempotency.get_result(tool_call.id) try: result = await registry.fn(validated_args, context) # 5. 结果裁剪：工具返回 10 万行数据，你得只回传摘要 return ToolResult(content=truncate_for_llm(result)) except TemporaryError as e: raise RetryableError(e)

对模型来说，工具只是一个function name + arguments。但对工程系统来说，一个工具就是一个接口，必须有版本、有超时、有鉴权、有限流、有监控。我曾经遇到过线上事故：一个搜索工具在外部 API 抖动时无限重试，把整个 Agent 循环卡死，用户等了两分钟没反应。加了超时和熔断之后，单次抖动最多让那轮回答慢 5 秒，但不会拖垮全部会话。

3.3 模型输出到工具调用的结构化落地

生产环境还有一个很容易踩的坑：模型输出的tool_calls本身不稳定。你会遇到：

• 返回了空的function.name
• 返回了 JSON 字符串但格式非法
• 返回了一个不在注册表里的工具名（幻觉）
• 连续多次返回完全相同的tool_call（模型卡循环）

对应的工程策略分别是：

1. Schema 强校验：所有工具入参必须有 JSON Schema，校验失败时，把错误信息回传给模型，让它修正，而不是直接终止。
2. 注册表白名单：registry之外的工具名一律拒绝，并返回“该工具不可用”，引导模型选择其他工具或直接回答。
3. 循环检测：记录最近几次模型返回的 tool_call 指纹，如果连续三次相同，中断循环并转入人工兜底或给出默认响应。

这些逻辑不会让你的 Agent 变得更聪明，但会让它不愚蠢地消耗你的钱和时间。生产系统的核心目标不是让 AI 表现出色，而是让 AI 失控时伤害最小。

4. 可观测性、超时与重试：决定生产级与 Demo 的分水岭

4.1 可观测性：AI 引擎比传统服务更需要 Trace

传统后端出问题，你可以看错误日志、看调用链、看慢查询。AI 引擎多了一个不确定因素：模型不是确定性代码——同样的输入，它可能正常返回，也可能因为语境漂移而返回完全不同的内容。这意味着你不仅要监控“服务是否报错”，还要监控“服务是否在做无效循环”“回答质量是否有波动”“用户是否正在遭受隐式失败”。

我的做法是把每一次模型调用、每一轮工具执行都记录为 trace span，贯穿全链路：

# 基于 OpenTelemetry 的语义约定，建议为 LLM 调用单独定义 span span = tracer.start_span("llm.completion", attributes={ "gen_ai.system": "openai", "gen_ai.request.model": "gpt-4o", "gen_ai.request.temperature": 0.2, "gen_ai.usage.input_tokens": usage.prompt_tokens, "gen_ai.usage.output_tokens": usage.completion_tokens, })

这些 trace 数据有几个用途：一是出事故时回放“模型当时收到了什么上下文”，二是统计不同 prompt 模板的 token 消耗，三是分析耗时瓶颈——是模型推理慢，还是工具调用慢，还是上下文太长导致 prefill 慢。没有这些数据，你只能靠用户吐槽来发现性能问题。

4.2 超时、重试与退避策略：别让一次抖动拖垮整个引擎

AI 引擎依赖的组件很多：模型 API、工具 API、向量数据库、缓存服务。任何一个上游抖一下，你的引擎就得跟着晃。传统后端的超时策略可以直接迁移过来，但要针对 LLM 的特点做调整。

• 模型调用超时：一般分两层。连接超时（比如 10 秒）和读取超时（比如 60 秒，具体看模型和上下文长度）。注意：流式响应下超时策略不一样，你需要设置“首个 token 到达超时”和“相邻 token 间隔超时”两个指标。
• 工具调用超时：按工具分别配置。内部数据库查询可以给 5 秒，外部第三方 API 如果业务允许，放宽到 10-15 秒，但绝对不能无上限。
• 重试策略：区分错误类型。429限流和5xx服务端错误可以重试；400参数错误不能重试（重试只会浪费钱）；网络超时要小心——请求可能已经在服务端被处理了，盲目重试可能造成重复扣费或幂等破坏。

退避算法我用的是带抖动的指数退避：delay = min(cap, base * 2^attempt) + random(0, jitter)。抖动很重要，否则大批并发请求同时重试，会把上游 API 打得更惨。具体参数我建议基准 1 秒起、上限 30 秒、最多重试 3 次。实测中这个配置在大多数场景下能平衡恢复速度和对上游的压力。

4.3 语义缓存：把重复请求挡在模型调用之前

这是成本优化里性价比最高的一步。很多 LLM 应用有大量重复或高度相似的请求——比如不同用户问了同一个产品问题、同一用户在不同会话里反复问“退款政策是什么”。如果每次都调模型，又慢又贵。

语义缓存的思路是：把 query 转成 embedding，在缓存中检索相似度高于阈值的历史问答，直接返回缓存结果。实现细节有几个注意点：

1. 相似度阈值要保守：设高了缓存命中率低，设低了容易答非所问。我习惯先在测试集上观察相似度分布，再选一个精确率和召回率平衡的点，通常是 0.92-0.97（取决于 embedding 模型）。
2. 缓存键要包含上下文信息：用户 ID、会话 ID、知识库版本号都应该纳入考虑。否则你升级了知识库，用户还命中旧答案，就变成事故了。
3. 只缓存无害请求：涉及查询个人敏感信息的请求不建议缓存，即使相似度再高也不要命中，否则可能把用户 A 的订单信息返回给用户 B。

生产级 AI 引擎的工程化，就是在这些看起来琐碎的决策里积累起来的。每一步单独看都不难，难的是把它们组合在一起还保持系统的简洁性和可维护性。

5. 多 Agent 与并行执行：从“一个循环”进化到“一组循环”

5.1 为什么需要多 Agent，而不是一个超级 Agent

AGI 还没到那个水平，所以你很难让一个 Agent 同时做好所有事。在工程实践中，我更推荐“多 Agent + 分工”的模式——不是因为它听起来高级，而是因为它把复杂任务拆成了可独立测试、独立扩展、独立降级的单元。

举书里的例子：一个企业文档问答系统，如果只用一个 Agent，它需要同时理解文档内容、调用检索工具、判断答案准确性、处理用户追问。这会导致 prompt 非常长、状态非常复杂，任何一个环节效果不好都很难定位。

拆成多 Agent 之后：

• Router Agent：判断用户意图，决定交给哪个下游 Agent。
• Retriever Agent：负责向量检索，输出候选片段。
• Reviewer Agent：负责检验答案是否有文档依据，无依据则要求重写。
• Summarizer Agent：负责把结果压缩成给用户的最终回复。

每个 Agent 的 prompt 更短、职责更清晰、指标更好定义。调优时你不再是“调一个巨型 prompt”，而是“调某个环节的 prompt”，效率高很多。

5.2 并行执行与任务调度：多 Agent 不是开线程就完了

多 Agent 真正的工程难点在于调度和并发控制。你不可能每来一个用户请求就拉起一轮完整的 Agent 执行链——那样成本失控，还可能因为并发太高触发上游限流。

我用的模式是一个三层结构：

# 伪代码：多 Agent 调度的主干 class AgentOrchestrator: def __init__(self, queue, workers, task_store): self.queue = queue # Redis Stream / RabbitMQ self.workers = workers # 按 Agent 类型区分的 worker 池 self.task_store = task_store # 任务元数据存储 async def submit(self, tasks: list[Task]) -> str: for task in tasks: await self.queue.push(task) return task.batch_id # 用于轮询整体完成状态 async def worker_loop(self, agent_type: str): while True: task = await self.queue.pop(agent_type) # 执行 Agent，把结果写回 task_store，触发下一个任务 result = await self.run_agent(task) await self.continue_chain(task, result)

几个关键的工程决策：

1. 队列必须支持延时重投：Agent 执行失败时，不应该立即重试，而应该放入延迟队列，等退避时间到了再投递。否则一个上游故障会导致海量任务同时重试。
2. 每个任务要可重入：任务状态要落库（pending / running / success / failed），Worker 崩溃后恢复时能重新领取未完成任务，而不是重复执行已完成的。
3. 并行度要有上限：同一个用户的多个子任务之间可以并行，不同用户之间更要隔离，避免一个用户的批量请求占满所有 worker 导致其他用户响应超时。

5.3 数据隔离与上下文共享

多 Agent 之间必然存在信息交换——Router 判断出的用户意图要传给 Retriever，Retriever 检索到的片段要传给 Summarizer。问题是：这些共享数据放在哪？

我的原则是“通过任务参数传递，不通过共享内存”。每个任务在创建时带上自己的上下文摘要：用户 ID、原始 query、上一步 Agent 的结论。后续 Agent 只依赖任务自带的上下文，不直接访问全局状态。这样做的好处是：

• 每个任务可以独立重试，不受其他任务影响。
• 任务的输入输出可记录、可审计、可回放。
• 不会出现“一个 Agent 改了共享状态导致另一个 Agent 行为异常”的灵异问题。

代价是上下文数据会有一点冗余——但这就是工程化的本质：用一点空间换系统的清晰度和稳定性，非常划算。

6. 评估、部署与版本管理：上线之后才是工程化真正的开始

6.1 离线评估集：你不该只靠“感觉它变聪明了”

AI 引擎没有传统意义上的单元测试，因为模型输出不唯一。但你有替代方案：回归评估集（Golden Set）。准备一批典型的输入和期望输出模式，每次修改 prompt、模型版本或工具逻辑后，跑一遍这些用例，比较输出质量。

质量评估可以分几个维度：

具体落地时，不需要一开始就做得很重。我从实践中的建议是：先维护 50-100 条高质量测试用例，跑一遍只需要几分钟，但能拦住大部分回归问题。等你的引擎复杂度上来了，再把评估接入 CI，每次提交自动跑。

6.2 Prompt 和模型即代码：版本管理你的“隐形代码”

传统代码库管的是.py和.ts文件，AI 引擎的代码库里还包含 prompt 模板、工具描述、系统提示词、甚至 embedding 版本。这些同样是代码，但它们有一个特点：改动很小，影响巨大。

比如你把 system prompt 里的一句话从“请用中文回答”改成“请用简体中文回答”，表面上只是两个字的变化，但可能让用户感受到“语气变冷了”。如果没有版本管理，你根本不知道哪次改动导致了评价下降。

我的经验是：

• 所有 prompt 模板都存成独立文件，允许 review，通过 CI 的评估集才能合并。
• 每次发布记录prompt 版本 + 模型版本 + 知识库版本 + 功能开关四位一体的发布清单。
• 任何时候都可以根据线上 trace 里的元数据，复现出当时用户看到的完整行为。

6.3 灰度发布与回滚：AI 引擎也要有刹车机制

AI 引擎的灰度发布比传统后端更复杂。传统后端灰度只需要关注“功能是否可用”，AI 引擎还要关注“回答质量是否有变化”。而质量是个模糊词，很难自动判定。

我的方案是分三个维度灰度：

1. 流量灰度：先切 5% 的真实用户流量到新版本，观察错误率和延迟是否异常。
2. 质量抽检：灰度期人工（或 LLM-as-Judge）抽检新版本的回答质量，与旧版本对比。
3. 反馈收集：在灰度版本上增加“用户是否满意”的反馈按钮，收集主观信号。

一旦发现异常，立即回滚——回滚不只是切代码，还需要把 prompt 版本、模型版本一起还原。这也是为什么前面强调发布清单要四位一体，因为 AI 引擎的行为是这些因素联合决定的，只回滚代码无法恢复行为。

7. 最后一公里：一个完整 AI 引擎的目录结构参考

整章内容比较细，我在书里附了一个参考目录，这里分享出来。它不是唯一答案，但按这个结构组织生产级 AI 引擎，能让每个关注点都有明确的落地位置：

your-ai-engine/ ├── app/ │ ├── engine/ │ │ ├── orchestrator.py # 多 Agent 调度 │ │ ├── executor.py # 工具执行器 │ │ ├── memory.py # 记忆与上下文管理器 │ │ └── evaluator.py # 质量评估器 │ ├── agents/ │ │ ├── router.py │ │ ├── retriever.py │ │ └── reviewer.py │ ├── tools/ │ │ ├── registry.py # 工具注册表 │ │ └── schemas/ # 每个工具的 JSON Schema │ ├── prompts/ │ │ ├── router_v1.yaml │ │ └── reviewer_v1.yaml │ ├── events/ │ │ ├── producer.py │ │ └── consumer.py │ └── telemetry/ │ ├── tracing.py │ └── metrics.py ├── tests/ │ ├── golden_set.json │ └── regression_runner.py ├── deploy/ │ ├── docker-compose.yml │ └── k8s/ └── config/ ├── production.yaml └── staging.yaml

这个结构最核心的原则是“关注点分离”。prompt 不进代码、工具不进循环、调度不进 handler、评估不离测试。每一层都可以独立演进，每一层都可以独立回滚。做到这一步，你的 AI 引擎才算真正“工程化”了，而不是一堆 prompt 和 API 调用的集合。

最后再分享一个我在实际项目中反复验证的体会：不要在架构上追求一步到位，而是先让 50 行循环跑通真实业务，再逐层把“不可靠、不可观测、不可维护”的部分替换掉。工程化是一个持续逼近的过程，它的目标不是造一个完美系统，而是让系统坏了能快速发现、能快速修复、能持续演进。你加上的每一次超时、每一条 trace、每一个版本标签，都是在给这台“AI 大脑”装上仪表盘和安全气囊——这比让它跑得快重要得多。