Agent实践系列走到第二篇这篇聊聊自定义模型封装。先交代一下背景我最近在做一个多Agent协作项目需要把一套内部微调过的模型接入现有的Agent框架。按常规思路直接调用框架默认的模型接口然后填API Key就完事了。但真正跑起来就会发现事情远没有那么简单——模型返回的格式、工具调用的触发、流式输出的处理、并发上限的控制这些全都要在“封装层”这一环解决。这篇文章就把我踩过的坑和最终沉淀下来的方案完整梳理一遍希望能给正在做Agent开发的同行省点时间。先说清楚这篇内容是什么、能做什么这是一套“把任意模型接入Agent框架”的可复用方法核心是自定义模型封装层的设计与实现。适合两类读者——第一类是已经在用LangChain、Dify这类框架但想摆脱对单一模型厂商依赖的开发者第二类是准备把私有化部署的模型接入Agent系统又不想被框架默认接口绑死的团队。下文所有代码和思路均来自实际项目不是PPT架构。1. 为什么绕不开自定义模型封装1.1 默认模型接入的局限绝大多数Agent框架开箱即用的体验确实好。以LangChain为例配置好OPENAI_API_KEY加载一个ChatOpenAIAgent就能跑起来。但这里有个隐藏前提你默认接受了框架为你选好的模型厂商和协议格式。一旦场景偏离这个前提问题立刻冒出来。比如企业内部私有化部署的模型往往只提供一个兼容OpenAI格式但又不完全一致的HTTP接口再比如某些垂直领域微调模型返回的JSON结构里带额外的业务字段Agent的解析器根本不认识更麻烦的是有些模型支持的工具调用参数格式跟框架默认的ToolCall schema对不上Agent执行工具时直接报错。我最初也觉得框架都提供了各种Model类的子类覆盖应该够了。实际去翻源码才发现适配层通常只处理了一两家主流模型的协议其他模型的接入都得自己写。而“自己写”这件事如果只是零散地改几个方法后面维护就是灾难。正确的做法是先做一层统一封装把模型的差异隔离在外面让Agent只跟你的封装层对话。1.2 自定义封装解决的核心问题用一句话概括自定义模型封装的价值它定义了Agent与模型之间的契约让你随时可以换模型、加逻辑、控成本而不需要动Agent的核心代码。具体到我这次实践中封装层至少解决了四个问题。第一是协议适配——把内部模型的HTTP接口翻译成框架理解的标准ChatModel接口第二是行为注入——比如系统提示词的统一拼接、敏感信息过滤、返回内容的脱敏第三是稳定性兜底——超时重试、熔断、降级都收口在封装层里Agent不用管第四个是观测——封装层统一埋点所有模型的请求延迟、Token消耗、返回状态都能拉到监控系统里。这四个点任何一个放到Agent主流程里去做都会让代码变得极其混乱。封装层的存在本质上是把“模型通信”的复杂度从“Agent编排”的复杂度中剥离出来。1.3 前置知识准备写自定义封装之前有几点基础需要先打牢。第一搞清楚你用的Agent框架中“模型”这个概念到底对应哪个基类。LangChain中如果你做的是对话式Agent一般继承BaseChatModel如果是纯补全式任务可能用BaseLLM。第二要理解框架怎么调用模型的——同步、异步、流式这三条路径对应的方法分别是_generate、_agenerate、_stream封装时最好都实现否则Agent在某些场景下会退化成阻塞式调用。第三建议先在框架自带模型上跑通一个最小Agent观察它内部到底调用了模型的哪些方法、传入了哪些参数再动手写自己的封装。很多新手上来就继承基类结果连bind_tools在底层是如何把schema传给模型的都没搞明白自然封装不对。我当初就是先断点调试了一遍ChatOpenAI的调用链才彻底理解了整个数据流这个步骤省不得。2. 设计一个可复用的模型封装层2.1 先搞清楚Agent框架里模型的职责在Agent的推理循环里模型承担的职责远不止“回答用户问题”这一项。一次典型的Agent执行过程是这样的用户输入先被组装成消息列表交给模型预测模型如果决定调用工具返回中会携带结构化的工具调用参数Agent解析这些参数执行对应工具把结果追加回消息历史然后再次调用模型循环往复直到模型给出最终答案。这意味着封装层必须对“工具调用”这个交互有完整的支持。如果模型本身不支持原生工具调用封装层要么负责把工具schema“翻译”成模型认识的格式要么在模型输出文本后自行解析出工具调用的JSON片段。前者叫原生接入后者叫解析式接入。现在的模型大部分都宣称支持工具调用但返回的JSON字段名经常不一样比如有的叫tool_calls有的叫function_call还有的嵌套在choices里——封装层必须把这些差异磨平。另一个职责是维持对话状态的正确性。Agent会把整段对话历史包括工具执行结果、系统提示发给模型封装层需要确保这些内容以模型能理解的方式传递。有些模型的接口要求把工具结果放在特定字段里而不是普通文本消息这也要在封装里做转换。2.2 需要实现的接口与契约以LangChain为例一个自定义ChatModel的核心接口由两部分组成必须实现的抽象方法和推荐覆盖的增强方法。抽象方法包括_generate接收messages列表返回ChatResult和_llm_type返回模型类型标识字符串。如果你希望Agent支持异步调用还要实现_agenerate支持流式输出就要实现_stream。增强方法里最关键是bind_tools。这个方法在框架内置模型里一般已经实现好了但自定义封装里必须自己管理接收一组工具schema在调用模型时拼接成模型要求的格式。比如你的模型接口要求的工具格式是{tools: [{name: search_news, description: 搜索新闻, parameters: {...}}]}而框架传入的是LangChain的BaseTool列表那封装层就要做字段映射。还有一点容易被忽略temperature、max_tokens这些采样参数的传递。框架的BaseChatModel构造函数通常有标准参数但自定义模型有自己的参数名。封装层要决定是沿用框架的命名还是暴露自己的参数接口然后在调用模型API时做参数映射。2.3 封装层的整体结构设计整体结构上我建议把封装层拆成三个子模块协议适配器、生命周期管理器、观测中间件。协议适配器负责跟具体模型API打交道包括HTTP请求的构造、认证头注入、响应解析。这个模块的变化频率最高——外部模型接口一调整只改适配器就行。生命周期管理器负责请求之外的事连接池管理、重试策略、超时控制、并发限流。这些逻辑跟具体模型无关属于通用能力单独抽出来既便于跨模型复用也方便做故障演练。观测中间件是给所有请求统一加日志、埋点、耗时统计的。我见过不少团队把这些埋点代码散落在Agent各环节结果排查问题时要翻好几个服务。放在封装层里一条链路的所有模型调用都被覆盖到观测起来非常清晰。这三个模块的组合方式推荐用构造器注入适配器是可变的生命周期管理器是固定的观测中间件则用装饰器或回调的方式挂在适配器外面。这种设计下将来接入第二个模型只需要写一个新的协议适配器其余两个模块直接复用。3. 实操从零封装一个自定义模型3.1 场景定义以“企业内部数据脱敏模型”为例理论讲再多不如直接跑一个完整的例子。我这次要解决的场景是这样的公司内部有一个微调过的对话模型对外提供HTTP接口但它在返回结果里可能携带内部敏感信息比如员工姓名、内部系统路径等需要在接入Agent之前做脱敏。同时这个模型接口遵循OpenAI的请求格式但响应里有个非标准的business_meta字段用于返回内部调用链路的追踪ID。需求拆解出来有四点一封装一个标准ChatModel让Agent不感知内部模型的差异二在请求发出前注入内部专用的系统提示词三在响应返回后对文本做一层脱敏过滤四把business_meta字段提取出来附加到响应的元信息里方便日志追踪。这个场景很有代表性——大多数企业接内部模型都会遇到类似的“标准协议私有字段业务逻辑”混合体。我们用LangChain的BaseChatModel来实现。3.2 核心代码实现自定义ChatModel直接看核心代码。先定义协议适配器它负责跟内部模型的HTTP接口通信import requests import json from typing import List, Dict, Any, Optional class InternalModelAdapter: 内部模型HTTP协议适配器 def __init__(self, endpoint: str, api_key: str, timeout: int 30): self.endpoint endpoint self.api_key api_key self.timeout timeout self.session requests.Session() self.session.headers.update({ Authorization: fBearer {api_key}, Content-Type: application/json }) def chat_completion(self, messages: List[Dict[str, Any]], tools: Optional[List[Dict]] None, temperature: float 0.7, max_tokens: int 1024) - Dict[str, Any]: payload { model: internal-chat-v2, messages: messages, temperature: temperature, max_tokens: max_tokens, } if tools: payload[tools] tools payload[tool_choice] auto resp self.session.post(self.endpoint, jsonpayload, timeoutself.timeout) resp.raise_for_status() return resp.json()这里我用了requests.Session而不是每次新建连接理由是Agent推理循环里会频繁调用模型Session可以复用TCP连接显著降低握手开销。timeout也设成了可配置参数防止模型接口卡死拖垮Agent。关键点来了payload里传的tools格式必须跟内部模型API约定的schema完全一致。如果你直接拿LangChain的Tool对象转JSON丢过来字段名对不上比如LangChain是name模型API是function.name模型就会忽略工具调用Agent的循环直接就断了。所以适配器里要有schema转换逻辑这里先预留口径后面在bind_tools里处理。接下来是核心的ChatModel封装类from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import ( BaseMessage, HumanMessage, SystemMessage, AIMessage, ToolMessage ) from langchain_core.outputs import ChatResult, ChatGeneration from langchain_core.callbacks import CallbackManagerForLLMRun from typing import List, Optional, Any, Dict class InternalChatModel(BaseChatModel): 自定义内部模型封装 endpoint: str api_key: str temperature: float 0.7 max_tokens: int 1024 adapter: InternalModelAdapter None internal_system_prompt: str 你是内部专属助手请严格遵守数据安全规范... def __init__(self, **kwargs): super().__init__(**kwargs) self.adapter InternalModelAdapter( endpointself.endpoint, api_keyself.api_key ) property def _llm_type(self) - str: return internal-chat def _convert_messages(self, messages: List[BaseMessage]) - List[Dict]: 将LangChain消息转换为模型API格式 converted [] for msg in messages: if isinstance(msg, SystemMessage): converted.append({role: system, content: msg.content}) elif isinstance(msg, HumanMessage): converted.append({role: user, content: msg.content}) elif isinstance(msg, AIMessage): if msg.tool_calls: # 如果AI消息里有工具调用原样透传给模型 converted.append({ role: assistant, content: msg.content or , tool_calls: [ { id: tc[id], type: function, function: { name: tc[name], arguments: json.dumps(tc[args]) } } for tc in msg.tool_calls ] }) else: converted.append({role: assistant, content: msg.content}) elif isinstance(msg, ToolMessage): converted.append({ role: tool, tool_call_id: msg.tool_call_id, content: msg.content }) else: converted.append({role: user, content: str(msg.content)}) return converted def _generate( self, messages: List[BaseMessage], stop: Optional[List[str]] None, run_manager: Optional[CallbackManagerForLLMRun] None, **kwargs: Any, ) - ChatResult: # 注入内部系统提示词 all_messages [SystemMessage(contentself.internal_system_prompt)] messages # 转换为API格式 api_messages self._convert_messages(all_messages) # 调用适配器 response self.adapter.chat_completion( messagesapi_messages, temperatureself.temperature, max_tokensself.max_tokens ) # 从响应中提取内容 choice response[choices][0] content choice[message].get(content) or # 脱敏处理这里用简单的正则替换做演示 content self._desensitize(content) # 提取业务元信息 business_meta response.get(business_meta, {}) # 构造AI消息保留工具调用与元信息 ai_message AIMessage( contentcontent, additional_kwargs{business_meta: business_meta} ) # 如果有工具调用解析进tool_calls if tool_calls in choice[message]: parsed_calls [] for tc in choice[message][tool_calls]: parsed_calls.append({ id: tc[id], name: tc[function][name], args: json.loads(tc[function][arguments]) }) ai_message.tool_calls parsed_calls return ChatResult(generations[ChatGeneration(messageai_message)]) def _desensitize(self, text: str) - str: 简单脱敏隐藏工号、手机号等敏感信息 import re # 隐藏形如工号:12345的内容 text re.sub(r(工号[:]?)\d, r\1***, text) # 隐藏内部路径 text re.sub(r/data/internal/[a-z0-9/_-], [内部路径], text) return text def bind_tools(self, tools): 将LangChain工具列表转换为模型API的工具格式 api_tools [] for tool in tools: api_tools.append({ type: function, function: { name: tool.name, description: tool.description, parameters: tool.args_schema.schema() if tool.args_schema else {type: object, properties: {}} } }) # 这里通过绑定一个内部参数来传递tools实际生产建议用ToolChoice类型 # 为简洁演示直接在实例上临时保存 self._bound_tools api_tools return self这里有几个设计细节值得单独说。第一_convert_messages里对AIMessage的tool_calls做了反向转换。因为Agent二次循环时会把上一次的工具调用作为消息历史重新发给模型如果这里不做转换模型收到的历史里工具调用就丢失了Agent就进入了“重复调用同一工具”的死循环。第二脱敏逻辑放在封装层而不是放在Agent外面。这样不管未来Agent怎么编排工作流模型返回的每个文本都被脱敏过不会遗漏。第三bind_tools暂时用了一个临时变量_bound_tools保存工具列表。实际生产环境应该把它放到kwargs里或者生成一个新的模型实例这样更符合不可变设计也更线程安全。我这里是为了演示简洁大家自己用的时候建议采用更严谨的方式。3.3 接入Agent并跑通推理循环封装类写好后接入Agent就非常顺滑了。以LangChain的create_react_agent为例假设我们复用工具调用Agent只需要把InternalChatModel实例传进去from langchain.agents import create_react_agent, AgentExecutor from langchain_core.tools import tool tool def search_news(keyword: str) - str: 根据关键词搜索最新新闻。 # 模拟新闻搜索实现 return f关于{keyword}的最新新闻行业大会将于下周召开。 model InternalChatModel( endpointhttp://internal-model-api:8080/v1/chat/completions, api_keyinternal-key, temperature0.5 ) # 绑定工具 model model.bind_tools([search_news]) agent create_react_agent(model, [search_news]) executor AgentExecutor(agentagent, tools[search_news], verboseTrue) result executor.invoke({input: 帮我搜索一下人工智能大会的最新动态}) print(result[output])跑通后有一个体验上的变化Agent在调用模型时business_meta追踪ID会出现在AIMessage.additional_kwargs里排查问题时直接按这个ID查内部模型的日志链路清晰得很。实际的推理循环中模型返回了工具调用Agent解析出search_news的参数执行完后把结果作为ToolMessage回填给模型模型再生成最终回答。整个过程跟使用官方模型完全一致区别只在内部协议的适配和业务字段的注入。这正是封装层想要的效果——对Agent透明。3.4 关键参数与调试技巧实际调试中有几个参数决定了Agent能否稳定运行。temperature建议设在0.3到0.5之间尤其是涉及工具调用时。温度太高模型可能改写出不存在的工具名或者参数结构错乱Agent直接解析失败。max_tokens的设定要预留出工具调用的JSON长度——很多模型API的max_tokens只算生成部分的token而一个工具调用可能本身就占用200-400 token如果设太小模型还没输出完工具参数就被截断返回的JSON一定是残缺的Agent解析直接抛异常。关于超时控制建议连接超时3秒读超时根据模型响应速度适度放宽到30-60秒。如果Agent场景需要多次串行调用模型读超时太长会导致用户等待过久太短又容易在模型思考复杂问题时误杀。我一般会把“是否重试”和“是否超时失败”分开配置重试只处理连接级别的错误比如ConnectionError、Timeout(connect)读超时则不再重试直接返回错误给Agent让Agent决定是换模型还是换提示词。调试时最实用的一个技巧把所有请求和响应日志都打开。我通常在适配器里加一个debug开关打印完整的请求payload和响应body。出问题先对比payload看自己被框架“翻译”成什么样子了问题往往一目了然。4. 进阶生产环境中的封装细节4.1 流式输出与Agent交互的冲突处理谈到流式输出很多人的第一反应是“加上这个函数Agent就能打字机效果了”。但真正进入Agent场景后流式输出和工具调用天然存在冲突。原因在于工具调用需要模型返回完整的结构化JSONAgent才能解析执行而流式输出是把内容逐字吐出来如果中途截断工具调用的JSON无法被完整拼接。所以我的建议是在Agent场景下默认关掉流式走_generate的完整返回路径只有在“纯对话、不带工具”的轻量场景下才启用_stream。如果你确实需要在Agent展示流式效果比如漂亮的前端聊天框那要在封装层做一个缓冲模型流式返回的内容暂存在一个队列里等待一段稳定时间比如300ms没有新token再尝试把累积内容解析成工具调用。解析失败则当作普通文本继续展示。这个缓冲机制虽然不复杂但可以极大改善用户体验。LangChain的_stream实现上有一个要求每产生一个chunk都要通过run_manager.on_llm_new_token回调否则外部监听器收不到流式事件。很多自己写封装的人漏了这一步导致前端始终等不到内容还以为模型卡了。4.2 并发与限流Agent应用最容易被低估的是并发压力。一个Agent在处理复杂任务时可能对同一个模型发起多次顺序调用单用户理解很简单但如果是10个用户同时用模型API瞬间就被打爆了。尤其是企业内部模型通常没有自动扩容能力限流必须提前做在封装层。我在封装层里加了一个简单的信号量限流器import threading import time class RateLimiter: 简单限流器固定窗口内限制并发数 def __init__(self, max_concurrent: int 8): self._semaphore threading.Semaphore(max_concurrent) self._lock threading.Lock() self._window_start time.time() self._request_count 0 self._max_per_window 100 def acquire(self): with self._lock: now time.time() if now - self._window_start 60: self._window_start now self._request_count 0 if self._request_count self._max_per_window: raise RuntimeError(请求频率超过限制请稍后再试) self._request_count 1 self._semaphore.acquire() def release(self): self._semaphore.release()这个限流器在适配器里包一层class RateLimitedAdapter: def __init__(self, inner_adapter, limiter): self._inner inner_adapter self._limiter limiter def chat_completion(self, *args, **kwargs): self._limiter.acquire() try: return self._inner.chat_completion(*args, **kwargs) finally: self._limiter.release()关键点在于超时时间和限流等待时间要分开计算。如果限流等待也算进超时里高并发时所有请求都会因为排队而超时引发连锁失败。正确做法是限流等待不计入超时只有真正开始调用模型后才开始读超时计时。另外提一句如果Agent框架本身支持异步并发比如LangChain的ainvoke封装层一定要记得实现_agenerate否则并发请求会退化成串行QPS直接拉低一个数量级。4.3 可观测性日志、追踪、评测模型封装层是观测Agent系统的最佳位置。因为所有Agent的推理行为都必然经过模型调用在封装层埋点能拿到最完整的数据。我建议至少记录以下几类信息。请求维度的日志至少要包含请求ID、会话ID、调用的工具名、模型名、输入token数、输出token数、耗时、返回码、business_meta。有了这些字段就可以画出“某用户在一次Agent任务里模型被调用了多少次、每次花了多久、消耗了多少token”。追踪维度建议给每次模型调用生成一个span_id跟随Agent的trace_id一起透传下去。这样如果用户反馈一个问题你能还原出完整调用链用户输入-模型决策-工具执行-模型再次生成每一步的耗时和决策内容都在。评测维度封装层应该周期性保存“模型输入输出对”。这些数据积累到一定量后可以用来离线评测模型迭代版本甚至做回放测试。Agent系统的质量很大程度取决于模型输出质量没有数据积累后面优化模型时完全没有依据。5. 常见问题与排查技巧实录实践这一个项目我整理了一份问题速查表按故障现象分门别类。很多时候报错信息长得吓人其实根源就那么几个。故障现象可能原因排查路径Agent调用模型报ValidationError消息格式转换缺失比如ToolMessage没有tool_call_id先打印_convert_messages的输出比对API文档模型始终不调用工具纯文本回答bind_tools没有生效或schema不匹配检查封装层_bound_tools是否有值打印API收到的tools字段工具有调用但Agent无法解析参数模型返回的JSON格式不符合LangChain的tool_calls结构在_generate里手动解析并打印确认id/name/args三要素齐全每次请求都很慢约等于超时没有复用连接或并发限流排队确认适配器有没有用Session限流窗口是否设置过大异步调用ainvoke反而更慢没有实现_agenerate走了同步阻塞实现_agenerate并复用同一个适配器模型返回中文被截断max_tokens太小预留不足调大max_tokens或改为按字符数估算token每次任务最后一个AI回复丢失消息历史转换时漏掉了AIMessage的tool_calls检查_convert_messages中AIMessage处理分支监控图表里“模型调用次数”远多于预期没有做重试收敛Agent循环中反复调用失败重试设置最大重试次数并在重试之间加退避时间5.1 最容易翻车的“工具调用ID丢失”这个坑我印象最深。Agent第一次调用工具后模型拿到了ToolMessage工具执行结果需要根据之前的工具调用ID关联上下文。如果封装层在转换历史消息时没有把AIMessage里的tool_calls字段传给下一次请求模型就无法理解“这个工具结果对应哪个调用”它可能会编造一个新的调用Agent就会陷入“猜测调用”的状态——看起来在跑但实际上在做无意义的重复劳动。解决办法就是我在_convert_messages里展示的AIMessage的消息体里必须保留tool_calls字段并且把tool_call_id和ToolMessage的tool_call_id保持一致。一旦两边ID对不上模型也会直接报格式错误。排查这问题时打印请求中的消息历史是最直接的手段。5.2 模型返回合法但Agent不执行工具另一种常见病模型明显返回了工具调用意图但Agent只是“看着模型说话”没有真正执行工具。原因往往在封装层没有把finish_reason传给框架。Agent内部会检查模型的finish_reason是否等于tool_calls以此决定是否解析工具调用。如果你在_generate里只提取了content而忽略了响应里的finish_reasonAgent就不知道这次生成是“想要调用工具”还是“正常回答”。修复方式并不复杂在_generate里读取响应choices[0][finish_reason]如果是tool_calls确保把这个信息附加到AIMessage的响应元信息中。具体到LangChain可以通过additional_kwargs里塞一个finish_reason字段。Agent的大部分实现会检查这个标志位看完成原因是“停止”还是“工具调用”。如果框架版本较老可能还需要在AIMessage的tool_calls非空时手动断言它是工具调用意图。5.3 重试机制的副作用重复执行工具封装层做了重试之后一定要警惕一个副作用——重试导致工具被重复执行。比如Agent调用模型生成工具参数模型返回超时了封装层自动重试一次第二次成功了。但问题在于如果第一次请求其实已经到达模型只是响应超时那么模型可能已经生成过一组工具调用重试后模型又生成了一组。如果Agent同时处理这两次结果同一个工具就被执行了两次。这个问题的根治思路是重试只能发生在请求发出之前而绝不能发生在响应超时之后自动重试。如果你必须要重试请把“生成工具参数”和“执行工具”拆开——生成阶段即使失败也不会有副作用执行阶段依靠业务幂等性来保证不重复。一个加分做法是给每个工具调用加一个request_id在幂等判断中做去重。5.4 封装层内存泄漏排查自定义封装如果没注意生命周期长时间运行后可能会内存暴涨。最常见的原因有两个一是requests.Session没关连接没有释放二是限流器、回调注册表里积压了太多历史数据。特别是你在封装层里保存“最近N条请求日志”这种操作如果N设成无限大跑一个晚上就把内存吃光了。最简单有效的办法不要自己做日志列表一切日志直接走logging模块或推到外部日志系统Session由with管理或显式关闭限流器的统计窗口每个周期清理一次。做完这三件事封装层基本就是无状态的内存问题基本可以杜绝。我这次项目落地之后整体收益非常直接新增一个模型接入的时间从原来的一个下午缩短到半小时只需要写一个适配器其余逻辑全部复用线上问题排查从“翻遍Agent代码”变成“看封装层日志”效率翻倍。这套方案的核心不是代码本身而是对“模型差异”的隔离思维——模型会变、框架会升级但封装层的契约是稳定的所有下游都依赖契约而不是依赖具体实现。最后分享一个我个人的使用习惯封装层建好后先写一个“模型直连测试脚本”和“Agent完整链路测试”每个模型接入前先跑这两关。直连测试只验证协议和格式每天定时跑确保模型API升级没破坏兼容性链路测试则模拟真实Agent场景跑两三个典型任务确认工具调用、消息历史、脱敏过滤都正常。这两份脚本目前还在持续跑着成了我维护这套系统的基线保障。如果你也在做Agent开发强烈建议把这两步固化到你的开发流程里能省掉后面大量排查问题的精力。