拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

从零手写AI工程:RAG、Agent与上下文管理实战指南

从零手写AI工程:RAG、Agent与上下文管理实战指南

1. 项目拆解:AI工程从零到一到底在学什么

先聊点扎心的:现在网上聊 AI 开发的帖子,十个里有八个是“调 API 三分钟做个聊天机器人”。真到了生产环境,需求一复杂,数据一脏,并发一上来,那些 demo 全得推翻重写。我见过太多人 LangChain 用得飞起,RAG 跑起来也能出结果,但问一句“召回率怎么算”“chunk_size 为什么取 512”“上下文超长了你怎么办”,立马卡壳。

“ai-engineering-from-scratch”这个项目,核心思想全在这几个字里:不借助高级封装,从零开始把 AI 应用的每一层亲手搭一遍。

为什么值得这么干?理由很简单。框架封装度高,是把双刃剑。它能让你三天出 demo,也能让你三年看不懂底层。一旦线上出问题,黑盒的排查成本远超你当初省下的那点时间。写这个项目的目的,就是把你从“调用者”掰回“构建者”。从头写一遍调用逻辑、检索逻辑、Agent 循环、评估链路,你才知道每个组件到底在解决什么问题、性能瓶颈在哪、边界条件在哪。

适合谁来参考呢?三类人:

  • 已经能用 LangChain/LlamaIndex 写 demo,但想更深一层理解原理的开发者;
  • 正准备做 AI 应用落地,不想被框架绑架、想自己掌控全流程的工程师;
  • 刚入门,想知道“AI 工程”到底包含哪些环节、应该怎么系统学习的新手。

这个仓库的理想学习路径,不是按框架文档走,而是按照 AI 应用的物理结构从薄到厚拆解:从大模型调用 → 上下文管理 → 检索增强 → 工具调用 → 评估反馈 → 部署监控。每加一层,你就更接近一个能扛住真实业务压力的系统。

2. 为什么“从零手写”比“直接上框架”更值得

我用个真实经历说明。之前做个内部知识库问答,最早图省事直接用 LangChain 的RetrievalQA,代码二十行,跑起来确实能答。结果上线第一天就翻车:用户问法稍有变化,检索回来的段落全是噪声,答非所问。我盯着 LangChain 的链式调用日志看了半天,不知道它内部做了哪些改写、用了什么检索策略,整个人懵住。

后来沉下心,把完整的 RAG 链路用原生代码重写了一遍,才发现问题出在 query 改写默认做了一堆花活,加上 top_k 取值过大,把无关段落一并塞进了上下文。两行参数的事,换成直接看底层代码,立刻一目了然。“从零手写”最大的价值,是让你拥有出问题时的定位能力。

再换个类比。你用自动挡开车,舒服是舒服,但车半路抛锚,你连引擎盖都不会开。从零手写就是逼你拆一遍发动机。虽然前期慢,但后面你对每个零件的脾气都门儿清。

从零手写还有几个实打实的优势:

  • 灵活可控:生产环境的需求千奇百怪,框架预设的管线往往满足不了。自己搭,想怎么改怎么改。
  • 便于排障:每一层都是你自己写的,日志、错误、性能瓶颈全在掌握之中。
  • 规避兼容性地狱:LangChain 的版本更新有多频繁、Breaking Change 有多痛,经历过的人都懂。自己写,依赖少,长期稳定。
  • 代码瘦身:调用一个几十兆的框架,只为用其中两个函数,冗余度高。裸写往往几千行代码就覆盖了核心业务。

当然,我不是劝你完全抛开 LangChain。等你亲手写过一遍底层逻辑,再回去用框架,你会看得懂它的设计、知道怎么定制。那时候框架才真正为你所用,而不是反过来。

3. 核心环节拆解:从模型到应用,每一层都在解决什么问题

3.1 模型层:打好地基,选对调用方式

AI 工程最底层的二选一:用托管 API,还是自部署模型。

托管 API 省心,但会碰到三座大山:成本不可控、数据出域、限流延迟。自部署模型用开源权重(Qwen、Llama、DeepSeek 等),前期折腾但长期成本低、可控性强。

工程上的建议是接口统一、模型可替换。所有代码不要跟某一家厂商的 SDK 强绑定,统一走 OpenAI 兼容协议。这样今天用 GPT-4o,明天换成 Qwen-Max,只改一个环境变量就行。这是我跟过的项目里最值得的一个决策——从没被单一厂商锁死过。

自部署的话,轻量场景用 Ollama 就够了,一行命令拉起模型,适合开发调试。生产环境追求吞吐,推荐 vLLM,PagedAttention 技术能把显存利用率拉高一大截,QPS 轻松翻倍。跑过 7B 模型的人应该深有体会,vLLM 与原生 transformers 部署的差距是肉眼可见的。

3.2 上下文工程:用一半费用做两倍效果

上下文管理是 AI 应用被低估的重头戏。很多工程问题,本质上就是上下文没管好。

  • 裁剪:对话变长后,不是无穷加 token,而是把早期消息做摘要压缩。核心思路是“用更少的 token 保留关键信息”,窗口能塞下更多有效内容。
  • 结构化注入:不要把整本书塞进 prompt,先用结构化提取,把关键事实压缩成条目,再喂给模型。
  • 动态检索:不靠手工控制上下文,靠向量检索动态取最相关的片段。这就是 RAG 的雏形。

上下文管理的核心指标是“有效信息密度”。同样 8K 窗口,A 方案只能容纳 20 轮对话,B 方案能容纳 50 轮且关键信息不丢失,B 的架构明显更健康。这行当里,省钱和省事往往是一回事。

3.3 RAG 层:谁说加个向量库就叫 RAG

RAG 看起来简单:文档切割、向量化、检索、拼接。实际工程里,每个环节都有讲究。

切分文档是第一个暗坑。固定 500 字切一刀,会把一段完整语义拦腰截断。我踩过之后改成了语义分段:用段落 Markdown 标题、列表、空行做自然边界,必要时结合滑动窗口重叠。分块质量直接决定检索质量,这步偷懒,后面全白搭。

Embedding 模型选型也不能忽视。中文场景,bge-m3系列效果稳且开源可商用,text-embedding-3-large也很能打但按量付费、量大肉疼。实测下来,换成中文优化过的 embedding 模型,top-5 命中率能提升 15%-20%,你检索效果差的锅,模型至少背一半。

检索策略同样不能无脑 top-k。固定取前 5 段必然带噪声。更好的做法是:向量检索取前 20 → 重排模型(如bge-reranker)精排取前 3-4。步骤多了,精度立竿见影。重排是 RAG 效果从“能用”到“好用”的分水岭。

3.4 Agent 层:最小可行循环

Agent 核心不是概念,是那个循环:理解任务 → 选择合适的工具 → 调用并观察结果 → 决定下一步。工程实现上,看清每个环节:

  • 意图识别:靠 prompt 明确列出可用工具和适用场景,模型负责路由。
  • 工具调用:原生 OpenAI Function Calling 或开源模型自带的工具调用能力,把结构化参数传给函数。
  • 循环控制:设最大迭代轮数(一般 5-8 轮)、条件终止、异常重试。不设上限,Agent 会自己死循环烧光你的预算,别问我是怎么知道的。
  • 记忆组织:短期记忆存当前任务上下文,长期记忆靠外部存储(典型如向量库)。

自己从零实现 Agent 的价值在于:你能精准感知每一步消耗的时间、token 和成功率,而不是面对一个“智能体框架”干瞪眼。看日志从“天书”变成“正常人话”,瞬间舒服。

4. 从零实操:一步步搭建你自己的 AI 应用

这一节是硬货。我按“最小前向路径”走一遍,完整体验 AI 应用的诞生过程。环境默认 Python 3.10+。

4.1 第一步:搭一个最简 LLM 调用服务

先不碰任何框架,裸调模型 API。

# requirements.txt openai>=1.30.0 python-dotenv fastapi uvicorn
# app.py import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() _client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL") or "https://api.openai.com/v1" ) def chat(messages, model=None): model = model or os.getenv("LLM_MODEL", "gpt-4o-mini") response = _client.chat.completions.create( model=model, messages=messages, temperature=0.3, stream=True ) full_text = "" for chunk in response: delta = chunk.choices[0].delta.content if delta: full_text += delta return full_text

几个关键点备注一下:

  • temperature=0.3:日常问答降到 0.3 能显著减少幻觉和发散。要创意写作再上调。
  • 流式接口:用户体验差异巨大。等 20 秒全量返回,用户早跑了;逐字输出,5 秒内就有“在干活”的感觉。生产环境务必流式。
  • 统一 base_url:换模型厂商零改动。我在.env里放OPENAI_BASE_URL,切换供应商时只改配置。

4.2 第二步:手写一个 RAG 全链路

这是整个项目的核心部分。完整实现下来的记忆点异常清晰,比看十篇教程管用。

# rag.py from openai import OpenAI import sqlite3 class VectorStore: """零依赖向量库:把向量存 SQLite,检索时全表扫""" def __init__(self, db_path): self.conn = sqlite3.connect(db_path) self.conn.execute("CREATE TABLE IF NOT EXISTS vectors (id TEXT PRIMARY KEY, embedding TEXT, content TEXT)") self.client = OpenAI() def embed(self, text): resp = self.client.embeddings.create( model="text-embedding-3-small", input=text ) return resp.data[0].embedding def add(self, doc_id, content): vec = self.embed(content) self.conn.execute( "INSERT INTO vectors VALUES (?, ?, ?)", (doc_id, json.dumps(vec), content) ) self.conn.commit() def search(self, query, top_k=5): q_vec = self.embed(query) rows = self.conn.execute("SELECT * FROM vectors").fetchall() scored = [] for doc_id, emb_json, content in rows: emb = json.loads(emb_json) # 余弦相似度 dot = sum(a*b for a, b in zip(q_vec, emb)) norm_q = sum(a*a for a in q_vec) ** 0.5 norm_e = sum(b*b for b in emb) ** 0.5 score = dot / (norm_q * norm_e + 1e-9) scored.append((score, content)) scored.sort(reverse=True, key=lambda x: x[0]) return scored[:top_k]

思路很直白:全表扫描 + 余弦相似度。性能远不如专业向量库,但作为教学训练,逻辑一目了然,便于理解核心机制。生产环境记得换 Milvus 这类专用引擎。

实际文档切分我用的是递归字符切分器,附上核心逻辑:

def chunk_text(text, chunk_size=512, overlap=80): """按分隔符优先策略切分,保留上下文重叠""" separators = ["\n\n", "\n", "。", "!", "?", ";"] current = "" for char in text: current += char if len(current) >= chunk_size: # 找最近的边界符切开 for sep in separators: idx = current.rfind(sep) if idx != -1: yield current[:idx+len(sep)] current = current[idx+len(sep):] break else: yield current current = "" if current: yield current

切分参数不是拍脑袋。512 是中等长度平衡点:太长,embedding 平均了语义导致检索不精准;太短,上下文碎片化丢失逻辑。80 的重叠让相邻片段保持线索。文本结构化越强的文档,切分效果越依赖分隔符匹配,这是我实测多轮之后的结论。

最后“检索 + 生成”拼装代码如下:

def rag_query(question): chunks = store.search(question, top_k=5) context = "\n\n".join(chunk for _, chunk in chunks) messages = [ {"role": "system", "content": f"基于以下资料回答问题。资料不充分时,直接说不知道,不要编造。\n\n资料:\n{context}"}, {"role": "user", "content": question} ] return chat(messages)

这套组合足以支撑起一个企业内部知识库问答应用。生产化还要加上:文档更新时自动重算 embedding、切分策略按文档类型配置、检索结果缓存等。但骨架就是这样。

4.3 第三步:给 Agent 加上工具调用

从零实现 Agent,绕不开工具调用。我用最简单的天气查询示例展示设计模式:

import json TOOLS = [ { "type": "function", "function": { "name": "get_weather", "description": "查询城市当前天气", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名"} }, "required": ["city"] } } } ] def get_weather(city): # 这里接真实天气 API return f"{city}:多云,24℃,东风3级" def agent_run(user_input, max_steps=5): messages = [{"role": "user", "content": user_input}] for step in range(max_steps): resp = client.chat.completions.create( model="gpt-4o-mini", messages=messages, tools=TOOLS, tool_choice="auto" ) msg = resp.choices[0].message if not msg.tool_calls: return msg.content messages.append(msg) for tool_call in msg.tool_calls: args = json.loads(tool_call.function.arguments) result = globals()[tool_call.function.name](**args) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result }) return "达到最大步数,已停止。"

这个模式就是所有 Agent 框架的底层骨架。max_steps=5为了防止模型陷入死循环。我实测过,不设上限的 Agent 经常在工具调用里转圈出不来,token 烧得人心疼。这也是生产环境永不省略的保险丝。

4.4 第四步:评估和回归——调优的照妖镜

没有评估体系的 AI 应用优化,全凭运气。自己手写 RAG 时,一定要把评估搭起来:

EVAL_SET = [ {"question": "公司的年假制度是怎样的?", "expected_keywords": ["累计工作年限", "年假天数"]}, {"question": "报销流程需要几步?", "expected_keywords": ["提单", "审批", "财务打款"]}, # ... 大概50条以上,覆盖各业务场景 ] def evaluate(): total, hit = 0, 0 for item in EVAL_SET: answer = rag_query(item["question"]) total += 1 if all(kw in answer for kw in item["expected_keywords"]): hit += 1 return hit / total

这个简单精确的“关键词命中率”指标,能当回归测试的守门员。每次改切分策略、换 embedding、调 top_k,跑一遍看分数变化——涨了就保留,跌了就打回。

更精细一点,可以用LLM-as-judge:

def judge_answer(question, answer, reference): prompt = f"""你是评估助手。判断以下回答是否准确。 问题:{question} 参考答案:{reference} 模型回答:{answer} 输出:完全正确 / 部分正确 / 错误""" return chat([...]) # 返回上述三选一

实践下来,先埋头搭好了评估集再做优化,效率是高一个层级的。没有度量,一切调优都等于蒙眼开车。

4.5 第五步:部署上线与监控

工程化最后一步是让应用真正 7x24 跑起来。上 FastAPI 封装:

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class Query(BaseModel): question: str @app.post("/query") def query_endpoint(q: Query): answer = rag_query(q.question) return {"answer": answer}

服务化暴露 HTTP 接口,前端/后端系统都能直接调用。别用 Jupyter Notebook 当生产服务,掉线一次口碑崩一次。

部署容器化 Dockerfile 要点:

FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

重头戏在监控。我强烈建议每个 AI 工程在日志之外加一层可观测性埋点,至少包含:

  • Token 消耗和成本:每轮对话花了多少钱,按用户聚合统计;
  • 响应延迟分段:embedding 耗时、检索耗时、生成耗时分开记录,定位瓶颈一目了然;
  • 检索命中率:用户点没点“有帮助”,是最直接的反馈信号;
  • 失败率:4xx/5xx、超时、拒绝率。

实测过 Langfuse,埋点完成后 dashboard 能直接显示每次调用的完整轨迹,花钱花在哪清清楚楚,排查问题效率极高。

5. 实战避坑:我在从零构建中踩过的真实问题

跟完上述步骤,你大概率会遇到下面这些坎,我提前把雷给你排了。

5.1 上下文吞金兽:prompt 越堆越多

最典型的问题。对话超过 20 轮,上下文里塞满了历史,每次调用把老账都翻出来算钱,响应也慢得像蜗牛。

排查思路:先打印每次 request 的 token 数,把缺失的信息暴露出来。解决三板斧:早期消息摘要化(详情见 3.2);窗口长度限制+滑动截断;对用户输入做长度预检,超额先压缩再入模型。

5.2 检索命中率虚低:问题出在切分,不是模型

有段时间我把切分调成 256 字后,检索全乱套。查了半天才发现固定长度切分把语义拆碎了。比如“报销流程”四个字段落在前一片,详细步骤说明在后一片,检索到的片段全是残肢。

解决思路:切分策略与文档结构强相关。Markdown 标题明确的,先按标题切;法律/说明书类句子自成段的,按句子切。确定策略后在评估集上跑分对比,别用感觉选方案。

5.3 模型乱编答案:幻觉压不下去

幻觉的根源,是模型在“知识空白区”做了补全。我的处理矩阵:

  • 系统提示里写死“资料里没写就回答不知道,禁止自行补充”;
  • temperature从 0.7 压到 0.2,效果很明显;
  • 要求输出引用来源,答案自带可追溯性;
  • 对关键数据问题走“检索-抽取-比对”逻辑,让模型输出结论后必须附上原文片段。

5.4 线上响应慢:卡在生成环节

AI 应用的延迟大头几乎都在生成阶段。RAG 链路里 embedding 加检索大概占 300ms,模型生成 500 token 要 6-10 秒。

优化手段:流式输出先让用户动起来;小问题用小模型路由(比如简单问答走 4o-mini,复杂逻辑走大模型);对高频问题做 exact-match 缓存;需要重逻辑的场景加前缀缓存。

5.5 第三方 API 抖动:重试策略直接决定体验

线上跑了一段时间你就会发现,再好的 API 也会偶尔抽风。工程上要实装:指数退避重试(1s、2s、4s,最大三次)、超时设置(connect 10s / read 60s)、失败降级策略(主模型故障切备选模型),这些就是应用稳定性的压舱石。

6. 从零到产线:下一个升级目标是哪三个方向

“from-scratch”项目跑通之后,整个体系的升级路线一般有三个优先方向,按投入产出比排优先级:

第一优先级:数据飞轮。把线上用户真实提问采集下来,每周筛选 badcase,补充进评估集。做评估不是为了这一版,是为了下一版。跑得越久,评估集越厚,系统调优的目标越清晰。

第二优先级:多路召回融合。单路向量检索能力有上限,试着叠加 BM25 关键词检索、SQL 结构化查询,最后用重排模型融合。QA 准确率通常能再上 5-10 个百分点。

第三优先级:领域微调(若确实需要)。RAG 跑通后,如果某些专业术语的生成风格始终不对,考虑用 LoRA 做一次轻量微调。这步是加分项,前两个方向没做扎实的情况下不建议碰。

实操中的心得体会,永远记住一句话:先跑通,再调优,先动手,再换框架。只要核心链路是你亲手搭的,后续无论接 LangGraph、上 Agent 工作流、做多模态,都不会慌。地基是你打的话,盖楼心里就有底了。

返回列表