RGA 系列写到第四篇。前面三篇分别聊了项目定位、整体架构和开发环境今天这篇直接进入正题把 API 地图画出来然后写第一个能跑起来的程序。所谓 API 地图说白了就是一张表——RGA 这台机器到底要消费哪些 API每个 API 用来干什么、走什么协议、用什么鉴权、大概花多少钱。别小看这一步我见过太多应用死在 API 管理混乱上密钥硬编码在代码里、不同供应商的 SDK 混在一处、报 401 了都不知道在查哪个服务。这篇我按自己的实操顺序来写包括跑通第一个程序的完整代码以及首轮实测遇到的高频报错和排查思路适合正在搭 AI 应用、特别是准备做检索增强RAG方向的朋友参考。1. 为什么先画 API 地图而不是直接写调用代码1.1 一张表解决这个服务是干嘛的的混乱RGARetrieval-Augmented Generation Assistant检索增强生成助手的核心循环其实不复杂用户提问 → 从知识库召回相关内容 → 把内容拼进上下文 → 交给大模型生成答案。但不复杂是就原理而言落到工程上每一个环节都要对接外部能力也就是一组 API。我见过不少朋友拿到类似需求直接开写今天看到 DeepSeek 便宜就用 DeepSeek明天觉得某个向量服务不错就切过去代码里 new 了四五个 client密钥散落在各个模块。等到要排查问题、算成本的时候才发现自己根本说不清系统到底依赖了几个外部服务。这不是代码能力问题是信息没有结构化。API 地图要解决的就是这个。它不需要多复杂至少记录这几列服务名称给这个 API 起一个内部代号比如chat_llm、embedding、parser。用途对话生成、向量化、文档解析、检索、业务数据等。端点地址base_url方便换供应商时全局排查。鉴权方式Bearer Token、签名、还是 PaaS 平台的 app_id/app_secret。模型与上下文长度例如 64k、128k、1M token这直接决定你后面怎么切片。价格口径按 token 计费还是按次计费每百万 token 多少钱。限流情况每分钟请求数限制并发上限。我把这张表放在项目的docs/api_map.md里每次新增或替换服务先改表再改代码。实测下来这个习惯让你在两周后回头改代码时不需要翻聊天记录去回忆那个 key 到底是哪家的。1.2 地图先行与边写边补的取舍有人会觉得做原型阶段画这么细是不是过度设计。我的取舍是第一版的地图可以只锁定两条硬依赖——大模型对话 API 和向量化 API其他全部推迟。原因是 RGA 的主循环只需要这两个就能转起来文档解析、网页搜索、业务数据 API 都是外围能力按需接就行。所以我的第一版 API 地图长这样对话层一个供应商、向量化一个供应商、检索先用本地内存实现、文档解析先手动喂文本。等第一版跑通再在地图上逐步补行。这个最小闭环的思路帮我避免了一上来就被各种工具细节拖住后面你会发现很多坑其实是等系统真正跑起来才暴露的提前接一堆服务只会让首轮排错无从下手。2. RGA 要接哪些 API一张全景表和三层拆解2.1 全景表先放我最终规划的全景表这是 RGA 完整形态下的 API 清单不是第一版就要全部接完层级用途候选服务鉴权方式备注对话生成回答用户问题、总结、改写DeepSeek、智谱 GLM、Kimi、讯飞星火Bearer TokenOpenAI 兼容格式第一版固定其中一家向量化把文本切成向量供语义召回BAAI/bge 系列硅基流动等平台托管、智谱 embeddingBearer Token中文场景优先 bge文档解析PDF/Word/PPT 转可索引文本MinerU、Unstructured、云厂商文档解析Token 或服务 URL 配置图片型 PDF 要带 OCR向量检索召回相似切片本地 FAISS轻量、服务化向量库本地调用或 Token第一版直接用内存业务数据行情、商品、店铺分析等外部数据东财股票数据、拼多多开放平台等各家签名/Token 不同按实际需求插件化接入2.2 对话层OpenAI 兼容格式成了事实标准现在国内主流的大模型厂商基本都提供了 OpenAI 兼容的 HTTP 接口这件事对开发者来说是个巨大的便利。意味着你不需要为每家写一套 SDK 调用逻辑只要改三个东西base_url、api_key、model名称。举例DeepSeek 的接口是https://api.deepseek.com模型名用deepseek-chat智谱是https://open.bigmodel.cn/api/paas/v4模型名用glm-4-air这类Kimi 是https://api.moonshot.cn/v1。讯飞星火早年是签名鉴权现在也提供了兼容格式但如果你用它的原生协议需要处理app_id、api_key、api_secret三个东西拼签名麻烦不少。这也是为什么我在 API 地图里把鉴权方式单独列出来——同是对话 API拿到手的东西可能完全不一样。第一版我选 DeepSeek 做主对话供应商核心原因是便宜、上下文给得大方、文档干净。但地图上我会把智谱和 Kimi 也列上因为不同任务的性价比差异以后一定会让你做切换。2.3 向量化与解析层检索增强的两个关键配角RGA 之所以叫检索增强关键就在这两层。对话 API 负责生成但生成得准不准取决于你喂给它的上下文也就是检索和解析的质量。向量化这块中文场景我优先推荐 BAAI 的 bge 系列模型。相比通用 embeddingbge 在中文语义匹配上更稳。你可以用托管平台提供的 bge 服务也可以本地起一个推理服务后者省 QPS 费用但多一份运维成本。第一版直接调 API 是最省事的只要拿到一个能返回向量数组的端点即可。文档解析层热词里频繁出现的 MinerU 和 Unstructured 都是这个角色。MinerU 在复杂 PDF多栏、表格、扫描件上表现不错Unstructured 胜在格式覆盖面广。这里要特别提醒很多接入 Unstructured 的项目都会踩到 dify unstructured api url is not configured for doc file processing 这类报错——本质是平台或插件不知道你的 Unstructured 服务跑在哪需要在配置里显式填一个可访问的 URL。这个坑后面单独展开。2.4 按需接入的业务数据 APIRGA 如果只做通用问答价值有限让它能查实时数据才有意思。比如东财的股票行情接口、拼多多开放平台的商品接口这类 API 的鉴权往往不是简单 Token而是签名或 OAuth和对话 API 完全是两套玩法。我的建议是不要把它们揉进主循环而是做成插件主程序只定义工具调用的接口具体实现各自维护。这也是后面演进篇的内容第一版先不碰。3. API Key 的正确打开方式环境变量与最小验证3.1 密钥绝不进代码环境变量加 .env热词里那些 401 unauthorized: incorrect api key provided 的报错十有七八和密钥管理有关。最常见的翻车姿势是把 key 直接写在脚本里然后整个仓库被推到公开平台几分钟后你的额度就开始燃烧。这种事真不是吓唬人我见过不止一次。正确做法密钥放环境变量本地开发用.env文件统一管理该文件必须进.gitignore。# .env DEEPSEEK_API_KEYsk-你的密钥 DEEPSEEK_BASE_URLhttps://api.deepseek.com EMBEDDING_API_KEYsk-你的向量服务密钥 EMBEDDING_BASE_URLhttps://api.siliconflow.cn/v1然后在项目入口加载# config.py import os from dotenv import load_dotenv load_dotenv() def get_env(name: str, required: bool True) - str: value os.getenv(name, ).strip() if required and not value: raise RuntimeError(f缺少环境变量: {name}) return value为什么非要包一层get_env因为直接os.getenv拿到的值可能带前后空格那个空格就是 401 的经典来源。.strip()能在源头解决。有人会遇到这样一个报错llm-deepseek: no api key for provider route deepseek-official。这通常不是 DeepSeek 的问题而是你用的网关/应用层比如某些 LLM 网关项目在启动时没有读到DEEPSEEK_API_KEY这个环境变量。排查顺序很固定先确认.env文件在不在当前工作目录再确认变量名是否完全一致DEEPSEEK_API_KEY和deepseek_api_key是两个东西最后确认应用是不是在启动阶段就加载了 dotenv。很多网关项目要求你在启动命令里显式传环境变量光靠.env文件不一定生效。3.2 写代码前先用 curl 验证密钥我强烈建议在写 Python 脚本之前先用一条 curl 把密钥和端点打通。这一步能省掉你后面 debug 时的大量自我怀疑。curl -s https://api.deepseek.com/chat/completions \ -H Content-Type: application/json \ -H Authorization: Bearer $DEEPSEEK_API_KEY \ -d {model: deepseek-chat, messages: [{role: user, content: ping}], max_tokens: 10}如果返回一个带id和choices的 JSON说明密钥和端点都正常。如果返回 401先检查复制 key 时有没有带上多余的空格或引号再确认你用的是不是这个供应商的 key。这里的坑在于很多平台的 key 都带sk-前缀你完全可能把 A 家的 key 填到 B 家的接口上报错同样是 401。另外注意控制台里显示的 key 可能是打码的比如sk-svcac****这种。打码显示是正常保护但你必须在创建时把完整 key 复制保存好之后很多平台不会再给你看第二次。真丢了就重新生成一个别拿打码的字符串去调试那只会无限 401。4. 第一个程序用一次检索增强生成跑通全链路4.1 目标与选型第一步先砍掉所有不必要的东西第一版程序的目标定得很小输入一个问题系统能从几段内置文档中召回相关内容拼进 prompt让大模型基于这些内容回答。不接文档解析、不用服务化向量库、不做流式输出、不做多轮记忆。为什么这么砍因为检索-增强-生成这条链路里每一步都可能出错你要的是一个可以逐个环节验证的最小闭环而不是一个失败时你根本不知道错在哪的庞然大物。向量检索我直接用了内存里的 numpy 算余弦相似度没有上 FAISS也没起 Docker 容器。这是故意的——第一版如果引入向量数据库就得处理 Docker 权限、端口映射、数据持久化一堆事这些和核心链路无关。等文档量上来再迁移到 FAISS 或服务化向量库代码改动也不过是替换一个函数。4.2 完整代码下面是rga_first_program.py的完整代码你可以直接抄走跑一遍# rga_first_program.py # 第一个程序一条 Query 走通检索 - 增强 - 生成全链路 import os import numpy as np from dotenv import load_dotenv from openai import OpenAI load_dotenv() # 1. 初始化两个客户端对话和向量化 chat_client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL, https://api.deepseek.com), ) embed_client OpenAI( api_keyos.getenv(EMBEDDING_API_KEY), base_urlos.getenv(EMBEDDING_BASE_URL), ) # 2. 内置文档第一版先不接 PDF用几段文本跑通链路 docs [ RGA 是一个检索增强生成助手核心流程是解析文档、切片、向量化、召回、交给大模型生成答案。, 向量检索比关键词检索更关注语义用户说怎么让昨天聊的东西不丢系统能匹配到记忆持久化相关的内容。, API Key 属于敏感信息必须放在环境变量里管理不能写进代码仓库也不能被版本控制工具提交。, ] # 3. 切片第一版按句号粗切避免长文本拖垮召回质量 chunks [] for doc_id, doc in enumerate(docs): for part in doc.split(。): text part.strip() if text: chunks.append({doc_id: doc_id, text: text 。}) # 4. 向量化 def embed(text: str): resp embed_client.embeddings.create( modelBAAI/bge-zh-v1.5, inputtext, ) return resp.data[0].embedding vectors [embed(c[text]) for c in chunks] # 5. 召回余弦相似度取 Top-K def cosine(a, b): a, b np.array(a), np.array(b) return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) 1e-9) def retrieve(query: str, top_k: int 2): qv embed(query) scored sorted( [(cosine(qv, v), i) for i, v in enumerate(vectors)], keylambda x: x[0], reverseTrue, ) return [chunks[i] for _, i in scored[:top_k]] # 6. 生成把召回结果拼进上下文 def ask(query: str) - str: hits retrieve(query) context \n.join(f[{h[doc_id]}] {h[text]} for h in hits) messages [ { role: system, content: 你是一个严谨的助手只依据参考资料回答问题参考资料里没有的信息明确说不知道不要编造。, }, { role: user, content: f参考资料\n{context}\n\n问题{query}, }, ] resp chat_client.chat.completions.create( modeldeepseek-chat, messagesmessages, temperature0.3, ) return resp.choices[0].message.content if __name__ __main__: print(ask(RGA 的核心流程是什么))4.3 运行与预期输出运行方式很简单pip install python-dotenv openai numpy python rga_first_program.py预期输出大致是根据参考资料RGA 的核心流程是解析文档、切片、向量化、召回、交给大模型生成答案。跑通后你可以做一个反向验证问一个参考资料里没有的问题比如RGA 支持图片识别吗如果系统老老实实说参考资料中没有提到说明 prompt 约束生效了如果它开始编说明 system prompt 写得还不够强硬需要加强。这一步很重要——检索增强系统的底线是没有依据就不回答宁可不答也别胡说。代码里有几个设计点值得说。切片按。粗切是刻意为之第一版最怕的是把整篇文档塞进一个 chunk导致召回时语义被稀释。temperature0.3是给问答场景定的太低会显得机械太高容易跑题0.3 到 0.5 是问答任务的常见区间。Top-K 选 2 是因为测试文档少等文档量上来再调。5. 首轮实测踩坑401、上下文超限与 Docker 权限的完整排查5.1 401 unauthorized从密钥到账户的逐层排查unexpected status 401 unauthorized: incorrect api key provided这类报错是 API 调试里出现频率最高的一条。我的排查套路固定如下第一确认密钥本身。复制时有没有带空格、引号、换行.env里值两侧有没有多余字符这些用print(repr(os.getenv(DEEPSEEK_API_KEY)))一眼就能看出来——repr会把隐藏字符暴露出来。第二确认密钥属于哪个供应商。sk-开头的 key 太多家都在用你把 DeepSeek 的 key 填到 OpenAI 兼容端点、或者填到某网关的 provider 配置里报错都是 401。对照 API 地图里的 base_url 逐项核对重点看 Authorization 头和请求的域名是不是同一家。第三确认账户状态。欠费、被限流、organization 被禁用都会以 401 或 403 的形式出现。热词里那条 this organization has been disabled 就是典型的账户层面问题——admin 已经停用组织或 token 失效普通开发者只能找组织管理员处理自建项目就检查自己的账单和 token 有效期。5.2 400 maximum context length上下文超限的应对api error: 400 this models maximum context length is 1048576 tokens这类报错说明你喂给模型的 prompt 超过了模型上下文上限。注意1M token 的模型也会超因为 RAG 场景里你可能会把大量检索结果直接拼进去几轮对话下来上下文滚雪球。解决思路是控制输入而不是提高限额。切片长度要控制比如每片 500 到 800 token并带少量重叠召回数量要限制Top-K 通常 3 到 8 就够了历史对话要截断只保留最近 N 轮。在代码里可以加一个硬保护生成 prompt 后先估算 token 数超过阈值就缩减召回数量或截断文档。估算可以用tiktoken偷懒一点就先按英文字符约 4 字符 1 token、中文约 1 到 2 个字符 1 token 来粗算反正只是做保护不是精确计费。5.3 Docker 权限问题一个会反复出现的环境刺客热词里那条permission denied while trying to connect to the docker api at unix:///var/run/docker.sock是典型的 Linux 环境问题。很多向量库、文档解析服务习惯用 Docker 启动但当前用户不在docker用户组里于是连不上 Docker 的 Unix socket。通常的解法sudo usermod -aG docker $USER然后退出重新登录让组权限生效。如果公司机器不方便这么搞也可以配置 rootless Docker或者干脆像第一版那样向量检索先用内存方案服务化容器等真正需要时再上。我的建议是做核心链路验证时尽量不要让 Docker 权限成为阻塞项先本地跑通再说。5.4 排查顺序总结把首轮实测的高频报错整理成表方便你对照报错现象优先排查修复动作401 incorrect api key密钥复制是否有隐藏字符是否填错供应商用repr()检查curl 直连验证no api key for provider route网关应用的环境变量是否加载确认.env在工作目录、变量名一致、启动时加载 dotenv400 maximum context lengthprompt 总 token 是否超限控制切片长度、Top-K、历史轮数加 token 保护400 organization disabled账户/组织状态检查账单、token 有效期联系管理员econnreset网络链路或服务端抖断增加超时与重试设置指数退避docker socket permission denied当前用户是否在 docker 组usermod -aG docker后重登或改用本地方案unstructured api url not configured平台配置里是否填了服务地址在 Dify/插件配置中填写可访问的 unstructured 服务 URL6. 从第一个程序到 RGA 主循环适配器与后续演进6.1 给每个 API 套一层适配器第一个程序跑通后我不建议马上加功能而是先做一次小重构把每个外部服务封装成接口。原因很现实——大模型供应商的价格和模型迭代太快你今天用的主力模型下个月可能就被新模型取代或者成本翻倍。如果调用逻辑散落在业务代码里每次切换都是一次伤筋动骨。最小化的适配器长这样class ChatProvider: def chat(self, messages: list[dict], temperature: float 0.3) - str: raise NotImplementedError class DeepSeekChat(ChatProvider): def __init__(self, api_key: str, base_url: str): self.client OpenAI(api_keyapi_key, base_urlbase_url) def chat(self, messages, temperature0.3): resp self.client.chat.completions.create( modeldeepseek-chat, messagesmessages, temperaturetemperature, ) return resp.choices[0].message.content换供应商时你只需要新增一个实现类并在配置里改一行。向量化、文档解析同理。这套东西花不了一个小时但会让后续每一步都轻松很多。6.2 后续可以长出来的东西第一个程序只是骨架RGA 真正成型还需要这些能力流式输出改善体验多轮对话加记忆文档加载做成异步队列检索环节加重排提高精度再加一个评测集定期验证回答质量。另外强烈建议把 API 地图升级成带观测数据的表格每次调用记录延迟和费用两周后你就能看出哪些调用值得缓存、哪些供应商该缩减用量。我自己的体会是API 地图不是画完就扔的静态文档它是项目活着的一部分。每次踩坑、每次换服务、每次调参都值得回填到那张表里。你会发现项目后期绝大多数诡异问题——突然变慢、费用异常、间歇性 401——都能在地图上找到线索。最后分享一个实操中的小习惯每个新接入的 API我都会先写一个最小调用脚本和业务代码完全隔离。跑通后这个脚本就是活文档也是以后排查问题的起点。第一个程序不用追求漂亮能稳定跑通再往上堆东西这条路我替你探过了稳。
