1. 为什么我劝你先别碰LangChain从零构建AI工程的原点这两年ai-engineering成了行业热词朋友圈里人人都在聊Agent、RAG、多模态编排GitHub上的star数像不要钱一样疯涨。但说实话真正动手自己跑通一条AI应用链路的人远比嘴上聊的人少。更少的是那种不靠一切魔法硬生生从底层把AI工程搭起来的人。我见过太多新人的学习路径是这样的先装LangChain再装个向量数据库然后复制一段对话机器人demo跑通之后就觉得自己懂AI工程了。但真到生产环境回调函数报错、上下文管理混乱、模型超时重试策略缺失、成本失控这些事一来整个人直接懵掉。因为框架把所有底层细节都藏起来了你根本不知道问题出在哪一层。所以我这篇想讲的不是如何用某某框架几天做出一个AI应用而是从零开始不依赖任何AI框架自己动手把AI工程的骨架搭出来的完整路径。这篇文章的受众很明确已经会写一点Python知道AI模型是什么听说过LLM、Agent、RAG这些词但想要真正理解AI工程底层逻辑而不是只会套框架的人。核心观点我先抛出来AI工程早期最重要的能力不是会用某个高级框架而是能在没有框架的情况下自己定义架构、自己处理数据流、自己控制成本和延迟。框架迟早会换但工程素养不会。我自己就是这条路走过来的。第一次做AI应用时被LangChain的链式调用折磨得不行——每加一个环节回调栈就深一层出了问题根本不知道是模型返回异常还是Prompt写坏了。后来我一气之下把所有框架代码删干净用原生HTTP请求重新实现了一遍才真正摸清了AI工程的核心链路从提示词组织、上下文管理到模型调用、结果校验、成本控制、持续评估每一步都清清楚楚。这篇文章就当是我把当时踩过的坑、总结出的经验原原本本分享出来。2. 最小可运行骨架一次原生HTTP调用背后藏了多少工程问题2.1 不用SDK用requests直接怼API很多人第一次接触大模型API都是直接在官方文档里复制SDK示例代码。这么做没错但如果你真想搞懂AI工程我强烈建议你先用requests库直接发一次HTTP请求。别管官方封装多优雅你先亲眼看看那个请求体长什么样。一个标准的对话补全API请求本质上就是POST一个JSON到指定接口核心参数就这几个import requests import json # 伪代码示例展示与大模型对话的最小请求体 def call_llm(prompt, api_key, base_urlhttps://api.example.com/v1): payload { model: your-model-name, messages: [ {role: system, content: 你是一位资深技术助手回答需准确、简洁。}, {role: user, content: prompt} ], temperature: 0.7, max_tokens: 1024 } headers { Authorization: fBearer {api_key}, Content-Type: application/json } resp requests.post( f{base_url}/chat/completions, headersheaders, jsonpayload, timeout30 ) if resp.status_code 200: data resp.json() return data[choices][0][message][content] else: raise Exception(fAPI call failed: {resp.status_code} {resp.text})这十几行代码跑通你才算真正触到了大模型API的骨骼。你会直观看到三个关键事实对话是一个数组。messages是消息数组system、user、assistant三条角色反复交替构成上下文。模型没有记忆它的记忆就是你每次请求携带的message数组。理解这一点你就知道了为什么说上下文管理是AI工程的命门。一切参数都是工程参数。temperature、max_tokens这些都直接和成本、质量挂钩。max_tokens设太小回答会被腰斩temperature随意调代码场景下输出会飘。失败模式多到超出想象。网络超时、限流429、内容截断、格式异常这些不是万一会发生而是一定会发生。没有处理这些异常的代码AI应用就是纸糊的。2.2 从一根直通管道到一张防御网必经的六个工程改造把上面的最小代码扩展到真实可用的工程需要给这个光秃秃的HTTP请求套上六层防御网。我逐个说**第一层密钥管理。**没有哪个正经项目会把api_key硬编码在源码里。环境变量读入是我建议的起点。至少做到代码里没有明文密钥这个底线。**第二层超时与重试。**我建议设置连接超时10秒、读超时60秒这种双超时策略。连接超时只管TCP握手建连不覆盖模型生成时间模型生成慢是正常现象直接粗暴给一个总超时会导致大量误杀。重试则要区分场景429限流响应可以重试4xx纯粹是请求错误就不要重试5xx可以重试一到两次。**第三层用户隔离。**多人共用一套API时必须能在代码里区分请求来源。常见做法是拿user_id可以放在请求体里便于平台侧做用量统计和审计追责时不至于抓瞎。**第四层成本配额。**这是几乎所有人第一次做AI工程都会忽略的事。不考虑成本的请求代码上线一晚上账单能让你怀疑人生。在每个用户维度、每天维度上做好token用量统计和上限控制看代码时一定要养成每次请求先算钱的习惯。**第五层结构化日志。**AI应用的调试难点在于不确定性问题。同一个Prompt两次返回可能天差地别。所以必须做日志记录请求参数、返回内容、耗时、token用量、错误信息为追溯和评估提供素材。**第六层结果校验。**模型输出是概率性的你不能假设它一定返回合法JSON、一定遵守约束。工程上要做输出Schema校验不符合就让模型自己纠错或降级使用。这六层防护每一层都值得做成独立的代码模块。等你全做完你的最小可运行骨架已经不是一个demo而是一个初具规模的AI应用了。这个过程就是LangChain这类框架替你做的事——它帮你把它们抽象封装了。但区别在于你自己做一遍出了问题你能定位、能修、能改用了框架你只能瞎试。2.3 单轮对话跑通之后下一步立刻做多轮状态管理单轮对话调通后新手第一个工程化需求就是多轮对话。我给你一条最朴素的路线不用数据库不用向量先在内存里用字典维护每个session_id对应的消息列表。这不是最优方案甚至生产环境很拉胯但它是帮你理解上下文为何重要的最短路径。# 一个最简的多轮会话管理内存版仅用于理解原理 session_memory {} def chat_with_session(session_id, user_message, api_key): # 取历史消息没有则初始化系统提示词 if session_id not in session_memory: session_memory[session_id] [ {role: system, content: 你是帮助用户解答技术问题的小助手。} ] history session_memory[session_id] history.append({role: user, content: user_message}) # 调用大模型 reply call_llm(history, api_key) history.append({role: assistant, content: reply}) # 截断限制长度防止无限膨胀 if len(history) 10: # 保留系统提示词和最近9条消息 session_memory[session_id] [history[0]] history[-9:] return reply你把这个跑起来聊个二三十轮就能感受到上下文管理最本真的痛点历史消息越积越多请求体越来越臃肿成本快速飙升而且模型对很久以前的信息记忆会漂移。这就是所有复杂RAG架构、上下文压缩策略、向量记忆机制要解决的那个最原始的问题原型。从零构建的意义就在这里你能清晰地看见问题的起源自然而然就知道解法为什么长那样。3. 数据工程从零做起检索增强不是魔法是清洗、索引、召回3.1 一个不装外部服务的本地向量检索替代方案聊到RAG检索增强生成大家都喜欢一上来就搞向量数据库。但我必须说如果没有索引和召回这一步的底层概念直接上向量库就是一个黑盒加另一个黑盒。我的建议是先做一个关键词倒排索引的本地检索器。为什么因为它逻辑透明再小的demo也能把数据进、索引建、检索出这一套完整跑通。等你知道检索的本质了再换向量也不是难事。# 简易倒排索引实现剖析检索的底层逻辑 import re from collections import defaultdict class SimpleRetriever: def __init__(self): self.documents {} # doc_id - 文本 self.inverted_index defaultdict(set) # 词 - 包含该词的doc_id集合 def add_document(self, doc_id, text): self.documents[doc_id] text tokens set(re.findall(r\w, text.lower())) for word in tokens: self.inverted_index[word].add(doc_id) def search(self, query, top_k3): tokens re.findall(r\w, query.lower()) scores defaultdict(float) for word in tokens: for doc_id in self.inverted_index.get(word, []): scores[doc_id] 1 # 简单词频打分这里可以替换为TF-IDF ranked sorted(scores.items(), keylambda x: x[1], reverseTrue) return [(doc_id, self.documents[doc_id]) for doc_id, score in ranked[:top_k]]这段代码肯定被懂行的人笑话太原始但这恰恰是它的价值——它把检索的真相摊在你面前文档切词词反查到文档再按命中次数打分排序。向量检索做的事情本质上也是在解决这个查找相关文档的问题只是换了一种更鲁棒的方式。当你看到TF-IDF可以替换Top-K怎么取都对结果有影响这样的关键决策点你就会明白为什么RAG的工程调优空间那么大。3.2 数据清洗的脏活累活喂给检索器的内容决定一切很多踩坑案例里RAG效果不好第一反应是换个更好的向量模型。但就我的经验而言80%的情况下问题压根不在检索算法而在数据进索引之前那一步——你没做清洗也没做切分数据直接是坨脏东西。做数据工程我总结四道工序格式归一。PDF、Word、网页抓取的文本各种乱码、多余空行、特殊字符先清理掉。这一项不做好切出来的chunk全是垃圾进垃圾出。内容去重。语义完全相同或几乎相同的文档片段要合并。不然检索返回三个结果内容却是一模一样的复制粘贴浪费时间也浪费token。粒度切分。这是RAG的核心艺术。切太短语义碎片化检索回来的是不完整的逻辑单元切太长又容易塞进无关噪声成本和延迟双高。我的经验是优先按自然语义段落标题层级切没有标题的按固定窗口如500字叠加重叠overlap取100字。这个比例需要在你的语料上实测没有放之四海皆准的参数。元数据标注。给每个切分片段打上来源、章节、时间等标签在检索后置过滤环节有大用。比如用户问今年的政策可以按时间元数据过滤掉去年的内容。这比在查询词上做文章简单得多。数据工程的每一步都是可以用代码显式控制的。做得多了你会有种每一条数据都经我手看过一遍的掌控感。这感觉在工程里的价值比什么都重要。3.3 从检索到片段到组织成可用上下文有一个容易被忽视的环节我早年做RAG最常犯的错是检索器返回了五个片段我就全部拼接硬塞给模型结果模型答非所问。后来我明白检索结果要经过一道组织工序重排。召回靠粗排比如向量相似度拿到以后可以做一次精排。简单做法是用模型打分、或按规则把与查询有强关键词重叠的排前面。这一步对最终答案质量影响经常是决定性的。筛选。设定相似度阈值不相关的宁可不要。宁可少给不要乱给。编排顺序。把多个片段按相关性高优先、时间新优先等规则排序。你给模型的上下文顺序会影响模型注意力分布不是小事。我在RAG项目里从零做的最大收获其实不是写代码的能力而是对上下文是一种可编程资源的体感。你能拼、能裁剪、能编排模型的输出上限就被你的上下文策略锁死了。这个观念建立起来后任何Retrieval框架在你眼里都只是一堆可替换的零件。4. 系统设计给了我另外半条命延迟、成本、评估缺一不可4.1 延迟预算悖论用户能等5秒但不能等5秒加1毫秒AI应用和传统应用一个很大的区别传统接口延迟你压到100毫秒以下就胜利了AI应用一次模型生成动不动几秒你不可能靠压性能做到即时响应。所以AI工程里的延迟优化思路要换。我自己的做法是延迟预算制先定一条硬基线比如用户从发问到看到开头的总时间不能超过3秒。然后拆解各部分预算环节典型耗时可能依服务波动优化思路网络请求50~200msCDN、就近接入LLM首token返回300ms~3s取决于模型和Prompt长度流式输出是必须的让用户先看到字在动短Prompt能显著压缩首token时间检索/RAG50~500ms索引优化采用缓存后续生成无法压缩与首token逻辑解耦先吐骨架再补细节这里最大认知点在于模型的生成时间你是无法压缩的你能压缩的是外围的检索和延迟预算分配。让用户在首token后感知到有反馈比压缩总延迟更重要。同样一个3秒的API请求一次性转半天才返回和流式快速吐前几个字用户体感天差地别。4.2 成本控制的算账思维从单次调用到规模化预算成本控制是AI工程绕不开的生死线。要控制成本第一步永远是会算账。我给你一个简单的成本评估模型假设你一次会话平均要打2次模型调用每次调用平均输入2000 token、输出300 token。那么一次会话消耗约为4300 token。如果按每百万token $5的价格估算这个数字随市场波动但量级差不多一次会话成本约0.02美元。听上去很少如果你的DAU是1000人每人每天平均发生10次会话那就是0.02美元乘以一万一天就要200美元一个月6000美元一年七万多美元。这几行字算完你立刻就会明白为什么生产级的AI应用都拼命做缓存和减少重复调用为什么上下文裁剪是一项省大钱的工作为什么Prompt压缩、模型蒸馏、本地小模型部署这些技术有巨大的商业价值。成本意识是刻进工程血液里的。每写一次LLM调用我都条件反射地问自己三句话这个调用能不能省能不能用更小的模型能不能缓存4.3 评估三件套离线指标、规则断言和回归集AI工程和传统工程最大的不同在于代码质量测试只能覆盖一部分。传统的单测只能测函数输入输出但模型的输出没法简单地做断言。所以需要一套评估三件套**第一件离线指标。**在固定数据集上算指标比如RAG领域的召回率、命中率。不需要理解语义用机械指标快速判断检索链路有没有退化。上线一个新清洗逻辑后先跑一遍召回指标比人工看十条case高效太多。**第二件规则断言。**针对输出做结构化和规则校验比如输出是不是JSON关键字段是否存在文本长度有没有极端值。我习惯用一个简单的pytest风格函数来做断言自动化程度高重现成本低。def assert_reply_quality(reply, required_fields, max_length2000): # 基础规则断言校验输出基本质量 assert len(reply.strip()) 0, 回复为空 assert len(reply) max_length, f回复超长: {len(reply)} assert not any(bad_word in reply for bad_word in [抱歉我不理解, Network Error]), 回复包含异常占位词**第三件回归集。**它是最重要的兜底防线。每个迭代版本都选一批覆盖典型场景的用例比如100条跑一遍对比新旧版本输出。一旦某条质量明显退化立刻回滚。这个思想抄的软件工程回归测试但它做在模型行为这一层上。没有回归集你的AI应用改代码就像蒙眼开车。我强烈建议在项目第一周就把评估三件套搭建好哪怕粗糙。因为人脑的好像还行式评估在数据量上来后必然失真。5. 完整可复现的例子一个带缓存、降级、评估的最小Agent5.1 模块划分与主流程不到200行做完整闭环把前面三章的东西串成一个最小但完整可运行的项目我用一个智能客服助手作为场景把项目拆成五个模块逻辑清晰到可以直接抄走ai-engineering-mini/ ├── client.py # 大模型API客户端超时、重试、限流 ├── cache.py # 缓存层命中即返回降低成本和延迟 ├── retriever.py # 检索层调用SimpleRetriever做RAG ├── evaluate.py # 评估层规则断言日志记录 └── main.py # 主流程编排以上四层我第一次实现这个项目的时候用了200行现在删删减减不到160行。这里贴主要几个模块的核心逻辑方便你抄作业client.py重点import requests import time class LLMClient: def __init__(self, api_key, base_url, model_name): self.api_key api_key self.base_url base_url self.model_name model_name def chat(self, messages, temperature0.7, max_tokens1024, max_retries2): payload { model: self.model_name, messages: messages, temperature: temperature, max_tokens: max_tokens } headers {Authorization: fBearer {self.api_key}, Content-Type: application/json} for attempt in range(max_retries 1): try: resp requests.post( f{self.base_url}/chat/completions, headersheaders, jsonpayload, timeout(10, 60) ) if resp.status_code 429: wait_time 2 ** attempt time.sleep(wait_time) continue if resp.status_code 500: time.sleep(2 ** attempt) continue resp.raise_for_status() data resp.json() return data[choices][0][message][content] except requests.exceptions.Timeout as e: if attempt max_retries: raise TimeoutError(LLM调用超时) from e time.sleep(1) raise Exception(LLM调用失败)cache.py重点哈希键过期策略class ResponseCache: def __init__(self, max_size10000, ttl3600): self.max_size max_size self.ttl ttl self.store {} self.expire {} def _normalize_key(self, messages): import hashlib, json # 关键序列化消息数组去掉temperature等会话性参数只保留确定性的键 canonical json.dumps(messages, ensure_asciiFalse, sort_keysTrue) return hashlib.sha256(canonical.encode()).hexdigest() def get(self, messages): key self._normalize_key(messages) if key in self.store and time.time() self.expire[key]: return self.store[key] return None def set(self, messages, reply): key self._normalize_key(messages) if len(self.store) self.max_size: # 简单淘汰策略清掉一半面向理解生产环境换LRU pop_count self.max_size // 2 for _ in range(pop_count): self.store.pop(next(iter(self.store))) self.store[key] reply self.expire[key] time.time() self.ttlmain.py主流程from client import LLMClient from cache import ResponseCache from retriever import SimpleRetriever from evaluate import assert_reply_quality def build_agent(llm_client, retriever, cache): def handle(user_query, user_idanonymous): # 1. 走缓存 cache_key_messages [ {role: system, content: 你是一名智能客服回答需友好且准确。}, {role: user, content: user_query} ] cached cache.get(cache_key_messages) if cached: return cached, cache # 2. 走RAG检索 docs retriever.search(user_query, top_k3) context \n\n.join([f[片段{i1}]\n{doc} for i, doc in enumerate(docs)]) messages [ {role: system, content: 你是智能客服结合给定资料回答用户问题。资料如下\n context}, {role: user, content: user_query} ] reply llm_client.chat(messages) # 3. 基础校验不达标就降级 try: assert_reply_quality(reply, required_fields[]) except AssertionError: reply 抱歉我暂时无法回答该问题已为您转接人工客服请稍候。 # 4. 写缓存和日志 cache.set(cache_key_messages, reply) return reply, llm return handle # 组装 if __name__ __main__: import os retriever SimpleRetriever() retriever.add_document(1, 产品支持远程办公模式所有文档云端同步。) retriever.add_document(2, 如需退款请在购买后7天内于设置-账单中提交申请。) client LLMClient( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL), model_nameos.getenv(LLM_MODEL_NAME) ) cache ResponseCache() agent build_agent(client, retriever, cache) result, source agent(如何退款, user_iduser_01) print(f[{source}] {result})就这么多一个最小Agent就完整串起来了。缓存命中代表的概念是低延迟低成本检索命中代表知识注入LLM调用代表生成能力规则校验代表质量兜底。所有AI工程的核心组件在这里都能找到对应物。5.2 七个翻车点清单全部是我真金白银踩出来的代码贴出来显得很简单但把它跑上生产坑会一个接一个冒出来。我把踩过的坑归纳成七条每条都是血泪教训看代码时对照一遍能给你省下一个月的排查时间API密钥泄露。最常见最致命。密钥一旦进Git仓库就等于裸奔。正确做法是环境变量加local环境隔离绝不能写死、更不能提交。超时和重试混为一谈。很多人重试机制做得太粗暴超时后立即重试。真实情况是超时往往是因为服务端负载过高或网络拥堵立即重试只会火上浇油。指数退避加抖动是基本修养。降级策略当缓存用。我上面写的回复转人工降级一部分人为了省成本把降级回复也缓存。这就大错特错了——降级回复是错误的退路缓存了错误答案用户会被误导排障时你还会被自己坑。流式响应下不能print。一旦你启用流式SSE输出逐字吐出内容传统的等全部完成再校验逻辑就失效了。必须在校验和流式之间做协商。这是很多新手初次做AI应用时最容易懵的点。评估时只记录正样本。你以为评估做了就够了你记录的只有成功解析的正常返回的。失败样本不记录等于失明。我会把每条失败调用的输入输出、耗时、错误码都留痕后续优化靠的就是这些坏案例。缓存键设计随便合。用户说什么就缓存什么看起来没问题实际上如果你缓存时把temperature这类参数塞进键同一个问题换个参数就变成不同键缓存命中率直线下降。缓存键要尽量确定性参数化。日志不全结构化。一开始图省事用print最后排障时根本没法查。规范是用结构化日志每条log至少包含request_id、user_id、调用耗时、模型参数、token消耗。这七个坑里五个属于不做不会发现做了必踩的类型。我现在做完一个从零项目回头看框架封装的那些为你省心的API发自内心觉得不了解底层的人用框架等于高估了自己的控制力。5.3 从最小骨架到生产系统的演进路线图最后送上一条我的演进建议你按这个顺序来每一步都有明确目标不会迷茫阶段一原生HTTP调用搭骨架跑通输入到输出的闭环建立请求、响应、日志、异常处理的基本盘。阶段二实现多轮会话和缓存解决上下文持久化与高频问答降本掌握状态与缓存设计。阶段三接入检索做RAG让AI应用拥有外部知识能力打通数据处理、索引、召回、排序、上下文组装链路。阶段四评估与回归体系建立离线指标、规则断言和回归集让模型行为变化可感知、可控制。阶段五迈向异步化把模型调用从同步请求改成消息队列Worker异步处理。比如客服助手可以先返回收到正在查询后台生成完成后推送结果这是生产级交互体验的门槛。阶段六可观测性接入指标采集和链路追踪让每一次请求的耗时、成本、缓存命中率全都可视化。这一层做完你的系统才敢谈上生产。这条路线走完你至少需要三到五个月的业余时间。但收获的东西不是会调框架API那种速成技能而是对AI工程整条链路每个关键决策点的判断力。我个人实际做下来最大的体会是AI工程从零开始不是说永不用框架而是你要清楚每个环节的原理才不会在出事时两眼一抹黑。等到你能把上面这些骨架独立实现一遍再回去看LangChain这类工具你会觉得它们不再是黑盒而是一套你可以理解的选择。到那时候你才算真正踏上了ai-engineering这条路的起点。下一步就不用靠别人的轮子了自己选、自己装。
从零构建AI工程：不依赖LangChain，原生HTTP调用与RAG实践
1. 为什么我劝你先别碰LangChain从零构建AI工程的原点这两年ai-engineering成了行业热词朋友圈里人人都在聊Agent、RAG、多模态编排GitHub上的star数像不要钱一样疯涨。但说实话真正动手自己跑通一条AI应用链路的人远比嘴上聊的人少。更少的是那种不靠一切魔法硬生生从底层把AI工程搭起来的人。我见过太多新人的学习路径是这样的先装LangChain再装个向量数据库然后复制一段对话机器人demo跑通之后就觉得自己懂AI工程了。但真到生产环境回调函数报错、上下文管理混乱、模型超时重试策略缺失、成本失控这些事一来整个人直接懵掉。因为框架把所有底层细节都藏起来了你根本不知道问题出在哪一层。所以我这篇想讲的不是如何用某某框架几天做出一个AI应用而是从零开始不依赖任何AI框架自己动手把AI工程的骨架搭出来的完整路径。这篇文章的受众很明确已经会写一点Python知道AI模型是什么听说过LLM、Agent、RAG这些词但想要真正理解AI工程底层逻辑而不是只会套框架的人。核心观点我先抛出来AI工程早期最重要的能力不是会用某个高级框架而是能在没有框架的情况下自己定义架构、自己处理数据流、自己控制成本和延迟。框架迟早会换但工程素养不会。我自己就是这条路走过来的。第一次做AI应用时被LangChain的链式调用折磨得不行——每加一个环节回调栈就深一层出了问题根本不知道是模型返回异常还是Prompt写坏了。后来我一气之下把所有框架代码删干净用原生HTTP请求重新实现了一遍才真正摸清了AI工程的核心链路从提示词组织、上下文管理到模型调用、结果校验、成本控制、持续评估每一步都清清楚楚。这篇文章就当是我把当时踩过的坑、总结出的经验原原本本分享出来。2. 最小可运行骨架一次原生HTTP调用背后藏了多少工程问题2.1 不用SDK用requests直接怼API很多人第一次接触大模型API都是直接在官方文档里复制SDK示例代码。这么做没错但如果你真想搞懂AI工程我强烈建议你先用requests库直接发一次HTTP请求。别管官方封装多优雅你先亲眼看看那个请求体长什么样。一个标准的对话补全API请求本质上就是POST一个JSON到指定接口核心参数就这几个import requests import json # 伪代码示例展示与大模型对话的最小请求体 def call_llm(prompt, api_key, base_urlhttps://api.example.com/v1): payload { model: your-model-name, messages: [ {role: system, content: 你是一位资深技术助手回答需准确、简洁。}, {role: user, content: prompt} ], temperature: 0.7, max_tokens: 1024 } headers { Authorization: fBearer {api_key}, Content-Type: application/json } resp requests.post( f{base_url}/chat/completions, headersheaders, jsonpayload, timeout30 ) if resp.status_code 200: data resp.json() return data[choices][0][message][content] else: raise Exception(fAPI call failed: {resp.status_code} {resp.text})这十几行代码跑通你才算真正触到了大模型API的骨骼。你会直观看到三个关键事实对话是一个数组。messages是消息数组system、user、assistant三条角色反复交替构成上下文。模型没有记忆它的记忆就是你每次请求携带的message数组。理解这一点你就知道了为什么说上下文管理是AI工程的命门。一切参数都是工程参数。temperature、max_tokens这些都直接和成本、质量挂钩。max_tokens设太小回答会被腰斩temperature随意调代码场景下输出会飘。失败模式多到超出想象。网络超时、限流429、内容截断、格式异常这些不是万一会发生而是一定会发生。没有处理这些异常的代码AI应用就是纸糊的。2.2 从一根直通管道到一张防御网必经的六个工程改造把上面的最小代码扩展到真实可用的工程需要给这个光秃秃的HTTP请求套上六层防御网。我逐个说**第一层密钥管理。**没有哪个正经项目会把api_key硬编码在源码里。环境变量读入是我建议的起点。至少做到代码里没有明文密钥这个底线。**第二层超时与重试。**我建议设置连接超时10秒、读超时60秒这种双超时策略。连接超时只管TCP握手建连不覆盖模型生成时间模型生成慢是正常现象直接粗暴给一个总超时会导致大量误杀。重试则要区分场景429限流响应可以重试4xx纯粹是请求错误就不要重试5xx可以重试一到两次。**第三层用户隔离。**多人共用一套API时必须能在代码里区分请求来源。常见做法是拿user_id可以放在请求体里便于平台侧做用量统计和审计追责时不至于抓瞎。**第四层成本配额。**这是几乎所有人第一次做AI工程都会忽略的事。不考虑成本的请求代码上线一晚上账单能让你怀疑人生。在每个用户维度、每天维度上做好token用量统计和上限控制看代码时一定要养成每次请求先算钱的习惯。**第五层结构化日志。**AI应用的调试难点在于不确定性问题。同一个Prompt两次返回可能天差地别。所以必须做日志记录请求参数、返回内容、耗时、token用量、错误信息为追溯和评估提供素材。**第六层结果校验。**模型输出是概率性的你不能假设它一定返回合法JSON、一定遵守约束。工程上要做输出Schema校验不符合就让模型自己纠错或降级使用。这六层防护每一层都值得做成独立的代码模块。等你全做完你的最小可运行骨架已经不是一个demo而是一个初具规模的AI应用了。这个过程就是LangChain这类框架替你做的事——它帮你把它们抽象封装了。但区别在于你自己做一遍出了问题你能定位、能修、能改用了框架你只能瞎试。2.3 单轮对话跑通之后下一步立刻做多轮状态管理单轮对话调通后新手第一个工程化需求就是多轮对话。我给你一条最朴素的路线不用数据库不用向量先在内存里用字典维护每个session_id对应的消息列表。这不是最优方案甚至生产环境很拉胯但它是帮你理解上下文为何重要的最短路径。# 一个最简的多轮会话管理内存版仅用于理解原理 session_memory {} def chat_with_session(session_id, user_message, api_key): # 取历史消息没有则初始化系统提示词 if session_id not in session_memory: session_memory[session_id] [ {role: system, content: 你是帮助用户解答技术问题的小助手。} ] history session_memory[session_id] history.append({role: user, content: user_message}) # 调用大模型 reply call_llm(history, api_key) history.append({role: assistant, content: reply}) # 截断限制长度防止无限膨胀 if len(history) 10: # 保留系统提示词和最近9条消息 session_memory[session_id] [history[0]] history[-9:] return reply你把这个跑起来聊个二三十轮就能感受到上下文管理最本真的痛点历史消息越积越多请求体越来越臃肿成本快速飙升而且模型对很久以前的信息记忆会漂移。这就是所有复杂RAG架构、上下文压缩策略、向量记忆机制要解决的那个最原始的问题原型。从零构建的意义就在这里你能清晰地看见问题的起源自然而然就知道解法为什么长那样。3. 数据工程从零做起检索增强不是魔法是清洗、索引、召回3.1 一个不装外部服务的本地向量检索替代方案聊到RAG检索增强生成大家都喜欢一上来就搞向量数据库。但我必须说如果没有索引和召回这一步的底层概念直接上向量库就是一个黑盒加另一个黑盒。我的建议是先做一个关键词倒排索引的本地检索器。为什么因为它逻辑透明再小的demo也能把数据进、索引建、检索出这一套完整跑通。等你知道检索的本质了再换向量也不是难事。# 简易倒排索引实现剖析检索的底层逻辑 import re from collections import defaultdict class SimpleRetriever: def __init__(self): self.documents {} # doc_id - 文本 self.inverted_index defaultdict(set) # 词 - 包含该词的doc_id集合 def add_document(self, doc_id, text): self.documents[doc_id] text tokens set(re.findall(r\w, text.lower())) for word in tokens: self.inverted_index[word].add(doc_id) def search(self, query, top_k3): tokens re.findall(r\w, query.lower()) scores defaultdict(float) for word in tokens: for doc_id in self.inverted_index.get(word, []): scores[doc_id] 1 # 简单词频打分这里可以替换为TF-IDF ranked sorted(scores.items(), keylambda x: x[1], reverseTrue) return [(doc_id, self.documents[doc_id]) for doc_id, score in ranked[:top_k]]这段代码肯定被懂行的人笑话太原始但这恰恰是它的价值——它把检索的真相摊在你面前文档切词词反查到文档再按命中次数打分排序。向量检索做的事情本质上也是在解决这个查找相关文档的问题只是换了一种更鲁棒的方式。当你看到TF-IDF可以替换Top-K怎么取都对结果有影响这样的关键决策点你就会明白为什么RAG的工程调优空间那么大。3.2 数据清洗的脏活累活喂给检索器的内容决定一切很多踩坑案例里RAG效果不好第一反应是换个更好的向量模型。但就我的经验而言80%的情况下问题压根不在检索算法而在数据进索引之前那一步——你没做清洗也没做切分数据直接是坨脏东西。做数据工程我总结四道工序格式归一。PDF、Word、网页抓取的文本各种乱码、多余空行、特殊字符先清理掉。这一项不做好切出来的chunk全是垃圾进垃圾出。内容去重。语义完全相同或几乎相同的文档片段要合并。不然检索返回三个结果内容却是一模一样的复制粘贴浪费时间也浪费token。粒度切分。这是RAG的核心艺术。切太短语义碎片化检索回来的是不完整的逻辑单元切太长又容易塞进无关噪声成本和延迟双高。我的经验是优先按自然语义段落标题层级切没有标题的按固定窗口如500字叠加重叠overlap取100字。这个比例需要在你的语料上实测没有放之四海皆准的参数。元数据标注。给每个切分片段打上来源、章节、时间等标签在检索后置过滤环节有大用。比如用户问今年的政策可以按时间元数据过滤掉去年的内容。这比在查询词上做文章简单得多。数据工程的每一步都是可以用代码显式控制的。做得多了你会有种每一条数据都经我手看过一遍的掌控感。这感觉在工程里的价值比什么都重要。3.3 从检索到片段到组织成可用上下文有一个容易被忽视的环节我早年做RAG最常犯的错是检索器返回了五个片段我就全部拼接硬塞给模型结果模型答非所问。后来我明白检索结果要经过一道组织工序重排。召回靠粗排比如向量相似度拿到以后可以做一次精排。简单做法是用模型打分、或按规则把与查询有强关键词重叠的排前面。这一步对最终答案质量影响经常是决定性的。筛选。设定相似度阈值不相关的宁可不要。宁可少给不要乱给。编排顺序。把多个片段按相关性高优先、时间新优先等规则排序。你给模型的上下文顺序会影响模型注意力分布不是小事。我在RAG项目里从零做的最大收获其实不是写代码的能力而是对上下文是一种可编程资源的体感。你能拼、能裁剪、能编排模型的输出上限就被你的上下文策略锁死了。这个观念建立起来后任何Retrieval框架在你眼里都只是一堆可替换的零件。4. 系统设计给了我另外半条命延迟、成本、评估缺一不可4.1 延迟预算悖论用户能等5秒但不能等5秒加1毫秒AI应用和传统应用一个很大的区别传统接口延迟你压到100毫秒以下就胜利了AI应用一次模型生成动不动几秒你不可能靠压性能做到即时响应。所以AI工程里的延迟优化思路要换。我自己的做法是延迟预算制先定一条硬基线比如用户从发问到看到开头的总时间不能超过3秒。然后拆解各部分预算环节典型耗时可能依服务波动优化思路网络请求50~200msCDN、就近接入LLM首token返回300ms~3s取决于模型和Prompt长度流式输出是必须的让用户先看到字在动短Prompt能显著压缩首token时间检索/RAG50~500ms索引优化采用缓存后续生成无法压缩与首token逻辑解耦先吐骨架再补细节这里最大认知点在于模型的生成时间你是无法压缩的你能压缩的是外围的检索和延迟预算分配。让用户在首token后感知到有反馈比压缩总延迟更重要。同样一个3秒的API请求一次性转半天才返回和流式快速吐前几个字用户体感天差地别。4.2 成本控制的算账思维从单次调用到规模化预算成本控制是AI工程绕不开的生死线。要控制成本第一步永远是会算账。我给你一个简单的成本评估模型假设你一次会话平均要打2次模型调用每次调用平均输入2000 token、输出300 token。那么一次会话消耗约为4300 token。如果按每百万token $5的价格估算这个数字随市场波动但量级差不多一次会话成本约0.02美元。听上去很少如果你的DAU是1000人每人每天平均发生10次会话那就是0.02美元乘以一万一天就要200美元一个月6000美元一年七万多美元。这几行字算完你立刻就会明白为什么生产级的AI应用都拼命做缓存和减少重复调用为什么上下文裁剪是一项省大钱的工作为什么Prompt压缩、模型蒸馏、本地小模型部署这些技术有巨大的商业价值。成本意识是刻进工程血液里的。每写一次LLM调用我都条件反射地问自己三句话这个调用能不能省能不能用更小的模型能不能缓存4.3 评估三件套离线指标、规则断言和回归集AI工程和传统工程最大的不同在于代码质量测试只能覆盖一部分。传统的单测只能测函数输入输出但模型的输出没法简单地做断言。所以需要一套评估三件套**第一件离线指标。**在固定数据集上算指标比如RAG领域的召回率、命中率。不需要理解语义用机械指标快速判断检索链路有没有退化。上线一个新清洗逻辑后先跑一遍召回指标比人工看十条case高效太多。**第二件规则断言。**针对输出做结构化和规则校验比如输出是不是JSON关键字段是否存在文本长度有没有极端值。我习惯用一个简单的pytest风格函数来做断言自动化程度高重现成本低。def assert_reply_quality(reply, required_fields, max_length2000): # 基础规则断言校验输出基本质量 assert len(reply.strip()) 0, 回复为空 assert len(reply) max_length, f回复超长: {len(reply)} assert not any(bad_word in reply for bad_word in [抱歉我不理解, Network Error]), 回复包含异常占位词**第三件回归集。**它是最重要的兜底防线。每个迭代版本都选一批覆盖典型场景的用例比如100条跑一遍对比新旧版本输出。一旦某条质量明显退化立刻回滚。这个思想抄的软件工程回归测试但它做在模型行为这一层上。没有回归集你的AI应用改代码就像蒙眼开车。我强烈建议在项目第一周就把评估三件套搭建好哪怕粗糙。因为人脑的好像还行式评估在数据量上来后必然失真。5. 完整可复现的例子一个带缓存、降级、评估的最小Agent5.1 模块划分与主流程不到200行做完整闭环把前面三章的东西串成一个最小但完整可运行的项目我用一个智能客服助手作为场景把项目拆成五个模块逻辑清晰到可以直接抄走ai-engineering-mini/ ├── client.py # 大模型API客户端超时、重试、限流 ├── cache.py # 缓存层命中即返回降低成本和延迟 ├── retriever.py # 检索层调用SimpleRetriever做RAG ├── evaluate.py # 评估层规则断言日志记录 └── main.py # 主流程编排以上四层我第一次实现这个项目的时候用了200行现在删删减减不到160行。这里贴主要几个模块的核心逻辑方便你抄作业client.py重点import requests import time class LLMClient: def __init__(self, api_key, base_url, model_name): self.api_key api_key self.base_url base_url self.model_name model_name def chat(self, messages, temperature0.7, max_tokens1024, max_retries2): payload { model: self.model_name, messages: messages, temperature: temperature, max_tokens: max_tokens } headers {Authorization: fBearer {self.api_key}, Content-Type: application/json} for attempt in range(max_retries 1): try: resp requests.post( f{self.base_url}/chat/completions, headersheaders, jsonpayload, timeout(10, 60) ) if resp.status_code 429: wait_time 2 ** attempt time.sleep(wait_time) continue if resp.status_code 500: time.sleep(2 ** attempt) continue resp.raise_for_status() data resp.json() return data[choices][0][message][content] except requests.exceptions.Timeout as e: if attempt max_retries: raise TimeoutError(LLM调用超时) from e time.sleep(1) raise Exception(LLM调用失败)cache.py重点哈希键过期策略class ResponseCache: def __init__(self, max_size10000, ttl3600): self.max_size max_size self.ttl ttl self.store {} self.expire {} def _normalize_key(self, messages): import hashlib, json # 关键序列化消息数组去掉temperature等会话性参数只保留确定性的键 canonical json.dumps(messages, ensure_asciiFalse, sort_keysTrue) return hashlib.sha256(canonical.encode()).hexdigest() def get(self, messages): key self._normalize_key(messages) if key in self.store and time.time() self.expire[key]: return self.store[key] return None def set(self, messages, reply): key self._normalize_key(messages) if len(self.store) self.max_size: # 简单淘汰策略清掉一半面向理解生产环境换LRU pop_count self.max_size // 2 for _ in range(pop_count): self.store.pop(next(iter(self.store))) self.store[key] reply self.expire[key] time.time() self.ttlmain.py主流程from client import LLMClient from cache import ResponseCache from retriever import SimpleRetriever from evaluate import assert_reply_quality def build_agent(llm_client, retriever, cache): def handle(user_query, user_idanonymous): # 1. 走缓存 cache_key_messages [ {role: system, content: 你是一名智能客服回答需友好且准确。}, {role: user, content: user_query} ] cached cache.get(cache_key_messages) if cached: return cached, cache # 2. 走RAG检索 docs retriever.search(user_query, top_k3) context \n\n.join([f[片段{i1}]\n{doc} for i, doc in enumerate(docs)]) messages [ {role: system, content: 你是智能客服结合给定资料回答用户问题。资料如下\n context}, {role: user, content: user_query} ] reply llm_client.chat(messages) # 3. 基础校验不达标就降级 try: assert_reply_quality(reply, required_fields[]) except AssertionError: reply 抱歉我暂时无法回答该问题已为您转接人工客服请稍候。 # 4. 写缓存和日志 cache.set(cache_key_messages, reply) return reply, llm return handle # 组装 if __name__ __main__: import os retriever SimpleRetriever() retriever.add_document(1, 产品支持远程办公模式所有文档云端同步。) retriever.add_document(2, 如需退款请在购买后7天内于设置-账单中提交申请。) client LLMClient( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL), model_nameos.getenv(LLM_MODEL_NAME) ) cache ResponseCache() agent build_agent(client, retriever, cache) result, source agent(如何退款, user_iduser_01) print(f[{source}] {result})就这么多一个最小Agent就完整串起来了。缓存命中代表的概念是低延迟低成本检索命中代表知识注入LLM调用代表生成能力规则校验代表质量兜底。所有AI工程的核心组件在这里都能找到对应物。5.2 七个翻车点清单全部是我真金白银踩出来的代码贴出来显得很简单但把它跑上生产坑会一个接一个冒出来。我把踩过的坑归纳成七条每条都是血泪教训看代码时对照一遍能给你省下一个月的排查时间API密钥泄露。最常见最致命。密钥一旦进Git仓库就等于裸奔。正确做法是环境变量加local环境隔离绝不能写死、更不能提交。超时和重试混为一谈。很多人重试机制做得太粗暴超时后立即重试。真实情况是超时往往是因为服务端负载过高或网络拥堵立即重试只会火上浇油。指数退避加抖动是基本修养。降级策略当缓存用。我上面写的回复转人工降级一部分人为了省成本把降级回复也缓存。这就大错特错了——降级回复是错误的退路缓存了错误答案用户会被误导排障时你还会被自己坑。流式响应下不能print。一旦你启用流式SSE输出逐字吐出内容传统的等全部完成再校验逻辑就失效了。必须在校验和流式之间做协商。这是很多新手初次做AI应用时最容易懵的点。评估时只记录正样本。你以为评估做了就够了你记录的只有成功解析的正常返回的。失败样本不记录等于失明。我会把每条失败调用的输入输出、耗时、错误码都留痕后续优化靠的就是这些坏案例。缓存键设计随便合。用户说什么就缓存什么看起来没问题实际上如果你缓存时把temperature这类参数塞进键同一个问题换个参数就变成不同键缓存命中率直线下降。缓存键要尽量确定性参数化。日志不全结构化。一开始图省事用print最后排障时根本没法查。规范是用结构化日志每条log至少包含request_id、user_id、调用耗时、模型参数、token消耗。这七个坑里五个属于不做不会发现做了必踩的类型。我现在做完一个从零项目回头看框架封装的那些为你省心的API发自内心觉得不了解底层的人用框架等于高估了自己的控制力。5.3 从最小骨架到生产系统的演进路线图最后送上一条我的演进建议你按这个顺序来每一步都有明确目标不会迷茫阶段一原生HTTP调用搭骨架跑通输入到输出的闭环建立请求、响应、日志、异常处理的基本盘。阶段二实现多轮会话和缓存解决上下文持久化与高频问答降本掌握状态与缓存设计。阶段三接入检索做RAG让AI应用拥有外部知识能力打通数据处理、索引、召回、排序、上下文组装链路。阶段四评估与回归体系建立离线指标、规则断言和回归集让模型行为变化可感知、可控制。阶段五迈向异步化把模型调用从同步请求改成消息队列Worker异步处理。比如客服助手可以先返回收到正在查询后台生成完成后推送结果这是生产级交互体验的门槛。阶段六可观测性接入指标采集和链路追踪让每一次请求的耗时、成本、缓存命中率全都可视化。这一层做完你的系统才敢谈上生产。这条路线走完你至少需要三到五个月的业余时间。但收获的东西不是会调框架API那种速成技能而是对AI工程整条链路每个关键决策点的判断力。我个人实际做下来最大的体会是AI工程从零开始不是说永不用框架而是你要清楚每个环节的原理才不会在出事时两眼一抹黑。等到你能把上面这些骨架独立实现一遍再回去看LangChain这类工具你会觉得它们不再是黑盒而是一套你可以理解的选择。到那时候你才算真正踏上了ai-engineering这条路的起点。下一步就不用靠别人的轮子了自己选、自己装。
