从零手写AI工程链路：模型调用、RAG与Function Calling

去年我把一个跑着的对话机器人项目停了不是因为效果差而是因为一个尴尬的问题没人能说清它到底为什么会说出某句话。当时我们用了现成的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(timeout60) 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: fBearer {self.api_key}}, jsonpayload, ) resp.raise_for_status() return resp.json()注意这里的 messages 是完整的消息数组我没有在这个函数里做任何“自动保存历史”的操作。原因很简单上下文管理策略应该由上层业务决定接入层只负责把消息可靠地送到模型那里。很多框架把历史记录也封装进了 client短期看方便长期看反而让调用链混乱。2.2 流式输出不是炫技而是生产环境的及格线如果你做过用户调研就会发现非流式接口在响应稍慢时用户会认为系统“卡死了”。流式输出的意义不只是体验问题它还能尽早地把首字呈现给用户掩盖长耗时。流式接口的返回是 SSEServer-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(timeoutNone) as client: async with client.stream( POST, f{self.base_url}/chat/completions, headers{Authorization: fBearer {self.api_key}}, jsonpayload, ) 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控制重试间隔。第三层设置总重试次数上限避免无谓地占用请求线程。下表是我在生产环境实际使用的重试参数错误类型是否重试初始等待最大等待重试次数网络连接超时是1s10s2429 限流是2s30s35xx 服务端错误是1s15s24xx 参数错误否--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_tokens8000) 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, keyscores.get, reverseTrue)实践中的一个技巧是对向量检索和关键词检索设置不同的 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*34} }, 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 工程链路欢迎照着这条路走一遍——不一定需要一模一样只要把每个环节的取舍真正想明白就是值得的。