在大型云厂商的 AI 项目里做过一轮端到端开发之后通常会对一个问题印象深刻真正让 AI 应用难以落地的往往不是模型选型而是把模型接进业务过程里的工程链路。需求拆解、数据清洗、文档分块、检索召回、提示词构造、结果评估、服务部署、成本控制和日志追踪任何一个环节缺位都会在某个瞬间让项目卡住。有人以为只要调通模型接口就完成了 AI 开发实际上那只是起点。这篇文章会围绕一条真实可复现的主线展开如何从零构建一个可交付的知识库问答 Agent覆盖 RAG 检索、Agent 工具调用、评测、部署和排错。学完这套思路碰到 AI 编程、AI Agent 开发、模型部署和 AI 应用学习路线等话题时都能找到对应的工程位置。AI 应用开发看起来门槛低真正做起来却到处都是细节。下面先把整个系统拆开搞清楚每个环节为什么存在然后再进入代码实现。1. 先把“构建 AI 系统”拆成四条可管理的链路1.1 真正的瓶颈在工程化不在模型选择很多团队第一次尝试 AI 应用时会花大量时间比较模型最后选定一个效果不错的模型就认为项目可以继续推进。实际上模型能力只是整条链路的起点。一个知识库问答系统能不能用取决于用户问题进来之后是否真的能检索到相关资料资料是否足够完整模型是否严格按资料回答回答里给的信息能不能追溯到原文。这些问题没有一个能通过“换更好的模型”直接解决。模型只会基于你提供的上下文生成回答如果上下文里没有答案或者上下文本身是错误信息模型能力再强也救不回来。这也是为什么在大规模 AI 项目里工程化能力比模型选型更影响交付质量。所谓工程化就是把一个想法变成稳定、可评估、可运维的系统的全过程。对知识库问答这类应用来说至少需要拆成四条链路。1.2 四条链路输入、处理、输出、反馈输入链路负责处理用户问题。用户输入往往带有噪声可能是错别字、口语化表达也可能混入了与当前业务无关的内容。输入链路要做的是清洗问题、补全必要信息、判断权限并决定是否需要进入问答主流程。处理链路是核心。它负责意图识别、检索增强、工具调用、上下文组装和多轮对话管理。RAG 和 Agent 都属于这一层。处理链路的质量直接决定回答是否有依据。输出链路负责把模型生成的文本变成可用的业务结果。包括格式约束、引用校验、敏感信息过滤和异常兜底。很多系统在验证 demo 时效果不错一上线就出问题问题往往出在输出链路没有做后处理。反馈链路负责把日志、用户反馈、评测结果收集起来形成持续改进的数据来源。没有反馈链路系统就会停留在“凭感觉调 prompt”的阶段。链路主要工作失败表现输入链路清洗、补全、权限控制输入稍有变化回答差异明显处理链路RAG、Agent、上下文组织回答无依据工具调用错误输出链路格式约束、引用、后处理输出不可使用无法溯源反馈链路评测集、日志、回归同一问题反复出现1.3 大厂经验里真正能带走的能力在大型云厂商的大规模 AI 项目里真正有价值的东西并不是某个具体业务代码而是围绕这四条链路沉淀下来的方法。业务代码会过时项目可能会因为组织调整而终止但数据清洗规则、评测集、prompt 模板、部署手册和排错清单这些资产可以迁移到任何新的 AI 项目里。可迁移能力落地方式数据清洗规则沉淀成脚本和检查清单评测集独立维护每次改动后回归Prompt 模板版本化管理记录每次调整原因部署方案形成检查单和回滚手册可观测性统一日志字段和 trace 关联理解了这四条链路再去做环境准备和代码实现就会清楚每一行代码到底在解决哪一个问题。2. 环境准备先跑通最小可复现的本地实验环境2.1 学习环境需要什么学习 AI 应用开发不需要一开始就搭建高可用的生产集群。最小的环境只需要三样东西Python 3.10 或更高版本一个可用的对话模型接口和 Embedding 模型接口一个本地开发工具链包括虚拟环境、编辑器、命令行向量数据库在最小阶段也不是必须的。可以先用普通列表存储文档向量用余弦相似度做检索。当文档量增长到几千条以上再引入专门的向量数据库。这个原则叫“最小闭环优先”。先用最少的组件跑通一次问答再逐步替换成生产级组件。如果一开始就引入消息队列、向量集群、模型网关学习成本会淹没核心思路。组件作用最小方案生产方案对话模型生成回答官方 API模型网关、多模型路由Embedding 模型将文本向量化官方 API批量任务与缓存向量存储保存文本向量Python 列表高可用向量数据库密钥管理保存 API Key.env 文件密钥管理服务监控观察运行状态无日志、trace、告警2.2 安装依赖和配置密钥先创建一个项目目录并在里面建立 Python 虚拟环境。mkdir ai-rag-agent cd ai-rag-agent python3 -m venv .venv source .venv/bin/activate然后安装最小依赖。pip install openai python-dotenv numpy如果要在后面加接口服务再安装 FastAPI 相关依赖。pip install fastapi uvicorn接下来创建环境变量文件。openai 新版 SDK 会自动读取OPENAI_API_KEY环境变量不需要在代码里硬编码密钥。OPENAI_API_KEY你的密钥 OPENAI_CHAT_MODELgpt-4o-mini OPENAI_EMBEDDING_MODELtext-embedding-3-small如果使用的是模型服务商的兼容接口可以在创建客户端时指定base_url模型名也要替换成服务商实际提供的名称。from openai import OpenAI import os client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), # 可选兼容网关时设置 )这里要注意不同版本的 openai SDK 在接口细节上可能有差异。落地时先运行一个最小调用确认client.chat.completions.create和client.embeddings.create在当前版本中可用再继续写上层逻辑。2.3 学习环境与生产环境的差异学习环境追求“能跑”生产环境追求“能长期稳定跑”。区别不只是组件丰富度而是每一类问题都要有明确处理方案。项目学习环境生产环境密钥.env 文件密钥管理服务禁止入库数据样例文档脱敏、权限控制、版本管理错误处理直接抛异常重试、fallback、告警延迟可接受超时控制、限流、缓存成本不考虑预算告警、token 统计下面的实现会在学习环境下运行但每一步都会指出生产环境需要补什么。3. 实现一个 RAG 知识库问答系统3.1 整体链路设计RAG 全称是 Retrieval-Augmented Generation意思是“检索增强生成”。核心思路很直接不直接让模型凭记忆回答而是先从知识库中检索出与问题相关的片段再把片段作为上下文交给模型生成回答。这样做有三个好处回答可以溯源到具体资料知识更新不需要重新训练模型减少模型胡编乱造的概率最小 RAG 链路如下将原始文档加载到程序中把长文档切分成小块对每个小块生成向量将向量保存到内存或向量数据库用户提问时把问题向量化用余弦相似度检索最相关的 top_k 小块将相关小块组装成 prompt调用对话模型生成回答3.2 文档加载与分块文档加载本身不复杂复杂的是分块。分块大小会直接影响检索质量。分块过大一个块里可能包含多个主题检索时命中噪声的概率变高分块过小上下文不完整模型无法理解完整逻辑。下面是一个按字符窗口切分的简单实现。from pathlib import Path def load_text(path: str) - str: return Path(path).read_text(encodingutf-8) def split_chunks(text: str, chunk_size: int 500, overlap: int 50) - list[str]: chunks [] start 0 n len(text) while start n: end min(start chunk_size, n) chunks.append(text[start:end]) if end n: break start max(end - overlap, 0) return [chunk.strip() for chunk in chunks if chunk.strip()]这里的chunk_size控制每块长度overlap控制相邻块之间的重叠长度。重叠是为了避免一个问题正好落在两个块的交界处导致检索时两边都只命中一半信息。实际项目中通常会先按文档结构切分例如 Markdown 标题、PDF 章节、表格边界然后再对块做大小控制。如果原始材料没有明确给出最佳分块参数落地前先准备一个小型测试集对比不同参数下的检索效果再决定使用多大的窗口。注意分块策略没有绝对最优。它和你文档的类型、用户问题的长度、Embedding 模型的输入限制都相关。3.3 向量化与检索向量化的目标是把一段文本映射成一组浮点数数组让语义相近的文本在向量空间中距离更近。Embedding 模型是专门为这个任务训练的模型。from openai import OpenAI import numpy as np client OpenAI() def embed_texts(texts: list[str]) - list[list[float]]: resp client.embeddings.create( modeltext-embedding-3-small, inputtexts, ) return [item.embedding for item in resp.data] def cosine_similarity(a: list[float], b: list[float]) - float: vec_a np.array(a) vec_b np.array(b) return float(np.dot(vec_a, vec_b) / (np.linalg.norm(vec_a) * np.linalg.norm(vec_b))) def retrieve( query: str, chunks: list[str], chunk_embeddings: list[list[float]], top_k: int 3, ) - list[tuple[str, float]]: query_embedding embed_texts([query])[0] scored [] for idx, emb in enumerate(chunk_embeddings): score cosine_similarity(query_embedding, emb) scored.append((score, idx)) scored.sort(keylambda x: x[0], reverseTrue) return [(chunks[idx], float(score)) for score, idx in scored[:top_k]]检索的原理不复杂但有两个关键点需要注意。第一top_k不是越大越好。上下文越长模型处理成本越高无关信息越多反而可能干扰回答。常见初始值是 3 到 5再根据评测结果调整。第二余弦相似度只是最小实现。当 chunk 数量达到数万条时每次都把全部向量取出来算一遍性能会明显下降。这时候需要引入向量数据库例如 Chroma、Qdrant、Milvus让数据库完成相似度检索。3.4 Prompt 组装与生成检索到相关片段之后下一步是把片段组装成 prompt并加上约束规则。def build_prompt(question: str, retrieved: list[tuple[str, float]]) - str: context \n\n.join( f[{i 1}] {chunk} for i, (chunk, _) in enumerate(retrieved) ) return f请根据下面资料回答用户问题。 规则 1. 优先使用资料中的信息。 2. 资料里找不到答案时明确回答“资料中未找到相关信息”不要编造。 3. 引用资料时在句末标注编号例如 [1]。 资料 {context} 用户问题{question} 请回答 def answer( question: str, chunks: list[str], chunk_embeddings: list[list[float]], top_k: int 3, ) - tuple[str, list[tuple[str, float]]]: retrieved retrieve(question, chunks, chunk_embeddings, top_k) prompt build_prompt(question, retrieved) resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0.2, ) content resp.choices[0].message.content return content, retrieved这个 prompt 里的三条规则各有意义。第一条让模型优先使用资料减少编造第二条给模型一个安全出口让它敢于说不知道而不是硬编一个答案第三条要求引用编号便于用户核对原始资料。temperature设置为 0.2是为了减少回答的随机性。知识库问答场景希望输出稳定、可复现因此较低的温度更合适。如果是创意写作可以调高。3.5 从 RAG 升级到 AgentRAG 只能从给定的知识库里检索信息。Agent 则多了一个能力调用外部工具。当问题需要查询实时数据、写文件、查数据库、调业务接口时Agent 可以决定调用哪个工具并把工具返回结果作为下一步推理的上下文。这种机制通常被称为 ReAct 循环模型思考下一步行动执行动作观察结果再继续推理直到能够给出最终回答。下面是一个简化版 Agent 示例演示工具调用循环。import json def get_weather(city: str) - dict: return {city: city, temp: 22, unit: celsius} def search_news(topic: str) - dict: return {topic: topic, news: [示例新闻一, 示例新闻二]} TOOLS [ { type: function, function: { name: get_weather, description: 查询指定城市的天气, parameters: { type: object, properties: { city: { type: string, description: 城市名例如 北京, } }, required: [city], }, }, }, { type: function, function: { name: search_news, description: 搜索指定主题的新闻, parameters: { type: object, properties: { topic: { type: string, description: 新闻主题, } }, required: [topic], }, }, }, ] def call_tool(name: str, arguments: str) - dict: args json.loads(arguments or {}) if name get_weather: return get_weather(**args) if name search_news: return search_news(**args) return {error: unknown tool} def run_agent(question: str, max_steps: int 5) - str: messages [ {role: system, content: 你是智能助手需要外部信息时调用工具。}, {role: user, content: question}, ] for _ in range(max_steps): resp client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolsTOOLS, ) msg resp.choices[0].message if not msg.tool_calls: return msg.content messages.append(msg) for call in msg.tool_calls: result call_tool(call.function.name, call.function.arguments) messages.append( { role: tool, tool_call_id: call.id, content: json.dumps(result, ensure_asciiFalse), } ) return 未能在限定步数内完成回答这个循环的关键在于模型返回一个tool_calls列表程序执行对应的工具函数然后把工具结果以roletool的消息追加回去。模型看到工具结果后可以生成最终答案也可以继续调用下一个工具。生产环境的 Agent 会比这个复杂很多必须考虑参数校验、工具权限、超时控制、重试、日志追踪和多轮记忆管理。这里的最小实现只用来理解机制。注意openai SDK 不同版本对tool_calls的字段名和消息结构可能有差异。使用前先打印一次resp确认当前版本的实际结构。4. 用评测集驱动质量改进4.1 为什么必须自建评测集很多人在开发阶段只靠几个手写问题验证效果输入几轮发现“看起来差不多”就认为系统已经完成。实际上这种方式会隐藏大量问题。因为同一个 prompt 在几个样例上表现好不代表换一批文档后仍然稳定。自建评测集的意义在于把质量判断从主观感受变成可重复执行的检查。每次修改分块参数、prompt 或模型都应该跑一遍同样的评测集对比改动前后的结果。否则你根本不知道一次修改到底改好了什么还是改坏了一个自己没注意到的场景。评测集至少要覆盖以下几类问题知识库中一定存在答案的问题知识库中完全找不到答案的问题答案分散在多个文档片段里的问题用户输入包含错别字或口语化表达的问题需要验证引用编号是否正确的问题4.2 检索指标与生成指标评测不能只看最终答案是否“像话”。RAG 系统可以拆成检索和生成两段每一段都有独立的指标。层级指标含义评估方式检索命中率正确答案是否在 top_k 结果中脚本统计检索上下文相关度检索出的片段是否与问题强相关人工抽样生成忠实度回答内容是否源自给定资料逐条标注生成引用正确率回答中引用的编号是否真实存在脚本校验成本平均 token 消耗一次问答消耗多少 token日志统计命中率是第一阶段最该看的指标。如果正确答案根本不在这轮检索结果里后续 prompt 写得再好模型也回答不出来。4.3 最小评测脚本下面用最简单的规则给 RAG 系统建立回归测试预设一批问题每个问题规定必须出现的关键词跑完整个链路后统计关键词命中率。def run_eval( questions: list[str], must_keywords: list[list[str]], chunks: list[str], chunk_embeddings: list[list[float]], ) - float: hit 0 total len(questions) for q, keywords in zip(questions, must_keywords): content, _ answer(q, chunks, chunk_embeddings) if all(k in content for k in keywords): hit 1 hit_rate hit / total print(fkeyword hit rate: {hit_rate:.2%}) return hit_rate这个脚本的局限很明显关键词命中不能完全代表语义正确但它能快速发现回归问题。如果一次改动让命中率从 90% 掉到 60%说明改动很有问题。更完善的评测需要引入大模型作为裁判或者人工标注。生产环境建议保留一个较小的“核心评测集”每次发布前跑一遍保证基本质量不退化。4.4 质量问题的定位与调整当评测结果不理想时先不要急着改 prompt而是先定位问题在哪一层。现象可能原因调整方向回答没有依据检索未命中调整分块、增加 top_k、优化 Embedding答案遗漏关键点上下文不完整减小分块、增加重叠、引入重排总回答不知道检索片段不相关或 prompt 限制过严先检查检索结果再放开 prompt引用编号错误输出后处理缺失生成后校验编号是否在上下文范围内最核心的经验是永远先看检索结果。把一次用户问题的 top_k 片段打印出来人工判断这些片段是不是真的能支撑答案。如果检索结果本身不对调 prompt 只是在错误基础上修修补补。5. 部署、成本与可观测性从能跑到能交付5.1 用 FastAPI 提供问答接口本地脚本能跑通之后下一步是把问答能力封装成 HTTP 接口。这里用 FastAPI 实现结构清晰且生态成熟。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class AskRequest(BaseModel): question: str class AskResponse(BaseModel): answer: str