★ FEATURED ARTICLE 做Agent开发的人迟早都会碰到模型接入这一关。项目名里写的是“Agent实践2-自定义模型封装”这其实是把Agent开发里最基础也最容易忽略的一层单独拎了出来不直接在业务代码里调用各家模型厂商的SDK而是做一个统一的自定义模型封装层让Agent通过同一套接口自由切换模型供应商。这不是炫技而是项目规模变大后必然要做的收敛。这篇文章会把封装层的设计思路、代码实现、框架接入和并发优化完整过一遍适合正在做Agent应用开发、想在多模型环境下保持代码整洁的朋友参考。1. 为什么Agent项目需要自定义模型封装1.1 摆脱“框架绑定”与“厂商绑定”的双重困境先说现象。你去翻开源的Agent项目LangChain有自己的一套BaseChatModelCrewAI出了独立的LLM类AutoGen用的是模型配置字典Dify则是平台级的模型供应商管理。每个框架都声称支持多模型但它们的抽象方式、工具调用格式、消息结构都不一样。更麻烦的是底层厂商SDK本身差别就很大OpenAI的chat.completions、Anthropic的messages、本地Ollama的/api/chat字段、角色、工具schema各不相同。如果你的Agent业务代码直接依赖这些SDK一旦框架升级、厂商接口变化、需要接入新的模型供应商改动面就会爆炸。我自己最早就是在业务代码里东一个openai.ChatCompletion.create西一个anthropic.Anthropic乍看能用等想统一日志、统一重试、统一统计成本的时候才反应过来这堆代码已经处处漏风。自定义模型封装的核心目标就是把“业务逻辑”和“模型供应商差异”彻底隔离让上层Agent代码只面对一个稳定的接口。1.2 统一封装带来的实际收益封装带来的收益不是理论上的是你能直接摸到的一键切换模型上线时用GPT-4o成本敏感时段切到国产模型或本地模型某个供应商限流了自动fallback到备选。只要这些逻辑收在封装层业务流程一行不用动。统一观测Agent跑挂了你能从封装层的日志里看到某次调用用了多少token、耗时多久、报了什么错。没有这层排查问题只能逐条翻各厂商的SDK内部日志非常痛苦。统一重试与限流把退避重试、并发信号量、超时策略收敛在封装层避免每个业务函数各自写一套行为还能保持一致。方便测试封装层可以注入MockTransport测试Agent逻辑时完全不用发真实请求速度和安全都更好。1.3 封装的边界别把一个“薄层”做成“全家桶”这是很容易上头的一个点。封装不是把模型SDK的所有能力包一遍你不需要在统一接口里暴露embedding、微调、文件接口这类东西。Agent需要的核心能力就两个聊天补全含工具调用和流式输出。把它俩做成稳定的协议其它能力继续走原生SDK反而是更健康的状态。接口越薄越不被厂商特性拖累新增一个供应商时只是写一个适配器而已。2. 封装层接口设计与协议选型2.1 最小接口消息序列进统一响应出不管哪个厂商Chat模型的调用本质都一样输入一串消息输出一条助手消息。我把统一接口设计成from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Any, AsyncIterator dataclass class ToolCall: id: str name: str arguments: dict dataclass class Usage: prompt_tokens: int 0 completion_tokens: int 0 total_tokens: int 0 dataclass class ChatResult: content: str tool_calls: list[ToolCall] | None None usage: Usage | None None raw: Any None # 保留原始响应便于排查 dataclass class StreamChunk: delta_content: str delta_tool_calls: list[ToolCall] | None None done: bool False usage: Usage | None None class ChatModel(ABC): abstractmethod def chat(self, messages: list[dict], tools: list[dict] | None None, **kwargs) - ChatResult: 非流式对话 abstractmethod async def achat(self, messages: list[dict], tools: list[dict] | None None, **kwargs) - ChatResult: 异步对话 abstractmethod def stream(self, messages: list[dict], tools: list[dict] | None None, **kwargs) - AsyncIterator[StreamChunk]: 流式对话这个接口看起来简单但它是刻意保持简单的。messages用统一的OpenAI风格消息列表role为system/user/assistant/tooltools用统一工具定义格式。kwargs传model、temperature、max_tokens这类通用参数。各厂商的差异全部下沉到适配器内部消化上层Agent代码始终面对同样的数据结构。2.2 为什么把OpenAI兼容协议当成统一底座我选OpenAI风格作为内部协议不是因为它最好而是它已经是事实上的生态标准。理由有三第一绝大多数模型服务都支持OpenAI兼容接口。OpenAI、DeepSeek、Moonshot、智谱、本地Ollama、vLLM、LM Studio甚至很多模型网关都暴露/v1/chat/completions。这意味着同一个OpenAI适配器通过改base_url和api_key就能对接一大批模型。第二消息格式最简单直接。OpenAI风格的消息就是一个{role, content}数组工具调用是消息里的独立字段读起来直观测试构造也容易。第三Agent框架几乎都以它为参考实现。LangChain、LlamaIndex、CrewAI的工具调用设计都是OpenAI风格接入时心智负担最小。Anthropic的Messages API稍微特殊system消息不是消息数组的一部分而是独立参数工具schema不是标准的JSON Schema而是input_schema字段assistant消息里工具调用是content里的一种block类型。这些差异正好由适配器来翻译统一协议外部完全感知不到。2.3 错误模型的标准化封装层另一个重要产出是标准化的异常。不同厂商报错方式五花八门OpenAI用HTTP状态码加错误体Anthropic有自己的错误类型Ollama可能直接返回字符串。我把它们统一成class ModelError(Exception): ... class ModelAuthError(ModelError): ... class ModelRateLimitError(ModelError): ... class ModelTimeoutError(ModelError): ... class ModelContextOverflowError(ModelError): ... class ModelBadRequestError(ModelError): ...这样Agent调度层做决策时就很简单ModelRateLimitError触发退避重试ModelAuthError直接告警别重试ModelContextOverflowError走上下文压缩或截断逻辑。标准化错误模型是“封装层是否合格”的第一个硬指标因为重试策略是Agent稳定性的基石。3. 手写一个可用封装层关键代码实现3.1 OpenAIStyleAdapter一个适配器覆盖大半主流模型基于上面的分析我优先实现了一个OpenAI风格适配器。用httpx而不是OpenAI官方SDK的最大好处是没有版本绑定问题协议一目了然而且能方便地注入MockTransport写测试。import json import httpx class OpenAIStyleAdapter(ChatModel): def __init__(self, base_url: str, api_key: str, model: str, timeout: float 30.0, max_retries: int 2): self.http httpx.Client( base_urlbase_url, headers{Authorization: fBearer {api_key}}, timeouthttpx.Timeout(timeout), ) self.model model self.max_retries max_retries def _build_payload(self, messages, tools, **kwargs): payload { model: kwargs.pop(model, self.model), messages: messages, stream: False, } if tools: payload[tools] tools payload.update(kwargs) return payload def chat(self, messages, toolsNone, **kwargs) - ChatResult: payload self._build_payload(messages, tools, **kwargs) # 简化写法实际生产环境要配合后面的退避重试 resp self.http.post(/v1/chat/completions, jsonpayload) if resp.status_code ! 200: self._raise_for_status(resp) data resp.json() return self._normalize_response(data) def _normalize_response(self, data) - ChatResult: msg data[choices][0][message] tool_calls [] for tc in msg.get(tool_calls, []) or []: tool_calls.append(ToolCall( idtc[id], nametc[function][name], argumentsjson.loads(tc[function][arguments] or {}), )) usage data.get(usage) return ChatResult( contentmsg.get(content) or , tool_callstool_calls or None, usageUsage(usage[prompt_tokens], usage[completion_tokens], usage[total_tokens]) if usage else None, rawdata, )这个适配器可以直接对接OpenAI也可以把base_url指向http://localhost:11434/v1对接Ollama或者指向任何OpenAI兼容服务。项目里我常做一个小工厂函数从配置项生成对应适配器新模型接入就变成一行注册。3.2 Anthropic适配器最难啃的消息翻译Anthropic是另一种主流模型它的消息结构差异处理不好Agent调用直接报400。核心翻译逻辑有四条system提取把messages里rolesystem的消息抽出来作为system参数单独传剩余的user/assistant/tool消息留在messages数组里。内容块转换OpenAI风格的content字符串直接映射到Anthropic的content文本块assistant带工具调用时转换为tool_use块arguments作为dict传进去并带上id。工具结果映射OpenAI的roletool消息在Anthropic里要变成roleuser且content为tool_result块每个块带tool_use_id对应之前的调用ID。工具schema转换把OpenAI的JSON Schema工具定义转成Anthropic的input_schema响应里解析tool_use块回统一的ToolCall。class AnthropicAdapter(ChatModel): def __init__(self, api_key: str, model: str): # Anthropic API 地址固定初始化代码这里略去 ... def _to_anthropic_messages(self, messages: list[dict]) - tuple[str, list[dict]]: system_parts [] converted [] for msg in messages: if msg[role] system: system_parts.append(msg.get(content, )) continue if msg[role] tool: converted.append({ role: user, content: [{ type: tool_result, tool_use_id: msg[tool_call_id], content: str(msg.get(content, )), }] }) continue converted.append({role: msg[role], content: msg.get(content, )}) return \n.join(system_parts), converted这几步转换看起来琐碎但漏任何一步都会出问题。尤其是tool_result忘记放到user消息里模型就无法正确理解工具执行结果Agent会陷入“反复调用同一个工具”的死循环。这里的适配器就像是电源转换插头两头互不认识没关系插头中间把电路转对了就行。3.3 流式输出Agent长任务体验的关键Agent在工具推理和长文本生成时用户能等的耐心上限很低流式是刚需。不同厂商的流式协议也不同OpenAI返回SSE事件流每个chunk里有choices[0].deltaAnthropic的流式事件类型更多有content_block_delta、message_delta等。统一抽象我用StreamChunk适配器把各自的流式事件转成统一的delta_content和delta_tool_calls。工具参数在流式场景下往往是分片拼起来的JSON所以我在流结束前先不解析等工具调用的arguments收齐后再json.loads。如果流中途断了工具参数不完整就触发重试或让Agent重新生成一步这是后续排查里很重要的处理逻辑。需要注意流式模式下很多厂商不返回usage字段。要统计token成本要么自己在输出端估算要么在适配器里记账请求侧的token可以数生成侧token按生成的chunk累计。我建议至少在封装层留一个累加器把用户看到的“实时反馈”和“成本统计”一起解决掉。4. 封装层接入Agent框架与并发优化4.1 接入LangChain实现BaseChatModelLangChain是目前生态最全的Agent框架之一。如果你的封装层想接入它不用改内部实现只需要写一个适配类继承BaseChatModel把_generate方法指向你的ChatModel.chat把_stream方法指向stream。之后所有LangChain的Agent、Tool、Memory都会基于你的模型工作。from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.outputs import ChatResult as LangChainChatResult class CustomModelLangChainAdapter(BaseChatModel): def __init__(self, model: ChatModel, **kwargs): super().__init__(**kwargs) self._model model def _generate(self, messages, stopNone, run_managerNone, **kwargs): # 把LangChain消息对象转成统一消息列表调用 self._model.chat ... return LangChainChatResult(generations[...])这里有一个坑LangChain的消息对象和统一消息格式不一样要先写一个转换函数把SystemMessage、HumanMessage、AIMessage、ToolMessage转成{role, content}字典。这个转换放在适配类里即可别污染封装层内部。4.2 接入CrewAI和Dify的思路CrewAI近几个版本允许传入自定义LLM对象但更稳妥的做法有两个一是把封装层包成LangChain模型再交给CrewAI使用二是如果封装层本身暴露一个OpenAI兼容的HTTP服务CrewAI直接用base_url指向它完全绕开SDK差异。Dify是平台型产品它有自己的模型供应商插件机制普遍做法是写一个Dify的模型provider插件把封装层的适配器包进去。这两条思路说明一个原则封装层对外提供多种“插座”形态比如Python对象、LangChain模型、OpenAI兼容HTTP服务但核心的转换逻辑只有一份。4.3 并发控制与带抖动的重试Agent跑起来之后并发问题迟早找你。我实测过一个典型场景50个Agent任务并发启动每个任务内部会连续多次调用模型瞬间对模型网关打出几百个并发请求结果就是大面积的429和504。解决办法分三层第一层连接池复用。用httpx.Client复用HTTP连接不要让每个请求都重新建TCP加TLS连接。我见过并发上来以后大量时间花在握手上的案例复用连接后延迟肉眼可见地下降。第二层信号量限流。在封装层用一个asyncio.Semaphore或线程池的threading.Semaphore限制最大并发数比如20。信号量的作用类似于收费站控制车流让超出能力的请求排队而不是一拥而上。别小看这个设计它能让整个系统的行为变得可预测。第三层带抖动的指数退避重试。对429和5xx错误重试退避时间用min(2 ** attempt, 8)秒再乘上一个随机因子避免所有请求同时重试形成“重试风暴”。对4xx错误不重试重试只会浪费配额。我的默认配置是最多重试3次超时分成连接超时和读取超时两档连接超时5秒读取超时根据任务复杂度设成60到120秒。4.4 Token管理与上下文溢出防御模型上下文窗口是硬约束。Agent循环里多轮工具调用很容易把上下文塞爆你会在最关键时刻收到context length exceeded错误。我建议在封装层加一个fit_messages_to_context方法策略很简单统计每条消息的token数OpenAI模型用tiktoken其它模型用字符数估算或调用服务自身的tokenize接口。如果总token超过阈值先丢弃最老的user/assistant消息保留system和最近几轮。如果还是很长把任务的关键信息外置只留摘要放回上下文。def fit_messages_to_context(self, messages, max_context_tokens, system_keepTrue): total self.count_tokens(messages) if total max_context_tokens: return messages kept list(messages) while self.count_tokens(kept) max_context_tokens and len(kept) 2: index 1 if system_keep and kept[0][role] system else 0 removed kept.pop(index) if removed[role] tool: break # 工具结果消息通常成对出现单独裁掉会破坏结构交给上层处理 return kept实际项目里我通常把max_context_tokens设成模型窗口的80%剩下的20%留给模型输出。这个冗余非常重要模型输出max_tokens设得再高也不会超出剩余窗口。5. 常见问题与排查技巧实录5.1 消息格式不兼容导致的400这是接入新模型厂商最常见的错误。报错通常很抽象比如OpenAI风格请求发到Anthropic返回messages: roles[0] must be alternating或者extra properties not permitted。排查思路就一句话先用最小消息集在两个协议之间手动对照。我写过一个测试用同一组fixture消息分别经过两个适配器断言输出格式为各自厂商文档要求的规范结构。这类测试不用发真实API跑CI成本几乎为零但能抓住80%的适配bug。5.2 工具调用解析错位Agent认为没有可用工具Anthropic适配器有个经典问题OpenAI的工具调用是assistant消息里的tool_calls字段而Anthropic是content里的tool_use块。如果适配器只去读message.tool_calls就会拿到空列表Agent会误以为自己不能调用工具直接拒绝执行任务。排查方法是在适配器加一条黄金日志每次解析完成把“原始响应的工具字段”和“统一结果的tool_calls”都打出来一眼就能看出映射是否丢失。5.3 流式中途断流与工具参数不完整流式断流这个事在弱网或长响应时很容易出现表现是Agent生成到一半停了或者工具调用的arguments是残缺JSON。我的处理方法是如果流在工具调用过程中断了不第一时间把残缺结果交给工具执行器而是把残缺的arguments拼上错误信息作为一条新的tool消息回传模型让模型自己修正。这个“让模型纠正自己”的模式比在业务层硬编码解析JSON稳得多。5.4 并发上来后的429/504排查排查这种问题不要一上来就改重试参数。先看日志分清是哪一层出的问题是封装层发出的请求过多信号量没生效还是模型网关已经限流429还是连接被耗尽出现大量connection reset。我整理了一个速查表现象常见原因首选处理偶发429触达模型限流阈值指数退避重试jitter随机化批量429/503并发超过模型服务能力信号量限流降低最大并发大量连接重置连接池耗尽或服务端断连复用httpx.Client检查超时配置长响应被截断read timeout过短分级超时读取超时单独调大上下文长度报错多轮工具调用撑爆窗口封装层fit函数、压缩历史5.5 Token统计与实际不一致流式响应经常没有usageOpenAI兼容服务和Anthropic有时候返回的usage也是估算值。我的经验是把“计费口径”和“日志口径”分开。日志里记录封装层累加的token数用于监控趋势计费以供应商账单为准但要定期抽验。Anthropic开了prompt caching的话响应里会出现cache_creation_input_tokens和cache_read_input_tokens字段这些字段在适配器里一定不要忽略直接塞进统一的Usage数据结构否则你看到的成本永远对不上。5.6 用MockTransport做离线测试最后分享一个我一直在用的做法封装层在做任何真实调用之前先用MockTransport验证协议正确性。httpx的MockTransport可以模拟任意响应而不用起真实服务。我在测试用例里准备了OpenAI格式和Anthropic格式的固定响应体跑一遍chat和stream断言统一的ChatResult字段齐全。这样每次升级依赖、加新适配器都是先过一轮不到一秒的离线测试再上真实环境。你敢说你的模型适配改得上手就没错过一次吗我是不敢所以我把这关前置了。老实说我做第一个Agent项目时也嫌封装麻烦觉得直接调SDK最快。后来项目到了第二阶段要同时支撑多个模型、多套环境、线上故障排查才开始后悔当初没早做。这一篇实践整理下来最大的经验就一句接口稳定比功能丰富重要模型差异全部关在适配器里你的Agent业务才能走得更远。
阅读完成 · 觉得有帮助？
