去年我把一个跑着的对话机器人项目停了，不是因为效果差，而是因为一个尴尬的问题：没人能说清它到底为什么会说出某句话。当时我们用了现成的AI框架，功能堆得很快，但线上一旦出异常，排查链路就像在黑盒里摸。后来我做了个决定——把 AI 工程这条链路上该有的东西拆开，从模型调用、流式输出、上下文管理，到 RAG、Function Calling、可观测性，全部用最少的依赖手写一遍。这就是 ai-engineering-from-scratch 这个项目的由来。
这篇文章不是教你用某个框架，而是分享一套适合“零依赖起步”的 AI 工程实现路径。读完你会知道，一个正经的 AI 应用从脚本变成系统，到底要越过哪些坎；也会明白那些封装好的框架，在帮你省时间的同时，又藏掉了哪些你必须补课的细节。适合两种人：第一种是已经写过简单的 ChatGPT 接口调用脚本、但没完整做过生产级 AI 系统的人；第二种是用框架半年以上、遇到深层问题只能靠猜的人。
1. 这个“从零开始”项目，到底在解决什么现实问题
1.1 脚本能跑和系统能上线，隔着好几道坎
很多人第一次接触大模型，代码长这样：读一个输入，拼一条 prompt，调用接口拿到字符串，print 出来。这在本地跑通很容易，但到生产环境就是另一回事。
生产环境的对话系统要考虑的事情，脚本阶段根本看不见：用户刷新页面时上一次的上下文去哪了？模型返回到一半超时怎么办？用户连续问十轮，聊天记录把 token 撑爆之后成本谁来兜底？系统要调用内部工具时，解析错了参数是重试还是直接报错？这些都不是“提示词写得好不好”能解决的，而是工程问题。
更麻烦的是，如果一开始就站在框架的肩膀上，这些问题会被包装成几个配置项。表面上省了事，实际上你失去了对数据流向的感知。出问题时框架的报错信息往往指向自己的抽象层，而不是根因。我自己遇到过一次诡异的工具调用循环，排查了一下午，最后发现是框架在某个版本里自动处理消息历史时把 tool 结果截断了——这类问题，不读源码根本发现不了。
1.2 项目划定边界：只用最少的依赖，手写六个关键模块
ai-engineering-from-scratch 的核心思路很简单：不追求复刻一个 LangChain，而是把 AI 应用必备的组件逐个手写实现。我在项目里圈定了六个模块作为主线：
- 模型接入层：不依赖官方 SDK，直接构造 HTTP 请求，支持流式响应；
- 上下文管理：token 预算、滑动窗口、旧消息压缩；
- 记忆持久化：会话存储与并发控制；
- RAG 检索链路：分块、混合检索、重排、引用溯源；
- Function Calling：工具注册、参数校验、执行与结果回填；
- 可观测性与评估：trace 贯穿、离线回归集、成本记录。
选这六块，是因为它们基本覆盖了一个“能回答复杂问题、能调用外部工具、能追责审计”的对话系统的主要骨架。框架能给你的，也就是这些模块的组合，区别在于手写时你会清楚地知道每一个步骤的输入输出和失败模式。
依赖方面我只保留了 HTTP 请求库、一个基础的 tokenizer 计数库、一个向量索引库，以及 pydantic 这类数据校验工具。剩下的逻辑全部自己写。
1.3 什么人适合照着这条路走一遍
如果你刚接触大模型应用开发，我建议你先把简单的接口调用跑通，再来看这个项目。如果你已经用框架做了两三个月、被一些隐形行为坑过，那么照着这条路重写一遍会有种“原来如此”的爽快感。
这个项目不适合以下场景：公司项目工期紧张、需要两天上线 Demo 的时候，请直接用框架；你的目标只是做一个网页版聊天玩具，也不需要从零造轮子。它适合的是学习和沉淀——当你需要给框架打补丁、做二次开发或者优化成本时，手写一遍带来的底层认知会直接转化为生产力。
2. 手写模型接入层：流式、重试与多 Provider 抽象是必修课
2.1 一个不依赖 SDK 的 Chat 接口封装长什么样
很多官方 SDK 只是对 HTTP 接口的封装，自己写一个并不难，但写的过程中你能更好地理解每次请求到底发送了什么。一个最小可用的封装，核心就四步：构造 URL、组装请求体、处理响应、统一错误。
下面是我在项目里用的一个极简版本，兼容 OpenAI 格式的接口，通过配置切换服务地址：
import json import time import httpx from typing import Optional class LLMClient: def __init__(self, base_url: str, api_key: str, model: str): self.base_url = base_url.rstrip("/") self.api_key = api_key self.model = model self.client = httpx.Client(timeout=60) def chat( self, messages: list[dict], temperature: float = 0.7, max_tokens: int = 2048, tools: Optional[list[dict]] = None, stream: bool = False, ) -> dict: payload = { "model": self.model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": stream, } if tools: payload["tools"] = tools resp = self.client.post( f"{self.base_url}/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json=payload, ) resp.raise_for_status() return resp.json()注意这里的 messages 是完整的消息数组，我没有在这个函数里做任何“自动保存历史”的操作。原因很简单：上下文管理策略应该由上层业务决定，接入层只负责把消息可靠地送到模型那里。很多框架把历史记录也封装进了 client，短期看方便，长期看反而让调用链混乱。
2.2 流式输出不是炫技，而是生产环境的及格线
如果你做过用户调研就会发现，非流式接口在响应稍慢时，用户会认为系统“卡死了”。流式输出的意义不只是体验问题，它还能尽早地把首字呈现给用户，掩盖长耗时。
流式接口的返回是 SSE（Server-Sent Events）格式，本质是一个持续的 HTTP 响应流，每一行是data: {...}格式的 JSON。手写流式解析的关键是正确切割事件块，并且区分content增量块与终止标记：
async def chat_stream( self, messages: list[dict], options: Optional[dict] = None, ): payload = { "model": self.model, "messages": messages, "stream": True, **(options or {}), } async with httpx.AsyncClient(timeout=None) as client: async with client.stream( "POST", f"{self.base_url}/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json=payload, ) as resp: async for line in resp.aiter_lines(): if not line.startswith("data:"): continue data_str = line[len("data:"):].strip() if data_str == "[DONE]": break chunk = json.loads(data_str) delta = chunk["choices"][0].get("delta", {}) if "content" in delta: yield delta["content"]这里我特别想提醒一个坑：不要在流式解析里用json.loads直接处理整行后，试图在同一个循环里写完整个业务逻辑。正确的做法是把解析出来的增量文本交给上层回调，让 UI 层或消息总线去处理转发。流式链路里最容易出问题的不是解析本身，而是下游消费者没做好背压控制——也就是下游处理速度跟不上上游时，内存占用会持续上涨。
2.3 限流与超时：重试策略决定系统在高峰期的生死
没有重试策略的 AI 应用，遇到一次限流就会把错误抛给用户。但重试不是简单地“失败了再来一次”，盲目重试会加剧服务端压力，甚至让整个系统进入雪崩状态。
我的做法分三层。第一层，区分错误类型：网络超时、429 限流、5xx 服务端错误分别处理。第二层，用指数退避加抖动（jitter）控制重试间隔。第三层，设置总重试次数上限，避免无谓地占用请求线程。
下表是我在生产环境实际使用的重试参数：
| 错误类型 | 是否重试 | 初始等待 | 最大等待 | 重试次数 |
|---|---|---|---|---|
| 网络连接超时 | 是 | 1s | 10s | 2 |
| 429 限流 | 是 | 2s | 30s | 3 |
| 5xx 服务端错误 | 是 | 1s | 15s | 2 |
| 4xx 参数错误 | 否 | - | - | 0 |
抖动算法我用的是random.uniform(0, current_delay)，而不是固定的当前退避值。这样能让多个并发请求的重试时间错开，避免出现“重试风暴”。
接入层写到这里，我踩过最大的坑是连接复用。如果每次请求都新建一个 TCP 连接，在高并发下握手开销会非常可观。使用httpx.Client（而非每次创建新的）可以复用连接池，这一点在长对话场景里效果尤其明显。
3. 上下文管理是工程问题，不是“多塞几轮对话”的问题
3.1 从不设上限的超长对话到 token 预算失控
很多新手的第一个错误是：把用户所有的聊天记录原封不动拼进 messages。这个做法在一两轮对话时没问题，但对话一旦超过十轮，情况就会迅速恶化。
先说成本：主流模型按 token 计费，历史消息每多一轮，每次请求都得多付一份全量历史的费用。一个高频场景如果每天有十万次请求，这意味着大量的成本消耗在重复传输相同的历史内容上。
再说效果：模型的注意力是有限的，历史聊得越久、内容越杂，模型就越难聚焦在当前问题上。我实测过，当历史消息超过一定阈值后，模型的回答连贯性会明显下降，有时候它甚至会把早期对话里的错误信息当成事实引用出来。这种“越聊越笨”的现象，根源往往不是模型本身，而是上下文管理没做好。
3.2 滑动窗口、摘要折叠与系统提示里的“善意谎言”
解决 token 爆炸，我采用了三层策略组合。
第一层是滑动窗口。保留最近 N 轮完整消息，更早的直接丢弃。这个 N 的值由 token 预算决定：先设定本次请求允许占用的上下文上限（比如 8000 token），再预留输出空间（比如 1000 token），剩下的才分配给历史消息。
第二层是摘要折叠。直接丢弃太早的对话会丢失重要信息，所以我会在滑动窗口之外维护一个“早期摘要”，用模型把更早的对话压缩成一两句要点。这相当于给模型一张“记忆压缩卡”。我实际用下来，摘要的更新频率不需要每次对话都做，可以每 5 轮更新一次，省掉很多额外调用。
第三层是系统提示里的说明。模型并不知道“早期对话已被折叠”，如果对话突然出现话题断层，它可能感到困惑。所以我会在系统提示里写一句“更早的对话内容已经过摘要压缩，其中可能缺少部分细节”，让模型在不完整的上下文里依然保持稳定状态。
def build_messages(history, summary, current_question): messages = [ {"role": "system", "content": f"你是一个对话助手。以下是更早对话的摘要：{summary}"}, ] # 保留最近的消息，最多按 token 预算截取 recent_messages = trim_to_token_budget(history, max_history_tokens=8000) messages.extend(recent_messages) messages.append({"role": "user", "content": current_question}) return messages3.3 会话持久化与并发写入的细节缺陷
上下文管理不只是内存里的数组操作。当用户刷新页面继续对话，系统必须能从持久化存储中恢复会话，这就要引入会话 ID 和消息表的读写。
消息表的设计至少要包含：会话 ID、消息角色、消息内容、时间戳、token 数。表结构不复杂，但会踩的坑是并发写入。用户可能在多个设备上同时发起对话，两个请求同时读取同一个会话的历史，然后各自写入后续消息，就会发生“覆盖丢失”。
我用的方案比较务实：给会话加版本号，写入时带上where version = 当前读到的版本，如果更新影响行数为 0，说明版本冲突，就重读最新历史并重新生成回答。用乐观锁而不是分布式锁，因为分布式锁在这种场景里往往过于重量级。这个处理逻辑看起来简单，但它避免了我线上遇到过的“用户看到两个互相矛盾回答”的尴尬。
4. 手写 RAG 的真正难点：检索质量才是核心矛盾
4.1 分块策略的取舍：小窗口漏语义，大窗口混噪音
RAG 的完整流程是：先对文档分块，再做向量化入库，检索时把用户问题向量化，在库里做相似度搜索，取 Top-K 块喂给模型。看起来很简单，但实际效果天差地别，问题往往出在分块这个不起眼的环节。
分块太小，比如按固定 100 字切，每一块承载的语义信息太少，检索时经常召回一些“看起来相似但不完整”的内容，大模型拿这些碎片拼不出正确答案。分块太大，比如整个章节塞进一个块，向量表示会被平均化，检索时容易被不相关的内容干扰。
我试过几种分块方式后，现阶段比较稳定的是结构优先 + 重叠窗口的组合。先用文档的结构标题切分，把内容切成一个个语义完整的段落；如果段落仍然过长，再按固定长度二次切分，并且保留前后各 50 字的重叠。这个重叠窗口能在一定程度上避免切断句子造成的语义漂移。
4.2 混合检索：向量召回之后为什么还要过一遍 BM25
纯向量检索的主要问题是：它只理解“语义相似”，对精确匹配反而容易失灵。比如一段代码里的函数名、一个产品型号、一个专利编号，向量检索很可能会因为“含义相近”而找到了别的文本，却漏了你真正要找的那个精确 ID。
我在项目里引入了混合检索——向量召回与 BM25 关键词检索并行跑，得到两个候选集合并集，再做分数融合。BM25 是经典的关键词匹配算法，不需要训练，对精确术语的召回表现很好。分数融合我用的是简单的倒数排名融合（RRF），实现成本低且调参轻松：
def rrf_merge(vector_hits: list[str], keyword_hits: list[str], k: int = 60) -> list[str]: scores: dict[str, float] = {} for rank, doc_id in enumerate(vector_hits): scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1) for rank, doc_id in enumerate(keyword_hits): scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1) return sorted(scores, key=scores.get, reverse=True)实践中的一个技巧是，对向量检索和关键词检索设置不同的 Top-K，比如向量招 100 个、关键词招 50 个，合并后再统一截取。宁可多召回也不要漏召回，因为后面还有重排环节兜底。
4.3 重排与溯源：让大模型“回答有据可查”
纯向量相似度排名和真正的“答案质量排名”之间还有一道鸿沟。拿一个更精准的重排序模型对粗召回的候选块做二次排序，是有明显收益的。重排模型的输入是被召回的文档块和用户问题，输出是一个相关性分数，它比双塔式向量检索更敏感，因为它能对文本对做深度融合。
我在重排后取前 5 块作为上下文喂给模型。同时每一块都携带了source_id和chunk_index元数据。这些元数据有两个用途：一是限定检索范围，比如用户指定只看某个章节时，直接在召回阶段过滤掉其他来源；二是让模型在回答中附上引用标记。
引用这个要求，需要在 prompt 里明确指示：“当你的回答引用了提供的资料时，请在句末标注[来源编号]”。不用依赖模型自觉，你只要在拼接上下文时把每个块的编号暴露给它。输出解析阶段，把[来源编号]提取出来映射回真实的文档链接，用户点开就能对照原文。这一步在落地企业内部知识库时几乎是刚需，否则回答错了也没法追责，大模型幻觉会被无限放大。
5. Function Calling 手写接入：把工具调用当成协议来处理
5.1 一次完整的工具调用链路长什么样
Function Calling 的本质是一个协议：模型在生成文本之外，还可以输出一个结构化的“工具调用请求”，你的系统负责执行这个请求，再把结果作为一条新消息回传给模型。
手写实现的核心是维护一个工具注册表。每个工具有三个要素：函数本体、JSON Schema 格式的参数声明、以及一段描述文字。发起请求时，把工具 schema 放到请求的tools字段里，模型在需要时会返回tool_calls数组。
TOOL_REGISTRY = { "calculator": { "schema": { "type": "object", "properties": { "expression": {"type": "string", "description": "数学表达式，如 12*3+4"} }, "required": ["expression"], }, "description": "计算数学表达式的结果。适合做单位换算、算术运算。", "func": calculate_expression, } } # 请求时组装 tools 参数 tools_schema = [ { "type": "function", "function": { "name": name, "description": meta["description"], "parameters": meta["schema"], } } for name, meta in TOOL_REGISTRY.items() ]模型返回tool_calls后，你要执行对应的函数，并把结果以role: "tool"的新消息追加进消息数组，再用新的消息数组请求一次模型。这次模型会把工具结果消化进回答里。
一条核心原则：工具调用只是“多一轮普通对话”。很多框架把这层包装得很神秘，但拆开看，就是一次普通的请求-响应循环，难点全在外围工程。
5.2 模型会“乱调用”：描述、校验、迭代上限一个不能少
工具调用在真实场景里最大的问题不是“模型不会调”，而是“模型乱调”。模型可能会推测空缺参数、把类型搞错、或者在明明不需要工具的场景里强行调用。
针对推测参数，需要做两件事。一是描述写清楚，避免歧义。比如一个查天气的工具，描述写成“根据城市名称获取当前天气”，参数要求明确“城市必须是标准中文城市名，不要包含‘市’字”等等。二是参数落地前严格校验，只要不符合 schema 就重新调用模型修正，而不是直接执行。pydantic 在这里非常顺手，把 JSON 数据直接校验成类型安全的对象。
死循环也必须兜底。模型在连续多次工具调用后可能仍然无法获得满意结果，然后继续调用、继续失败。我实现了一个迭代上限控制，默认最多允许 5 轮工具调用循环，超过次数直接把当前状态返回给用户并说明“处理链路过长”。
5.3 工具即权限边界：安全防护不能交给模型自觉
把工具视作普通函数，容易忽略一个关键问题：工具执行是有实际影响的。查询只读数据没问题，删除操作、发送通知、修改订单状态，这些被执行后会产生真实副作用。
我的原则是两条。第一，高风险工具一律走“人工确认”模式：工具调用结果只生成一个待确认指令，返回给前端等用户点击确认后才真正执行。第二，在工具执行层加上白名单和超时控制，无论模型说什么，函数级权限不能越过预设边界。
有一次我让模型尝试调用一个内部“重发通知”的工具，它因为用户措辞“能不能再发一次”就自动执行了。自那以后所有带副作用的工具都改了确认流程。模型的职责是生成意图，执行权和确认权必须掌握在系统流程手里。
6. 从能跑到能上线：可观测性、评估与稳定性补课
6.1 Trace 贯穿全链路：出了问题先问“哪个环节最慢”
手写组件的另一个好处是，你有机会在每一个环节埋点。我给每一次用户请求生成一个trace_id，从入口日志开始贯穿到模型调用、检索、重排、工具执行。每条日志记录三件事：发生时间、耗时、当时的关键数据大小。
这个习惯帮了我大忙。有一次用户反馈“回答越来越慢”，我按 trace 一查，发现耗时没增加在模型调用上，而是某个文档集合的 vector 检索时长在持续上升。原因是库里的向量数据增长后，暴力计算相似度的开销非线性上涨，而框架默认观察不到这一层。后来我给向量索引加了一个分片策略，慢查询就消失了。
成本记录也放在这个环节。每条 trace 里带上本次请求的 prompt token 数和 completion token 数，汇总到一张成本表，每天跑一个定期报表。如果没有这层数据，你永远不会知道某些高频场景到底在烧多少预算。
6.2 评估不靠感觉：离线测试集与回归保护
改一个 Prompt 或改一段检索逻辑，怎么判断是变好了还是变坏了？靠肉眼抽几条样本看，根本不可靠。我在项目里维护了一个小型但真实的测试集，大约有一百多组问题-标准答案对，覆盖日常问答、引用溯源、数值计算、多轮对话等场景。
跑回归的方式分两种。第一种是确定性规则评估，适合有明确答案的场景，比如计算题，直接比对模型输出里能否提取出正确数值。第二种是 LLM-as-judge，用一个更强大的模型给回答打分。打分维度也是二元化的：相关性和正确性分开评，避免把“说得漂亮但不正确”的答案误判为高质量。
回归测试的频率不需要每次提交都跑全量，但每条关键 Prompt 修改后必须跑一遍。否则最常见的局面是：这次改完一个场景效果好了，一周后发现另一个场景悄悄变差了，而且完全不知道是哪次改动引入的。
6.3 模型降级、缓存与成本控制的实际操作
生产系统不能只有一个模型供应商。线上出故障时，我采用降级策略：主模型请求失败后，自动切换到备用模型，同时在回答里标注“系统当前处于降级模式，部分质量可能下降”。
另一层是缓存。对于相同或高度相似的问题，直接把之前的完整回答缓存起来。缓存键不只是问题文本，而是“规范化后的问题 + 相关工具结果”，可以显著提升高频重复场景的响应速度并降低成本。代价是要控制缓存过期时间。
还有一个细但效果明显的优化：路由分流。对于简单问题，完全可以先让一个小模型判断复杂度，再决定是直接用便宜的小模型回答，还是把请求转给大模型。虽然多了一次判断请求，但综合成本往往最多省到两三成，复杂场景稳定后收益更明显。
做完这些，一个“能回答问题”的项目，才变得像“能长期运营的系统”。可观测性、评估和稳定的降级策略，单看每一项都不是很起眼，但它们组合在一起，决定了你在线上出事时是有序处理还是一团乱麻。
最后说一点个人体会。写完这个项目之后，再回头看那些框架，我不再觉得它们是黑盒了。遇到框架行为不符合预期时，我大致能猜到它在哪一层做了什么事情，也能直接翻到对应的源码去确认。这种底气，不是靠读文档获得的，而是靠把每条链路亲手实现一遍换来的。如果你也在 AI 工程这条路上往前走，试试隔一段时间就放下框架，从零把一个模块写干净。你会发现那些“看上去很简答、做起来总出问题”的细节，才是真正让系统稳定运行的关键。
这个项目后续我会继续补两件事：一个是更完整的离线评估集，另一个是缓存命中率和成本报表的可视化展示。代码已经在仓库里持续维护，如果你也在手写你自己的 AI 工程链路，欢迎照着这条路走一遍——不一定需要一模一样，只要把每个环节的取舍真正想明白，就是值得的。
