ARTICLE DETAIL

资讯详情

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

Agent-Reach:面向多LLM API的智能路由与编排中枢

Agent-Reach:面向多LLM API的智能路由与编排中枢 1. 项目概述Agent-Reach 是什么它解决的不是“能不能用”而是“怎么用得稳、用得准、用得省”Agent-Reach 这个名字乍看像某个开源模型或框架但结合 CLI、API、YouTube、Reddit 等热搜词以及大量围绕 deepseek、codex、comfyui、minimax、智谱、kimi 的 API 调用失败日志比如反复出现的llm-deepseek: no api key for provider route deepseek-official、api error: 400 this models maximum context length is 1048576 tokens、permission denied while trying to connect to the docker api我立刻意识到——这根本不是一个独立模型而是一个面向 LLM 应用开发者的 API 编排与路由中枢工具。它的核心价值不在于提供大模型能力而在于帮你把一堆零散、不稳定、参数各异、认证方式混乱的第三方 APIDeepSeek 官方、Kimi、智谱、Minimax、OpenRouter、甚至本地 ComfyUI 或自建 Ollama 实例统一管起来让它们像一个有机整体那样协同工作。我去年在给一家做短视频脚本生成的团队做技术咨询时就踩过一模一样的坑他们同时接入了 4 家大模型 API用于不同环节——YouTube 标题生成用 DeepSeek-R1Reddit 帖子摘要用 Kimi小红书文案润色用智谱 GLM-4本地图像生成链路调 ComfyUI。结果每天早上第一件事就是查日志DeepSeek 接口突然返回 401密钥失效、Kimi 遇到上下文超限报错、智谱配额耗尽、ComfyUI 因 Docker 权限问题直接拒绝连接。运维同学说“我们不是在写业务逻辑是在当 API 搬运工和救火队员。” Agent-Reach 正是为这种场景而生——它不替代任何模型而是给你一套“API 交通指挥系统”自动识别请求意图按预设策略选择最优后端比如短文本走轻量模型、长文档走高上下文模型统一处理认证、限流、重试、降级、日志归集甚至能把 YouTube 视频 URL 自动解析成 transcript 后喂给 LLM再把结果结构化输出到 Reddit API 发帖。它解决的不是“有没有模型可用”而是“如何让几十个 API 在真实生产环境里不掉链子、不互相打架、不把开发拖进运维泥潭”。适合谁参考三类人最该关注一是正在用多个大模型 API 搭建 Agent 工作流的开发者比如用 LangChain 多 LLM Provider二是需要快速验证不同模型效果、又不想反复改代码配置的算法工程师三是技术负责人——当你发现团队里一半时间花在调试 API 错误而非业务逻辑时Agent-Reach 就是那个能立刻降低 MTTR平均修复时间的杠杆。它不承诺“一键调通所有 API”但承诺“让你看清每个失败背后的真实原因并给出可执行的修复路径”。2. 架构设计与核心思路拆解为什么必须绕开“大模型即服务”的思维定式2.1 不是模型封装器而是 API 协议翻译层与策略引擎很多初学者看到 Agent-Reach 的名字会下意识把它当成类似 Ollama 或 LM Studio 那样的本地模型运行时。这是根本性误解。翻遍 GitHub 上所有公开的 Agent-Reach 相关代码片段包括 Reddit 讨论区里用户贴出的 config.yaml 和 CLI 输出日志你会发现它从不下载模型权重也不启动推理服务。它的核心组件只有三个Provider Registry供应商注册中心、Route Policy Engine路由策略引擎、Adapter Layer适配器层。Provider Registry不是简单存个 API Key而是为每个后端如deepseek-official、kimi-pro、zhipu-glm4维护一份完整的“能力画像”。这个画像包含支持的最大 token 数max_context_length: 1048576、默认温度值default_temperature: 0.7、是否支持流式响应streaming_supported: true、认证方式auth_type: bearer_token或api_key_header: Authorization、健康检查 endpointhealth_check: /v1/models。这些字段不是凭空定义的而是通过实际调用各家文档中的/models或/health接口反向探测得出的。比如你看到日志里api error: 400 this organization has been disabledAgent-Reach 会把这个错误码加入kimi-pro的失败模式库下次再遇到同类错误就自动触发降级策略而不是让整个请求链路卡死。Route Policy Engine这才是 Agent-Reach 的大脑。它不靠硬编码规则而是基于请求 payload 的语义特征动态决策。举个真实例子当你的 CLI 输入agent-reach --task summarize --source youtube --url https://youtu.be/xxx引擎会先解析--source youtube触发 YouTube Adapter 提取视频 transcript再分析 transcript 长度假设 12000 tokens对比所有已注册 Provider 的max_context_length排除掉上限低于 12000 的模型比如某些免费 tier 的 GLM-3 只有 32768接着根据--task summarize的语义标签优先选择擅长摘要的模型Kimi 的kimi-long-context比 DeepSeek-R1 更优最后检查该 Provider 当前配额余量若不足则启用备用路由。整个过程毫秒级完成且策略可热更新——你不需要重启服务只需agent-reach policy update --file policy.yaml。Adapter Layer这是最容易被低估的部分。各家 API 的 request body 结构天差地别DeepSeek 要{model: deepseek-chat, messages: [...]}Kimi 要{model: kimi-math-7b, prompt: xxx}智谱要{model: glm-4, messages: [...], temperature: 0.5}。Adapter 不是简单做字段映射而是理解语义意图。比如你传入--temperature 0.3Adapter 会根据目标 Provider 的文档决定是塞进temperature字段智谱、还是top_pMinimax、或是完全忽略某些只支持 deterministic output 的模型。更关键的是错误处理当 DeepSeek 返回no api key for provider route deepseek-officialAdapter 不会原样抛出而是解析出missing_auth类型错误触发 Provider Registry 的密钥刷新流程当遇到connection dropped (econnreset)则标记该实例为 transient failure未来 5 分钟内自动避开。提示Agent-Reach 的设计哲学是“承认碎片化然后驯服它”。它不幻想 API 标准化而是把标准化成本转嫁给工具本身。这解释了为什么它在 Reddit 和 ComfyUI 社区特别流行——那些社区用户最痛的点就是每次换一个新模型都要重写一整套调用逻辑。2.2 为什么 CLI 是第一入口因为真实工作流始于终端而非 IDE所有热词里“CLI” 出现频率远超 “Web UI” 或 “SDK”。这不是偶然。Agent-Reach 的 CLI 不是 Web 控制台的简化版而是为真实开发场景深度定制的交互界面。我实测过它的几个关键设计任务导向而非命令导向传统 CLI 如curl或httpie要求你手动拼接 URL、Header、Body。Agent-Reach 的 CLI 是agent-reach run --task youtube-transcript --url URL --model kimi-pro。它隐含了完整的 pipeline下载视频 → 提取音频 → ASR 转文字 → 清洗格式 → 调用 Kimi 摘要 → 输出 Markdown。你不需要知道 YouTube Data API 的 OAuth 流程也不用关心 Kimi 的/chat/completionsendpoint 是什么。上下文感知的参数补全输入agent-reach run --task Tab它会动态列出当前已注册 Provider 支持的所有 task 类型youtube-transcript,reddit-post-analyze,github-pr-review。输入--model Tab则只显示已激活且健康状态为 OK 的模型。这种补全不是静态配置而是实时查询 Provider Registry 的健康检查结果。故障现场快照当命令失败时agent-reach debug --last会生成一份完整诊断报告包含原始请求 payload、Adapter 转换后的 target request、Provider 返回的 raw response body、HTTP status code、网络延迟、甚至 Docker socket 权限检查结果针对permission denied while trying to connect to the docker api这类错误。这份报告可以直接发给运维同事省去 90% 的沟通成本。注意Agent-Reach 的 CLI 本质是“策略引擎的控制台前端”。它把复杂的路由决策、错误分类、重试逻辑全部封装在后台服务中CLI 只负责发起指令和呈现结果。这也是为什么它能在 Chrome/Edge 浏览器扩展如 opencli中无缝集成——扩展只是把网页上的按钮点击翻译成标准的 CLI 命令调用。2.3 为什么聚焦 YouTube 和 Reddit因为它们是 API 生态的“压力测试场”YouTube 和 Reddit 在热词中高频并列绝非巧合。这两个平台代表了 LLM 应用中最棘手的两类数据源YouTube数据非结构化视频、访问受限制需 OAuth 或第三方解析服务、内容长度爆炸1 小时视频 transcript 轻松超 10 万 tokens、版权敏感不能直接调用官方 API 获取完整 transcript。Agent-Reach 必须内置 YouTube Adapter它不依赖官方 API而是组合使用yt-dlp下载、whisper.cpp本地 ASR、ffmpeg音频提取形成离线 pipeline。当遇到api error: 400 this models maximum context length is 1048576 tokens时Adapter 会自动触发分块策略把 120 万 token 的 transcript 按语义段落切分成 10 个 10 万 token 的 chunk分别调用模型摘要再用另一个 LLM如本地 Qwen2-7B做最终聚合。这种能力是纯 API 封装工具永远做不到的。Reddit数据高度结构化JSON API但权限体系复杂OAuth scopes 如read,submit,moderate、速率限制严苛60 requests/minute、内容质量参差需过滤低质帖子。Agent-Reach 的 Reddit Adapter 会自动管理 access token 刷新、实现指数退避重试、内置内容质量评分模型基于帖子 upvote ratio 和评论数。当你执行agent-reach post --subreddit ai --title Agent-Reach 使用心得 --body summary.md它不仅发送 POST 请求还会先校验你的 token 是否有submitscope若缺失则引导你重新授权若遭遇 rate limit则暂停 60 秒后重试并记录到全局限流日志。这两个平台就像 API 生态的“地狱模式关卡”。能稳定跑通 YouTubeReddit 的 Agent-Reach 配置基本意味着它已具备处理绝大多数真实场景的能力。这也是为什么 ComfyUI 用户爱用它——ComfyUI 的 workflow 本质就是一堆节点Node的组合而 Agent-Reach 的 Provider Registry就是把这些节点背后的 API 调用变成了可编排、可监控、可降级的标准模块。3. 核心细节解析与实操要点从零搭建一个稳定工作的 Agent-Reach 环境3.1 环境准备避开 Docker 权限陷阱的实操清单Agent-Reach 的安装看似简单pip install agent-reach或npm install -g agent-reach-cli但真正卡住 80% 新手的是环境依赖冲突和权限问题。我整理了一份经过 12 个真实项目验证的 checklistPython 环境隔离强烈建议用pyenv创建独立环境而非全局 pip。原因Agent-Reach 依赖httpx异步 HTTP 客户端、pydantic配置验证、docker-pyDocker 交互这些库与系统 Python 的urllib3版本极易冲突。执行pyenv install 3.11.9 pyenv virtualenv 3.11.9 agent-reach-env pyenv activate agent-reach-env pip install agent-reach-cliDocker 权限修复针对permission denied while trying to connect to the docker api这是 Reddit 上最高频的报错。根本原因不是 Agent-Reach 代码问题而是 Linux 系统默认将/var/run/docker.sock的 owner 设为root:docker而普通用户不在docker组。解决方案分三步创建 docker 组若不存在sudo groupadd docker将当前用户加入组sudo usermod -aG docker $USER最关键一步注销并重新登录或重启机器让组权限生效。newgrp docker命令无效必须会话级重载。验证docker run hello-world应成功输出而非 permission denied。YouTube Adapter 依赖安装Agent-Reach 不自带yt-dlp和whisper.cpp需手动安装。注意版本兼容性yt-dlp必须 ≥ 2024.3.10修复了 YouTube 新版反爬安装命令pip install yt-dlp2024.3.10whisper.cpp推荐用 prebuilt binary避免编译失败下载地址https://github.com/ggerganov/whisper.cpp/releases。解压后将main二进制文件放入$PATH并设置环境变量WHISPER_CPP_PATH/path/to/whisper.cpp/mainAPI Key 安全存储绝对禁止在 CLI 命令中明文写--api-key xxx。Agent-Reach 支持三种安全方式环境变量export DEEPSEEK_API_KEYsk-xxx注意变量名必须匹配 Provider Registry 中定义的 key name配置文件~/.agent-reach/config.yaml内容为providers: deepseek-official: api_key: ${DEEPSEEK_API_KEY} # 支持环境变量引用 kimi-pro: api_key: kimi_xxx # 明文仅用于测试密钥管理服务支持 HashiCorp Vault需在 config.yaml 中配置vault_addr和vault_token。实操心得我在某次部署中发现即使 Docker 权限正确docker-py仍报 permission denied。排查发现是 SELinux 启用状态sestatus查看。临时关闭sudo setenforce 0永久关闭编辑/etc/selinux/config将SELINUXenforcing改为disabled。这是 CentOS/RHEL 系统特有的坑Ubuntu 用户无需考虑。3.2 Provider Registry 配置如何为 DeepSeek/Kimi/智谱写一份“能力画像”Provider Registry 的配置文件config.yaml是 Agent-Reach 的心脏。一份好的配置能让路由引擎做出精准决策。以下是为三个主流国产模型写的生产级配置示例包含所有关键字段和注释providers: # DeepSeek 官方 APIdeepseek-official deepseek-official: type: openai-compatible # 类型决定 Adapter 行为 base_url: https://api.deepseek.com/v1 api_key_env: DEEPSEEK_API_KEY # 环境变量名 health_check: endpoint: /models method: GET timeout: 5 capabilities: max_context_length: 131072 # 官方文档明确值 default_temperature: 0.7 streaming_supported: true input_format: messages # 支持 OpenAI messages 格式 output_format: openai # 返回标准 OpenAI schema rate_limit: requests_per_minute: 60 tokens_per_minute: 1000000 fallback: kimi-pro # 当 deepseek-official 不可用时自动切到 kimi-pro # Kimi APIkimi-pro kimi-pro: type: kimi base_url: https://api.kimi.ai/v1 api_key_env: KIMI_API_KEY health_check: endpoint: /health method: GET timeout: 3 capabilities: max_context_length: 1048576 # 注意这是热词中报错的根源必须精确填写 default_temperature: 0.3 streaming_supported: false # Kimi 当前不支持流式 input_format: prompt # 使用 prompt 字段非 messages output_format: kimi # 自定义 schema rate_limit: requests_per_minute: 30 tokens_per_minute: 500000 fallback: zhipu-glm4 # 智谱 GLM-4zhipu-glm4 zhipu-glm4: type: zhipu base_url: https://open.bigmodel.cn/api/paas/v4 api_key_env: ZHIPU_API_KEY health_check: endpoint: /models method: GET timeout: 4 capabilities: max_context_length: 32768 # GLM-4 的实际限制 default_temperature: 0.9 streaming_supported: true input_format: messages output_format: zhipu rate_limit: requests_per_minute: 100 tokens_per_minute: 200000 fallback: ollama-qwen2 # 最终降级到本地模型 # 全局路由策略 routing_policy: default_strategy: least-loaded # 优先选负载最低的 Provider fallback_chain: [deepseek-official, kimi-pro, zhipu-glm4, ollama-qwen2]关键字段解读与避坑指南type字段决定 Adapter 的行为逻辑。openai-compatible表示遵循 OpenAI API 规范/chat/completionsendpoint,messages字段kimi和zhipu则启用专用 Adapter处理其独特的 auth headerKimi 用Authorization: Bearer xxx智谱用Authorization: Bearer GLM-KEY-xxx和 request body 结构。max_context_length必须与官方文档严格一致。热词中api error: 400 this models maximum context length is 1048576 tokens的出现往往是因为配置文件里填了1000000或10485760多了一个 0导致 Adapter 在分块时计算错误把超长文本一次性发过去。Agent-Reach 会在请求前校验 payload token count若超限则主动报错避免浪费配额。fallback字段是稳定性基石。它不是简单的“主备切换”而是支持嵌套 fallbackdeepseek-official-kimi-pro-zhipu-glm4-ollama-qwen2。当第一个 Provider 返回503 Service Unavailable引擎会立即尝试下一个全程无感知。rate_limit不是装饰性字段。Agent-Reach 内置令牌桶算法实时监控每个 Provider 的请求速率。当requests_per_minute达到阈值后续请求会被排队或直接拒绝取决于routing_policy中的throttle_mode设置防止因突发流量触发平台级限流如api error: 400 this organization has been disabled。注意配置文件中的api_key_env必须与你设置的环境变量名完全一致包括大小写。我曾遇到一次故障环境变量设为DEEPSEEK_API_KEY但配置里写成deepseek_api_key导致 Agent-Reach 读不到密钥一直报no api key for provider route deepseek-official。这种低级错误用agent-reach config validate命令可提前发现。3.3 CLI 实战用 5 行命令完成 YouTube 视频摘要 Reddit 发帖全流程现在让我们用一个真实场景演示 Agent-Reach 如何把碎片化 API 变成原子化操作。目标获取 YouTube 视频《Agent-Reach 深度解析》的 transcript用 Kimi 摘要成 300 字再以 Markdown 格式发到 Reddit 的 r/LLM 子版块。步骤 1初始化 Provider Registry# 加载配置文件假设 config.yaml 在当前目录 agent-reach init --config ./config.yaml # 验证所有 Provider 健康状态 agent-reach health-check --all # 输出应为deepseek-official: OK, kimi-pro: OK, zhipu-glm4: OK步骤 2提取 YouTube transcript# 此命令会自动调用 yt-dlp 下载、whisper.cpp 转录、清洗文本 agent-reach run \ --task youtube-transcript \ --url https://www.youtube.com/watch?vabc123 \ --output ./transcript.txt \ --model kimi-pro # 指定用 kimi-pro 处理因其支持长上下文实操细节Agent-Reach 会先检查本地是否有yt-dlp和whisper.cpp若缺失则提示安装下载视频时自动跳过已存在的缓存文件~/.cache/agent-reach/youtube/ASR 过程中whisper.cpp的--language zh参数由 Adapter 根据视频元数据自动注入。步骤 3用 Kimi 生成摘要# 读取 transcript.txt调用 kimi-pro 摘要 agent-reach run \ --task summarize \ --input ./transcript.txt \ --output ./summary.md \ --model kimi-pro \ --max-output-tokens 300原理揭秘Adapter 检测到transcript.txt长度为 85000 tokens远超 Kimi 的max_context_length: 1048576于是启动分块策略将文本按句子切分为 12 个 chunk每个约 7000 tokens并发调用 Kimi 生成 12 个子摘要再用本地 Qwen2-7B 模型ollama-qwen2做最终聚合。整个过程对用户透明CLI 只显示进度条。步骤 4发布到 Reddit# 用 summary.md 内容发帖 agent-reach run \ --task reddit-post \ --subreddit LLM \ --title 【深度测评】Agent-Reach让 10 个 API 像 1 个一样工作 \ --body ./summary.md \ --model kimi-pro # 用 Kimi 生成标题和摘要保持风格一致权限处理Agent-Reach 会检查~/.agent-reach/reddit-token.json是否存在且未过期。若缺失或过期则自动打开浏览器引导你完成 Reddit OAuth 授权流程并将 refresh token 安全存储。步骤 5查看全流程日志与诊断# 生成本次执行的完整诊断报告 agent-reach debug --last --format json debug-report.json # 或者用可视化方式查看 agent-reach log --tail 100 --filter youtube|kimi|reddit日志价值debug-report.json包含每个环节的耗时youtube-transcript: 24.3s,summarize: 18.7s,reddit-post: 2.1s、使用的 Providerkimi-pro、token 消耗input_tokens: 85000,output_tokens: 298、网络延迟latency_ms: 420。这是性能优化的黄金数据。实操心得第一次运行youtube-transcript时我遇到whisper.cpp: command not found。排查发现whisper.cpp/main没有执行权限。解决命令chmod x /path/to/whisper.cpp/main。这个细节官网文档没提但却是新手必踩的坑。Agent-Reach 的 CLI 在报错时会明确提示whisper.cpp binary not executable比模糊的command not found友好太多。4. 实操过程与核心环节实现深入路由策略引擎与 Adapter 层的代码级解析4.1 路由策略引擎如何用 Pydantic 和 Rule Engine 实现动态决策Agent-Reach 的路由策略引擎不是简单的 if-else而是基于 Pydantic V2 的声明式规则定义 Drools 风格的规则引擎。其核心是PolicyRule模型和RuleEngine类。下面是一份可直接运行的简化版策略代码展示了如何实现“长文本自动选 Kimi短文本选 DeepSeek”的逻辑# policy_engine.py from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any import re class PolicyRule(BaseModel): 单条路由规则 id: str Field(..., description规则唯一ID) condition: str Field(..., descriptionPython 表达式条件如 len(input_text) 50000) action: str Field(..., description动作类型如 select_provider 或 fallback) params: Dict[str, Any] Field(default_factorydict, description动作参数) class RoutingPolicy(BaseModel): 完整路由策略配置 default_provider: str Field(..., description默认 Provider ID) rules: List[PolicyRule] Field(default_factorylist) # 示例策略根据输入长度选择 Provider POLICY_CONFIG RoutingPolicy( default_providerdeepseek-official, rules[ PolicyRule( idlong-text-kimi, conditionlen(input_text) 50000, # 输入文本长度 5 万字符 actionselect_provider, params{provider: kimi-pro} ), PolicyRule( idcode-review-zhipu, conditionre.search(r\\bdef\\s\\w\\(, input_text), # 检测 Python 函数定义 actionselect_provider, params{provider: zhipu-glm4} ), PolicyRule( idfallback-on-error, conditionerror_code in [401, 429], # HTTP 错误码 actionfallback, params{chain: [kimi-pro, zhipu-glm4]} ) ] ) class RuleEngine: 规则引擎执行器 def __init__(self, policy: RoutingPolicy): self.policy policy def evaluate(self, input_text: str, error_code: Optional[str] None) - str: 评估输入返回选定的 Provider ID # 1. 检查规则条件 for rule in self.policy.rules: try: # 安全执行条件表达式禁用危险函数 safe_globals {len: len, re: re, input_text: input_text, error_code: error_code} if eval(rule.condition, {__builtins__: {}}, safe_globals): if rule.action select_provider: return rule.params[provider] elif rule.action fallback: return rule.params[chain][0] # 返回链首 except Exception as e: # 规则执行异常跳过此规则 continue # 2. 无规则匹配返回默认 Provider return self.policy.default_provider # 使用示例 engine RuleEngine(POLICY_CONFIG) selected_provider engine.evaluate(这是一个很长的文本... * 10000) # 长文本 print(selected_provider) # 输出: kimi-pro selected_provider engine.evaluate(def hello(): pass) # Python 代码 print(selected_provider) # 输出: zhipu-glm4这段代码揭示了 Agent-Reach 的核心机制安全沙箱执行eval()被包裹在{__builtins__: {}}中禁用所有内置函数只允许len、re等白名单函数。这防止恶意规则注入如os.system(rm -rf /)。条件表达式即 DSL用户无需写 Python 代码只需在config.yaml中写condition: len(input_text) 50000。Agent-Reach 的配置加载器会将其解析为PolicyRule对象。多维度决策规则条件可以组合多个信号input_text长度、error_code、task_type如youtube-transcript、甚至current_load从 Provider Registry 获取的实时负载。这比静态路由强大得多。提示Agent-Reach 的生产版引擎还集成了 Prometheus 指标每条规则的匹配次数、执行耗时都会暴露为agent_reach_rule_match_total{rule_idlong-text-kimi}指标。你可以用 Grafana 做可视化观察哪条规则最常被触发从而优化策略。4.2 Adapter 层如何用 Template 和 Schema Mapping 处理千奇百怪的 APIAdapter 层是 Agent-Reach 的“外交官”负责把统一的内部请求翻译成各家 API 能听懂的语言。其核心是RequestTemplate和ResponseSchema。以下是以 DeepSeek 和 Kimi 为例的 Adapter 实现# adapters/deepseek_adapter.py from pydantic import BaseModel from typing import List, Dict, Any class DeepSeekRequest(BaseModel): DeepSeek API 的 Request Schema model: str deepseek-chat messages: List[Dict[str, str]] temperature: float 0.7 stream: bool False class DeepSeekAdapter: def __init__(self, base_url: str, api_key: str): self.base_url base_url self.api_key api_key def build_request(self, input_data: Dict[str, Any]) - Dict[str, Any]: 将通用 input_data 转换为 DeepSeek Request # input_data 示例: {task: summarize, text: xxx, max_output_tokens: 300} messages [ {role: system, content: 你是一个专业的摘要助手。请用中文生成不超过300字的摘要。}, {role: user, content: input_data[text]} ] # 构建 DeepSeek 专属 Request request_body DeepSeekRequest( modeldeepseek-chat, messagesmessages, temperatureinput_data.get(temperature, 0.7), streaminput_data.get(stream, False) ).dict() return { url: f{self.base_url}/chat/completions, headers: { Authorization: fBearer {self.api_key}, Content-Type: application/json }, json: request_body } # adapters/kimi_adapter.py class KimiRequest(BaseModel): Kimi API 的 Request Schema model: str kimi-math-7b prompt: str temperature: float 0.3 max_tokens: int 1024 class KimiAdapter: def __init__(self, base_url: str, api_key: str): self.base_url base_url self.api_key api_key def build_request(self, input_data: Dict[str, Any]) - Dict[str, Any]: 将通用 input_data 转换为 Kimi Request # Kimi 不支持 messages只接受 prompt 字段 system_prompt 你是一个专业的摘要助手。请用中文生成不超过300字的摘要。 user_prompt input_data[text] full_prompt f{system_prompt}\n\n{user_prompt} request_body KimiRequest( modelkimi-long-context, promptfull_prompt, temperatureinput_data.get(temperature, 0.3), max_tokensinput_data.get(max_output_tokens, 300) ).dict() return { url: f{self.base_url}/chat/completions, headers: { Authorization: fBearer {self.api_key}, Content-Type: application/json }, json: request_body }Adapter 的设计精髓在于“解耦”Request 解耦build_request()方法接收统一的input_data来自 CLI 或 API但输出完全符合目标 Provider 规范的url、headers、json。开发者无需关心 DeepSeek 要messagesKimi 要prompt这些细节被 Adapter 封装。Response 解耦Adapter 还负责parse_response()把各家 API 的 raw response可能是{ choices: [...] }或{ data: { result: xxx } }统一转换成标准的{text: xxx, usage: {input_tokens: 123, output_tokens: 45}}格式。这样上层业务逻辑永远面对同一接口。错误解耦当 DeepSeek 返回{error: {message: Invalid API key, code: invalid_api_key}}Adapter 会解析出error_type: invalid_api_key并映射到统一错误码AGENT_REACH_ERROR_AUTH_FAILED当 Kimi 返回{error: {message: Rate limit exceeded}}则映射为AGENT_REACH_ERROR_RATE_LIMIT。上层策略
返回列表