尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

Agent工程化实战:从RAG到知识库问答系统的完整架构解析

Agent工程化实战:从RAG到知识库问答系统的完整架构解析 最近行业里有一件事值得关注人工智能创业公司“自变量”被曝已向港交所递交上市申请。公开信息显示这家公司成立时间不长按报道口径大约两年半而股东名单中出现了字节跳动、阿里巴巴、美团等互联网大厂的名字。很多人把这当作融资新闻或资本话题来讨论但从技术角度看我更关注的是另一层内容一家以“大模型 Agent”为核心方向的公司从成立到规模化交付靠的到底是什么样的工程能力。本文不讨论上市合规、估值定价、二级市场预期这些内容只从技术视角做一次复盘式拆解。我们会围绕 Agent 系统的基本概念、RAG 检索增强、模型统一调用层、工具调用、上下文管理、评测与安全边界展开并用一个尽量简单、可以本地运行的知识库问答 Agent 项目演示这类系统的核心骨架。如果你正在做大模型应用开发、Agent 架构设计或者准备从 Demo 走向生产环境这篇文章应该会给你一套相对完整的参考。1. 背景一家 AI Agent 创业公司的上市递表与技术信号1.1 为什么技术人应该关注这条消息“自变量”这个名字本身很有程序员的趣味。在数学里自变量是输入因变量是输出在机器学习里输入特征和预测结果之间的关系也天然带着“自变量到因变量”的影子。放到大模型应用语境下我们可以理解为模型接收用户的指令和外部上下文经过推理与工具调用最后输出一个可验证的结果。这个“输入—推理—行动—输出”的闭环就是 Agent 系统的核心。从公开报道来看自变量主打的方向和 Agent、AI 应用相关。放在 2025 年之后的大模型行业环境里这不是孤例。上一阶段大家比拼的是基座模型参数和基础能力而到了应用层竞争阶段谁能把模型能力封装成客户真正用得起来的产品谁就有希望跑出商业闭环。字节跳动、阿里巴巴、美团这些大厂坐镇股东席从战略上各有自己的算力、云生态、本地生活或 ToB 场景诉求但从技术信号来看至少说明一件事资本和产业都已经把“Agent 能否规模化落地”当成一个关键命题。1.2 大模型应用层为什么开始成为焦点大模型技术栈大致可以分为三层基础模型层、模型服务与工具链层、应用与 Agent 层。基础模型层的特点是资金密集、算力密集玩家数量相对有限模型服务与工具链层解决的是“让人们更方便地调用模型”的问题而应用与 Agent 层则直接对接业务解决“模型到底能帮企业做什么”的问题。Agent 刚好处于应用层的核心位置。一个真正可用的 Agent 不只是“套着聊天框的模型接口”它需要感知上下文、拆解任务、调用搜索或数据库等外部工具、管理多轮记忆并且在犯错时具备一定的恢复能力。这些能力叠在一起才让 Agent 从“看起来聪明”变成“在具体岗位上有用”。这也是为什么我认为无论外面的资本消息如何变化技术人真正要补的课是 Agent 工程化。1.3 从 Demo 到交付工程化成了护城河很多大模型项目在演示阶段效果很好一旦进入生产就暴露出一堆问题接口超时、上下文长度不够、回答幻觉、工具调用失败、数据权限不安全、成本过高。这些问题没有一个能靠换一个更大的模型解决必须靠系统架构和工程规范来解决。因此这篇文章把重点放在一套可落地的最小架构上。我们不会追求面面俱到而是先把一条主线打通模型调用层、向量检索层、Agent 编排层、记忆层再加上评测和排查方法。这个闭环跑通之后你再去理解市面上各种 Agent 框架思路会清晰很多。2. Agent 相关核心概念2.1 什么是 AgentAgent 当前并没有一个严格统一的定义但业界比较认可的描述是Agent 是大模型驱动的智能体它能够在一定的目标约束下通过“感知—规划—行动—观察”的循环来完成任务。拆开来看感知获取用户输入、外部数据、API 返回结果。规划把复杂任务拆解成子步骤。行动调用工具、查询数据库、请求外部系统。观察读取工具调用结果判断当前状态是否达成目标。和普通的大模型 API 调用相比Agent 最大的区别是引入了循环和工具。普通 API 调用是“问一句答一句”Agent 可以根据中间结果继续往下走直到任务完成或达到上限。常见误区是把 Agent 和聊天机器人画等号。聊天机器人主要做文本对话Agent 则要“动手做事情”比如查询订单、生成报表、修改工单状态、自动修复代码这些都需要工具调用能力。2.2 RAG 是 Agent 最常见的知识来源RAG 的全称是 Retrieval-Augmented Generation也就是检索增强生成。它的思路很直接在让模型回答之前先从外部知识库中检索出与问题相关的资料片段并把这些片段拼进提示词让模型基于这些资料生成答案。为什么要引入 RAG主要有三个原因减少幻觉模型不会凭空编造太多细节因为回答所需的依据已经放在上下文里。知识更新及时四两拨千斤地解决“模型知识停留在训练截止日期”的问题。可追溯可以告诉用户答案来自哪份文档这在企业场景中是基本要求。RAG 不是 Agent但 Agent 系统里通常会集成 RAG。我们可以把 RAG 理解为 Agent 的“记忆外挂”或“知识工具”。2.3 Workflow 与 Agent 的区别工作流Workflow是写死的流程先走步骤 A再走步骤 B如果判断条件为真则执行 C。它稳定、可控、结果可预期适合业务流程固定的场景。Agent 则更有自主性模型自己决定下一步调用哪个工具、是否需要继续检索、是否已经完成任务。它灵活但稳定性和可控性都更差。实际项目中最合理的做法不是二选一而是混合使用。简单且固定的流程写成 Workflow复杂且需要决策的部分交给 Agent。比如一个客服系统基本接待流程可以固定但遇到投诉升级时交给 Agent 去分析用户意图、查询工单历史、生成处理建议。2.4 大模型应用系统的基本组成一个生产级的大模型应用通常会有这些模块模型接入层负责和不同大模型 API 通信做 Key 管理、超时重试、限流、日志。知识库层负责文档切分、向量化、存储、检索、重排。编排层负责决定调用哪个工具、如何拼接提示词、如何处理多轮状态。工具层封装外部系统 API比如订单查询、数据库操作、消息发送。记忆层保存短期会话信息和长期用户画像。评测与监控层用测试集评估回答质量用日志和链路追踪定位线上问题。安全与权限层控制谁可以访问哪些文档、哪些工具能被调用。这篇文章用最小案例覆盖前五层然后在后面的最佳实践里讨论评测、监控和安全。3. 环境准备与项目结构3.1 技术选型与版本说明本文示例以 Python 3.10 为主依赖尽量精简。核心组件如下FastAPI提供 Web 接口方便通过 HTTP 调用 Agent。Uvicorn运行 FastAPI 的服务器。openai 官方 Python SDK一般大模型服务都提供兼容接口可以统一调用。numpy实现简单的向量余弦相似度计算。python-dotenv读取本地环境变量。Redis 略作了解即可本文不强制依赖用于生产环境的多轮会话存储不是必须。版本需要根据你的项目实际情况调整本文重点演示配置思路不把版本号写死。建议在安装前确认 openai SDK 与你使用的大模型网关 API 版本兼容。安装依赖pip install fastapi uvicorn openai numpy python-dotenv如果你使用的是国内云厂商或开源模型网关通常也会提供 OpenAI 兼容的/v1/chat/completions与/v1/embeddings接口这种情况下 openai SDK 可以直接配置base_url使用。3.2 项目目录结构我们用一个极简但层次清晰的目录来组织代码agent-demo/ ├── .env.example ├── requirements.txt ├── app/ │ ├── __init__.py │ ├── llm.py │ ├── vector_store.py │ └── main.pyllm.py负责封装大模型和 Embedding 的调用。vector_store.py负责文档写入和向量检索。main.py负责提供/ingest和/ask两个 HTTP 接口。.env.example保存环境变量示例。这样的结构虽然简单但已经体现出分层思想模型调用不直接散落在业务代码里检索逻辑独立成模块外部表现用 HTTP 接口暴露。3.3 环境变量设计环境变量我们放在.env文件中MODEL_BASE_URLhttps://your-api-endpoint.example.com/v1 MODEL_API_KEYsk-your-api-key CHAT_MODEL_NAMEyour-chat-model EMBED_MODEL_NAMEyour-embed-model注意MODEL_BASE_URL后面是否需要/v1取决于你的服务商要求。建议先在自己的测试环境里用一条最简单的请求确认接口路径再填到这里。4. 核心模块拆解与最小实现4.1 LLM 统一调用层为什么需要单独封装一个模型调用层因为在真实项目中你的 Agent 可能要对接不同服务商的大模型也可能是同一个服务商的多个模型比如区分“快速便宜的轻量模型”和“回答质量更高的主力模型”。如果每个业务模块都直接调用 SDK将来切换模型时会非常痛苦。一个最小封装如下# 文件路径app/llm.py import os from openai import OpenAI class LLMClient: def __init__(self): # 这里用环境变量避免把密钥写死在代码里 api_key os.getenv(MODEL_API_KEY, EMPTY) base_url os.getenv(MODEL_BASE_URL, http://localhost:8000/v1) self.client OpenAI(api_keyapi_key, base_urlbase_url) self.chat_model os.getenv(CHAT_MODEL_NAME, your-chat-model) self.embed_model os.getenv(EMBED_MODEL_NAME, your-embed-model) def chat(self, messages, temperature0.3, max_tokens1024): resp self.client.chat.completions.create( modelself.chat_model, messagesmessages, temperaturetemperature, max_tokensmax_tokens, ) return resp.choices[0].message.content def embeddings(self, texts): resp self.client.embeddings.create(modelself.embed_model, inputtexts) return [item.embedding for item in resp.data]这里需要注意几个细节api_key可以由网关层校验也可以是一个不存在的占位符比如用 vLLM 部署本地模型时很多时候要求 key 写EMPTY。temperature是采样温度。知识库问答场景建议用偏低的 0.2 到 0.4减少随机性创意写作场景才考虑调高。max_tokens要结合模型上下文长度设置避免输出超过模型限制。在更大的项目里模型层还应该加上超时、重试、限流和日志。最小示例中我们只保留核心调用逻辑工程化增强会在最佳实践部分说明。4.2 向量检索层向量检索的核心思路是把文本转换成向量然后用向量之间的距离或相似度来衡量文本之间的语义相关度。生产环境里我们通常使用 Milvus、pgvector、Elasticsearch、Redis 向量检索等专用存储。为了方便你直接跑通示例下面给出一个用 numpy 实现的简易向量检索模块。它适合理解原理也适合中小数据量的原型验证。# 文件路径app/vector_store.py import numpy as np class SimpleVectorStore: def __init__(self, embed_fn): # embed_fn 是一个接收字符串列表并返回向量列表的函数 self.embed_fn embed_fn self.documents [] self.metadata [] self.vectors [] def add_document(self, text, metadataNone): self.documents.append(text) self.metadata.append(metadata or {}) vec np.asarray(self.embed_fn([text])[0], dtypenp.float32) self.vectors.append(vec) def search(self, query, top_k3): qvec np.asarray(self.embed_fn([query])[0], dtypenp.float32) scores [] for vec in self.vectors: cos_sim float(np.dot(qvec, vec) / ( np.linalg.norm(qvec) * np.linalg.norm(vec) 1e-12 )) scores.append(cos_sim) ranked sorted( zip(self.documents, self.metadata, scores), keylambda x: x[2], reverseTrue ) return ranked[:top_k]这个模块的原理是每个文档入库时都调用一次 embedding 接口得到向量后存起来查询时把用户问题也转换成向量然后逐个计算余弦相似度最后返回相似度最高的 top_k 条结果。cos_sim的结果范围是 -1 到 1越接近 1 表示语义越接近。当然不同 Embedding 模型的空间分布不同直接对比绝对分数没有意义我们只需要用它做排序。这个简易方案不适合生产环境因为数据量增大后线性扫描所有向量的成本会越来越高。生产环境建议替换为专门的向量数据库但“文本转向量、相似度排序、取 top_k”这个逻辑是不变的。4.3 Agent 编排与工具调用Agent 编排层解决的核心问题是模型怎么决定调用工具怎么读懂工具返回结果以及怎么判断任务是否完成。目前接入工具调用最主流的方式是 function calling。OpenAI 兼容接口的流程是调用模型时传入工具定义模型在需要时返回一个 tool_calls 结构里面包含工具名称和参数然后我们在代码里执行真正的工具再把结果以 tool 角色消息返回给模型继续推理。一个简单的工具定义如下tools [ { type: function, function: { name: get_order_status, description: 根据订单号查询订单状态, parameters: { type: object, properties: { order_id: {type: string, description: 订单编号} }, required: [order_id] } } } ]调用链路的伪代码思路如下resp client.chat.completions.create( modelchat_model, messagesmessages, toolstools, ) message resp.choices[0].message if message.tool_calls: for tool_call in message.tool_calls: tool_name tool_call.function.name arguments tool_call.function.arguments # 在这里执行真正的工具函数注意不要直接 eval result dispatch_tool(tool_name, arguments) # 把结果追加到对话消息中让模型基于结果继续生成最终回答 messages.append(message) messages.append({ role: tool, tool_call_id: tool_call.id, content: str(result) })判断一次 Agent 循环是否结束常见条件有三个模型没有返回新的 tool_calls循环次数达到上限工具返回了明确的终止信号。为了防止模型陷入死循环生产环境必须设置最大轮数比如 5 次或 10 次。这里特别强调一点不要直接用 eval 去执行模型传来的字符串代码这是一个严重的安全风险。工具参数应该做严格的 JSON 解析、类型校验和权限校验。如果模型返回的工具参数不符合定义应该让模型重新生成而不是强行执行。4.4 记忆与上下文管理Agent 在真实业务中通常需要多轮对话这就必须有记忆模块。记忆分为短期记忆和长期记忆。短期记忆一般存储在 Redis 或内存中以会话 ID 为 key保存最近的若干轮对话消息。由于大模型上下文长度有限我们不能无限地把历史对话塞进去通常只保留最近 N 轮或者对更早的对话做摘要压缩。长期记忆常见做法是把用户的关键信息、偏好、历史工单等写入用户画像库需要时通过检索召回相关记忆再注入提示词。在最小示例里我们用单次请求演示核心流程不专门引入 Redis。但你可以这样设计接口增加session_id参数后端用session_id作为 key将历史消息列表存入 Redis每次查询时取回最近的 10 条消息拼进 messages 数组。5. 完整实战搭建可运行的知识库问答 Agent5.1 需求定义我们做一个最小可运行的知识库问答 Agent目标是用户通过 HTTP 接口上传一段文档。文档被向量化并存储到简易向量检索库中。用户提问时系统从知识库中检索最相关的资料片段。大模型基于这些资料片段生成回答并返回命中的参考来源。这个案例虽然简单但它已经把 RAG 的主链路完整跑通了。5.2 实现代码首先创建依赖文件。# requirements.txt fastapi uvicorn openai numpy python-dotenv然后是入口文件# 文件路径app/main.py from fastapi import FastAPI from pydantic import BaseModel from dotenv import load_dotenv from llm import LLMClient from vector_store import SimpleVectorStore load_dotenv() app FastAPI() llm LLMClient() store SimpleVectorStore(llm.embeddings) class IngestRequest(BaseModel): text: str metadata: dict {} class AskRequest(BaseModel): question: str app.post(/ingest) def ingest(req: IngestRequest): store.add_document(req.text, req.metadata) return {message: ok, total_docs: len(store.documents)} app.post(/ask) def ask(req: AskRequest): hits store.search(req.question, top_k3) context \n\n.join([doc for doc, _, _ in hits]) messages [ { role: system, content: 你是一个企业知识库助手。请根据参考资料回答问题。 如果参考资料不足以回答请明确说明不知道不要编造。, }, { role: user, content: f参考资料\n{context}\n\n问题{req.question}, }, ] answer llm.chat(messages, temperature0.2) return { answer: answer, references: [ {text: doc, score: score, metadata: meta} for doc, meta, score in hits ], }代码逻辑并不复杂但有几个地方值得展开解释。load_dotenv()会把.env文件中的配置加载到系统环境变量这样LLMClient就能拿到正确的MODEL_BASE_URL、MODEL_API_KEY等配置。IngestRequest和AskRequest是 Pydantic 请求模型FastAPI 会基于它们自动校验请求参数。如果你不传metadata会使用默认空字典。在/ask接口中references字段非常有价值。它一方面方便用户判断答案的出处另一方面也让开发者在调试时能看到检索阶段命中了哪些文档定位“答非所问”时很快就能知道是检索问题还是生成问题。5.3 启动与验证在项目根目录下先创建一个.env文件内容参考.env.exampleMODEL_BASE_URLhttps://your-api-endpoint.example.com/v1 MODEL_API_KEYsk-your-api-key CHAT_MODEL_NAMEyour-chat-model EMBED_MODEL_NAMEyour-embed-model启动服务uvicorn app.main:app --host 0.0.0.0 --port 8000然后我们可以先用 curl 写入一条文档curl -X POST http://127.0.0.1:8000/ingest \ -H Content-Type: application/json \ -d {text: 自变量是一家人工智能创业公司聚焦大模型 Agent 开发目标是帮助企业完成复杂的知识管理与流程自动化任务。}预期响应{message: ok, total_docs: 1}接着提问curl -X POST http://127.0.0.1:8000/ask \ -H Content-Type: application/json \ -d {question: 自变量这家公司是做什么的}预期响应中应该包含根据资料生成的答案同时 references 里带回了我们写入的那条文档文本。如果返回结果为空或者提示没有知识库内容优先检查是否成功调用了/ingest接口embedding 接口是否可用CHAT_MODEL_NAME和EMBED_MODEL_NAME是否填写正确。5.4 如何扩展成生产级方案上面的最小示例可以直接跑通但离生产环境还有一段距离。扩展方向主要有几个。第一把SimpleVectorStore替换成专业向量数据库。数据量达到十万条甚至百万条文档时内存线性扫描完全不可行。建议使用 Milvus、pgvector、Elasticsearch、OpenSearch 这类具备索引能力的存储并把检索条件加上租户 ID 或权限标签过滤。第二引入重排Rerank模型。向量检索先召回 top 50再用一个更精确的排序模型选出 top 5。重排能明显提升回答准确性代价是会多一次模型调用和少量延迟。第三把纯提示词式 RAG 升级为具备工具调用能力的完整 Agent。比如用户问“帮我查一下上周的生产数据并生成摘要”系统需要先通过 API 访问数据库再基于数据库结果执行分析最后生成摘要。这就是从 RAG 到真正 Agent 的进阶路径。6. 常见问题与排查思路6.1 高频问题列表问题现象常见原因解决思路接口返回 401 或 403API Key 错误、网关鉴权失败检查环境变量中的 Key 和 base_url用小脚本单独验证接口返回 404请求路径错误或模型名不存在确认网关地址是否包含/v1确认模型名准确写入文档后检索不到embedding 接口报错或分块不一致加日志检查 embedding 返回确认写入和查询使用同一个模型回答完全没用到资料上下文注入失败或检索为空检查 messages 中的参考资料字段分析 references 是否为空Agent 陷入死循环工具调用没有终止条件设置最大轮数检查工具返回结果格式是否被模型理解回答内容模糊或编造检索片段信息不足或 top_k 太小增加 top_k、缩小分块粒度、引入重排运行一段时间后变慢向量检索线性扫描、上下文过长更换索引型向量库、做会话摘要压缩6.2 一个真实排查案例之前我遇到过一种很典型的情况文档已经写入成功但是每次提问模型回答都像在“自由发挥”完全没有参考文档内容。排查时先看/ask返回的 references 字段发现 references 为空说明问题出在检索阶段根本轮不到生成阶段。继续排查发现SimpleVectorStore内部在调用 embedding 时抛了异常但异常被上层某些逻辑吞掉了导致调用方以为检索成功、只是没有结果。解决方案是补全异常处理不吞异常至少要把 embedding 失败、向量检索失败和模型生成失败区分开来分别记录日志。这个案例给我们的经验是RAG 系统排错要先分段定位。把“文档导入”“检索召回”“提示词拼装”“模型生成”四段拆开分别验证就能快速缩小问题范围。7. 最佳实践与工程建议7.1 架构上的建议模型调用建议统一收敛到一个网关层。无论是用自研封装还是接入开源网关组件都要把 Key 管理、超时重试、限流、模型路由、日志审计集中在同一处。这样当上游模型服务不稳定时我们可以在网关层做降级切换而不是去改每个业务模块。工具调用一定要做权限控制。Agent 能够调用的系统越重要风险越高。数据库工具只能执行白名单 SQL消息工具只能推送给授权用户外部 API 请求必须经过参数校验。最小权限原则在这里不是一句空话给每个 Agent 分配一个专用的服务账号只开放它完成本职工作所需的最小权限。7.2 成本与性能优化大模型应用的成本大头通常来自 Token 消耗。常见优化策略包括对高频问题做缓存相同问题的回答直接命中缓存。对 Embedding 结果做缓存避免重复请求 embedding 接口。把长文档提前切片检索时只把命中的 top_k 片段放进上下文而不是全文塞入。简单问题用轻量小模型复杂问题才路由到参数更大的模型。延迟方面除了模型本身推理速度影响最大的是上下文长度和工具调用轮数。尽量控制单轮输入的 Token 数量减少不必要的多轮工具调用能显著改善用户体验。7.3 安全与合规边界RAG 系统很容易出现一个安全隐患用户通过恶意构造问题诱导模型输出知识库中无权限的敏感内容。因此文档入库时就要做权限标注检索阶段必须带上用户权限过滤而不是检索完再筛选。另外提示词注入也需要防。外部输入可能包含“忽略之前指令”之类的文本。如果我们把外部文档或搜索结果原样拼进提示词模型可能被带偏。缓解手段包括在系统提示词中强调上下文只能作为参考资料、对输入做敏感词过滤、将工具参数与提示词内容分开处理。7.4 可观测性与上线流程Agent 和普通接口最大的区别是状态多、链路长。一次回答可能经历多次模型调用、工具调用和检索过程。因此日志里至少要记录用户问题、检索命中的文档 ID、工具调用名称与参数、模型生成的回答、耗时和 Token 消耗。有条件的话引入链路追踪工具把一次 Agent 请求的所有步骤串起来。上线流程建议采用灰度发布先让 Agent 处理 5% 的流量人工抽看回答质量评估准确率和工单解决率再逐步放大。一旦出现回答质量下降或工具调用异常可以快速回滚到上一版本。8. 总结与学习路线这篇文章从“自变量赴港递表”这条新闻切入梳理了 Agent 类大模型应用公司背后的工程主线。我们重点完成了三件事第一理解 Agent、RAG、Workflow 这些概念之间的关系第二用一个最小项目跑通了模型调用、向量检索、知识库问答的完整链路第三整理了生产环境中最常见的排查思路和最佳实践。如果你接下来想继续深入我建议的学习顺序是先把本文代码跑通再尝试换成真正的向量数据库然后给系统加上 Redis 会话记忆再研究 function calling 和工具调用最后搭一套简单的评测集。评测集可以先用 50 条自己业务中的真实问题每次改动提示词、分块策略或模型后都拿这 50 条问题做回归对比。没有评测机制你很难判断改动是变好还是变坏。在实际项目中优先级可以这样排先保证数据权限和安全边界再建立回归评测然后做检索质量优化最后才去追求复杂的多 Agent 编排。把基础链路做稳Agent 才能真正从演示变成生产力。
返回列表