RGA 系列写到第四篇。前面三篇分别聊了项目定位、整体架构和开发环境，今天这篇直接进入正题：把 API 地图画出来，然后写第一个能跑起来的程序。所谓 API 地图，说白了就是一张表——RGA 这台机器到底要消费哪些 API，每个 API 用来干什么、走什么协议、用什么鉴权、大概花多少钱。别小看这一步，我见过太多应用死在 API 管理混乱上：密钥硬编码在代码里、不同供应商的 SDK 混在一处、报 401 了都不知道在查哪个服务。这篇我按自己的实操顺序来写，包括跑通第一个程序的完整代码，以及首轮实测遇到的高频报错和排查思路，适合正在搭 AI 应用、特别是准备做检索增强（RAG）方向的朋友参考。
1. 为什么先画 API 地图，而不是直接写调用代码
1.1 一张表解决"这个服务是干嘛的"的混乱
RGA（Retrieval-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 Token（OpenAI 兼容格式） | 第一版固定其中一家 |
| 向量化 | 把文本切成向量，供语义召回 | BAAI/bge 系列（硅基流动等平台托管）、智谱 embedding | Bearer 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 系列模型。相比通用 embedding，bge 在中文语义匹配上更稳。你可以用托管平台提供的 bge 服务，也可以本地起一个推理服务，后者省 QPS 费用但多一份运维成本。第一版直接调 API 是最省事的，只要拿到一个能返回向量数组的端点即可。
文档解析层，热词里频繁出现的 MinerU 和 Unstructured 都是这个角色。MinerU 在复杂 PDF（多栏、表格、扫描件）上表现不错，Unstructured 胜在格式覆盖面广。这里要特别提醒：很多接入 Unstructured 的项目都会踩到 "dify unstructured api url is not configured for doc file processing" 这类报错——本质是平台或插件不知道你的 Unstructured 服务跑在哪，需要在配置里显式填一个可访问的 URL。这个坑后面单独展开。
2.4 按需接入的业务数据 API
RGA 如果只做通用问答，价值有限；让它能查实时数据才有意思。比如东财的股票行情接口、拼多多开放平台的商品接口，这类 API 的鉴权往往不是简单 Token，而是签名或 OAuth，和对话 API 完全是两套玩法。我的建议是不要把它们揉进主循环，而是做成插件：主程序只定义"工具调用"的接口，具体实现各自维护。这也是后面演进篇的内容，第一版先不碰。
3. API Key 的正确打开方式：环境变量与最小验证
3.1 密钥绝不进代码：环境变量加 .env
热词里那些 "401 unauthorized: incorrect api key provided" 的报错，十有七八和密钥管理有关。最常见的翻车姿势是把 key 直接写在脚本里，然后整个仓库被推到公开平台，几分钟后你的额度就开始燃烧。这种事真不是吓唬人，我见过不止一次。
正确做法：密钥放环境变量，本地开发用.env文件统一管理，该文件必须进.gitignore。
# .env DEEPSEEK_API_KEY=sk-你的密钥 DEEPSEEK_BASE_URL=https://api.deepseek.com EMBEDDING_API_KEY=sk-你的向量服务密钥 EMBEDDING_BASE_URL=https://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_key=os.getenv("DEEPSEEK_API_KEY"), base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com"), ) embed_client = OpenAI( api_key=os.getenv("EMBEDDING_API_KEY"), base_url=os.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( model="BAAI/bge-zh-v1.5", input=text, ) 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)], key=lambda x: x[0], reverse=True, ) 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( model="deepseek-chat", messages=messages, temperature=0.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，导致召回时语义被稀释。temperature=0.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 model's 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在工作目录、变量名一致、启动时加载 dotenv |
| 400 maximum context length | prompt 总 token 是否超限 | 控制切片长度、Top-K、历史轮数；加 token 保护 |
| 400 organization disabled | 账户/组织状态 | 检查账单、token 有效期，联系管理员 |
| econnreset | 网络链路或服务端抖断 | 增加超时与重试，设置指数退避 |
| docker socket permission denied | 当前用户是否在 docker 组 | usermod -aG docker后重登，或改用本地方案 |
| unstructured api url not configured | 平台配置里是否填了服务地址 | 在 Dify/插件配置中填写可访问的 unstructured 服务 URL |
6. 从第一个程序到 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_key=api_key, base_url=base_url) def chat(self, messages, temperature=0.3): resp = self.client.chat.completions.create( model="deepseek-chat", messages=messages, temperature=temperature, ) return resp.choices[0].message.content换供应商时，你只需要新增一个实现类，并在配置里改一行。向量化、文档解析同理。这套东西花不了一个小时，但会让后续每一步都轻松很多。
6.2 后续可以长出来的东西
第一个程序只是骨架，RGA 真正成型还需要这些能力：流式输出改善体验，多轮对话加记忆，文档加载做成异步队列，检索环节加重排提高精度，再加一个评测集定期验证回答质量。另外强烈建议把 API 地图升级成"带观测数据"的表格，每次调用记录延迟和费用，两周后你就能看出哪些调用值得缓存、哪些供应商该缩减用量。
我自己的体会是：API 地图不是画完就扔的静态文档，它是项目活着的一部分。每次踩坑、每次换服务、每次调参，都值得回填到那张表里。你会发现，项目后期绝大多数"诡异问题"——突然变慢、费用异常、间歇性 401——都能在地图上找到线索。
最后分享一个实操中的小习惯：每个新接入的 API，我都会先写一个最小调用脚本，和业务代码完全隔离。跑通后这个脚本就是活文档，也是以后排查问题的起点。第一个程序不用追求漂亮，能稳定跑通再往上堆东西，这条路我替你探过了，稳。
