Agent开发自定义模型封装实战：从接口适配到流式与工具调用

做 Agent 开发久了你会发现，真正决定上层框架能不能落地跑通的，往往是模型这一层。框架自带的 ChatOpenAI、ChatAnthropic 这些现成类，只是包装好了官方 API；生产环境里我们更多时候面对的是自研模型、私有化部署的推理服务、或者某个兼容内部鉴权的模型网关。这时候，“自定义模型封装”就成了 Agent 项目里绕不开的一步。说白了，就是把自己的模型服务接入 Agent 框架的标准接口，让它能像官方模型一样被调度、被流式输出、被工具调用。

这篇文章我打算用一个完整的实战例子，把这套封装该怎么做、要处理哪些坑、以及为什么很多细节必须在一开始就决定好，一次讲清楚。内容会以 LangChain / LangGraph 生态为主，因为目前国内团队做 Agent 开发，这个框架的占比确实最高。如果你用的是 Spring AI、AutoGen 或者自研编排层，核心思路一样，只是接口名字换一换。适合正在做 Agent、但卡在模型接入环节的同学参考。

1. 先搞清楚：到底什么是“自定义模型封装”

1.1 封装在 Agent 链路里的位置

一个 Agent 系统拆到最粗，可以分成四层：编排层、工具层、记忆层、模型层。编排层决定什么时候调用工具、什么时候反问用户；工具层负责跟外部系统交互；记忆层管多轮对话的上下文；模型层则是所有这些决策的“大脑”。

问题在于，编排层和工具层都活在框架的世界里，它们对模型只有一个要求：你实现我定义的接口，我就用你。而真实世界的模型服务千奇百怪，有的走 OpenAI 兼容协议，有的走私有 RPC，有的只是一个本地进程，拉起来还得等三五秒。自定义模型封装干的活，就是在模型服务和框架之间写一个适配器，把框架的统一调用翻译成你这个模型服务听得懂的请求，再把它的响应翻译回框架认识的结构。

我见过不少同学一开始图省事，直接在 AIAgent 类的回调里写死 HTTP 请求，看起来也能跑。但等到要换模型、要加流式、要支持工具调用、要压并发的时候，代码就成一坨浆糊。封装不是多写一层，是把“模型服务长什么样”和“Agent 逻辑怎么跑”这两件事彻底隔离。隔离做好了，以后换模型只是换一个封装类的事，Agent 上层一行不改。

1.2 封装的本质：接口适配与能力对齐

很多人理解封装，以为就是把 HTTP 请求包一层函数。其实封装的核心是两件事：接口适配和能力对齐。

接口适配好理解，框架调你的时候传的是 BaseMessage 列表加 generation config，你把它翻译成模型服务要的 JSON 结构。响应回来的时候，再把模型吐的字儿转成 AIMessage，这属于“翻译”。

能力对齐就微妙了。框架默认模型有三项能力：流式输出、工具调用、多轮记忆。你的模型服务不一定全都支持。比如自研模型没做 function calling，只支持裸文本；再比如有些内部模型服务支持流式，但中间会把多段内容粘在一起再吐出来，流式形同虚设。封装层的职责就是把这些能力差异“抹平”：模型不支持工具调用，你在封装里用提示词模板模拟一个；模型流式不稳定，你在封装里做缓冲和超时兜底。

这个角色有点像充电器上的电源转换芯片，不只是一个转接头，还是个带稳压、限流、短路保护的转换器。你在封装层多花的心思，决定了 Agent 上层应用是稳如老狗，还是时不时出各种玄学问题。

2. 动手前的准备：框架选型与接口分析

2.1 主流 Agent 框架里的模型接口长什么样

先看清楚你用的框架到底要求你实现哪些方法。我在 LangChain 生态里做封装，最常打交道的是 BaseChatModel 这个抽象类。它的核心方法就几个：

• _generate：接收消息列表和配置，返回 ChatResult，这是非流式入口；
• _stream：返回一个迭代器，逐 chunk 吐数据，这是流式入口；
• _agenerate和_astream：上面的异步版本，生产环境基本用这个；
• bind_tools：把工具定义绑定到模型上，让模型具备 tool calling 能力。

Spring AI 那边类似，核心是 ChatModel 接口，重点实现call和stream两个方法，工具调用则通过ChatOptions传工具列表。AutoGen 走的是 ModelClient 的create方法。不管框架怎么包装，本质都是“输入一组消息，输出一个新消息，可选地带上工具调用指令”。

所以你在开始写代码之前，第一件事是找到框架文档里那个“自定义模型”或者“自定义 LLM”的页面，把抽象类的方法签名抄下来对着看。很多报错的根源，不是你的模型服务有问题，而是你少实现了某个方法，框架在特定场景下调用时就崩了。

2.2 先盘点你手头模型的“家底”

磨刀不误砍柴工。封装之前，你最好先拿命令行或者 Postman，把你模型服务的接口文档翻一遍，填一张“家底表”。我每次做新封装，第一件事就是拉一个表格记录下面这些字段，避免写到一半才发现某个参数不支持。

我见过最典型的翻车场景：一个团队从 OpenAI 切到自研模型，以为协议兼容就能无缝切换，结果自研模型压根不认system角色，所有系统提示词全被忽略。这种问题你在盘点阶段就会发现，而不是等上线之后业务方来找你。

3. 核心环节拆解：流式输出、工具调用与消息协议

3.1 流式输出为什么是 Agent 的刚需

很多刚接触封装的同学会觉得，流式是加分项，不是必选项。真不是。Agent 场景里，模型经常要生成一大段代码、一个长分析，如果走非流式，用户要盯着一个转圈图标等十几秒甚至一分钟，体验直接崩。流式还能让用户看到“模型正在做什么”，尤其在多步 Agent 任务里，前一秒在思考，后一秒在调工具，这种透明度能极大减少焦虑感。

流式的实现原理不复杂。你的模型服务如果走 HTTP，一般是用 SSE 或者 chunked 传输，一行一行吐 token；如果是自研 RPC，通常也有回调接口。封装层要做的事，就是把这些不同形式的 token 流，统一转换成框架认识的ChatGenerationChunk对象，然后 yield 给上层。

这里有个细节：框架的流式迭代器一旦开始，中途如果出错，要么抛异常终止整个生成，要么把异常吞掉返回不完整内容。我在封装里默认的做法是，流式过程中如果收到模型服务的错误码，先把缓冲池里已经收到的内容吐完，再抛一个带上下文的异常。这样上层至少能拿到部分结果，日志里也能看到是哪个环节断了。

3.2 工具调用的封装策略

工具调用是整个封装里最“坑”的部分，因为模型服务的能力参差太大。如果你的模型服务原生支持 function calling，比如走 OpenAI 兼容协议且支持tools参数，那封装相对简单，把框架的工具 schema 转成你的协议格式就行。

麻烦的是模型服务不支持原生 function calling。这种情况下，业界最常见的做法是用提示词模拟：在 system 提示词里把可用工具的描述、JSON 格式要求写清楚，要求模型在需要调用工具时输出一段特定格式的 JSON。然后在封装层加一个轻量的输出解析器，把模型吐出来的文本里的 JSON 抽出来，转成一个带有tool_calls字段的 AIMessage。框架看到这个消息，就知道要去执行工具了。

这个方案有几个关键坑。第一，模型吐的 JSON 经常带多余的前缀或代码块标记，比如喜欢包在json里，解析时要去掉。第二，长文本生成时 JSON 可能被截断，需要做修复，我习惯把这种情况降级为普通文本回复，而不是让整个 Agent 崩溃。第三，提示词里工具描述不能太长，否则模型会忽略部分工具，我一般控制在“工具名 + 一句话用途 + 参数 JSON Schema 精简版”的格式。

3.3 消息协议与多轮对话状态管理

Agent 的多轮对话，本质上就是消息列表的不断追加。封装层要把框架的 BaseMessage 列表翻译成模型服务的消息结构。这里最容易踩坑的是角色映射。

框架里有 system、human、ai、tool 四种角色，模型服务不一定全部支持。很多自研模型只认识 user 和 assistant，甚至 system 也要特殊处理。我的习惯做法是：模型不支持 system 角色时，把 system 内容拼到第一条 user 消息的开头，用特殊分隔符标一下，实测下来比丢给别的字段更稳。

另外，工具调用结果的角色映射在 LangChain 里是 ToolMessage。有些模型服务不认识这个角色，你就得把工具结果转成 user 或 assistant 消息，同时保留对应的工具调用 ID。多轮 Agent 里，工具调用和工具结果必须成对出现，顺序错一个，模型就会“失忆”，上下文里出现对话历史和工具结果纠缠不清的问题。封装的时候，务必在构建请求体时做一遍消息顺序校验。

4. 实操：一个自定义模型封装的完整实现

4.1 设计封装类与目录结构

下面我给出一个基于 LangChain 的完整示例。先说设计思路：我倾向于用组合而不是继承大改，也就是封装类内部持有我们自己的 HttpClient，对外只实现 BaseChatModel 的抽象方法。这样模型服务的 SDK 升级、协议调整，只影响内部逻辑，不改变类的外部接口。

目录结构大概长这样：

my_agent_project/ ├── models/ │ ├── __init__.py │ ├── custom_chat_model.py # 自定义模型封装类 │ └── client.py # 内部模型服务的 HTTP 客户端 ├── tools/ │ └── ... ├── graph.py # Agent 编排逻辑 └── config.py

4.2 核心代码：请求、响应与流式适配

直接上代码。下面这个封装类，对接一个假设的内部模型服务，协议走 OpenAI 兼容的 /chat/completions，但鉴权方式是我们内部私有签名。

from typing import Any, AsyncIterator, Dict, Iterator, List, Optional import httpx from langchain_core.callbacks import CallbackManagerForLLMRun from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import ( AIMessage, AIMessageChunk, BaseMessage, HumanMessage, SystemMessage, ToolMessage, ) from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult from langchain_core.utils.function_calling import convert_to_openai_tool class CustomChatModel(BaseChatModel): """对接内部模型服务的自定义模型封装。""" model_name: str = "internal-llm-v1" api_base: str = "http://internal-model-service.example" temperature: float = 0.7 max_tokens: int = 1024 timeout: float = 30.0 def _get_endpoint(self) -> str: return f"{self.api_base}/chat/completions" def _build_payload( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, **kwargs: Any, ) -> Dict[str, Any]: # 消息角色映射：LangChain 消息 -> 模型服务协议 mapped = [] for msg in messages: if isinstance(msg, SystemMessage): mapped.append({"role": "system", "content": msg.content}) elif isinstance(msg, HumanMessage): mapped.append({"role": "user", "content": msg.content}) elif isinstance(msg, AIMessage): if msg.tool_calls: mapped.append({ "role": "assistant", "content": msg.content or "", "tool_calls": [ { "id": tc["id"], "type": "function", "function": { "name": tc["name"], "arguments": tc["args"], }, } for tc in msg.tool_calls ], }) else: mapped.append({"role": "assistant", "content": msg.content}) elif isinstance(msg, ToolMessage): mapped.append({ "role": "tool", "content": msg.content, "tool_call_id": msg.tool_call_id, }) payload = { "model": self.model_name, "messages": mapped, "temperature": self.temperature, "max_tokens": self.max_tokens, "stream": False, } if stop: payload["stop"] = stop payload.update(kwargs) return payload def _generate( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> ChatResult: payload = self._build_payload(messages, stop, **kwargs) payload["stream"] = False # 内部模型服务使用私有签名鉴权 headers = self._build_signature_headers(payload) with httpx.Client(timeout=self.timeout) as client: resp = client.post(self._get_endpoint(), json=payload, headers=headers) resp.raise_for_status() data = resp.json() content = data["choices"][0]["message"]["content"] tool_calls = data["choices"][0]["message"].get("tool_calls") ai_message = AIMessage(content=content) if tool_calls: ai_message.tool_calls = [ { "id": tc["id"], "name": tc["function"]["name"], "args": tc["function"]["arguments"], } for tc in tool_calls ] return ChatResult(generations=[ChatGeneration(message=ai_message)]) def _stream( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> Iterator[ChatGenerationChunk]: payload = self._build_payload(messages, stop, **kwargs) payload["stream"] = True headers = self._build_signature_headers(payload) with httpx.Client(timeout=self.timeout) as client: with client.stream( "POST", self._get_endpoint(), json=payload, headers=headers ) as resp: resp.raise_for_status() for line in resp.iter_lines(): if not line.startswith("data: "): continue data_str = line[len("data: "):] if data_str == "[DONE]": break # 这里可以补 json 解析异常兜底 chunk_data = json.loads(data_str) delta = chunk_data["choices"][0]["delta"] if "content" in delta: content = delta["content"] if content: chunk = ChatGenerationChunk( message=AIMessageChunk(content=content) ) if run_manager: run_manager.on_llm_new_token(content) yield chunk async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs): # 异步版本，生产环境建议用 httpx.AsyncClient async with httpx.AsyncClient(timeout=self.timeout) as client: payload = self._build_payload(messages, stop, **kwargs) payload["stream"] = False resp = await client.post( self._get_endpoint(), json=payload, headers=self._build_signature_headers(payload), ) resp.raise_for_status() data = resp.json() content = data["choices"][0]["message"]["content"] ai_message = AIMessage(content=content) return ChatResult(generations=[ChatGeneration(message=ai_message)]) @property def _llm_type(self) -> str: return "custom-internal-chat-model" def bind_tools(self, tools: List[Any], **kwargs: Any) -> "CustomChatModel": """把工具列表绑定到模型实例，供 Agent 编排层调用。""" # 这里用框架的转换函数把 Pydantic 工具转成 OpenAI 风格 schema openai_tools = [convert_to_openai_tool(tool) for tool in tools] copied = self.copy() copied.tools = openai_tools # type: ignore[attr-defined] return copied

代码里几个地方要展开说明。

消息映射我把它单独抽成_build_payload，因为你之后大概率还要支持多模态消息，到时候只改这一个方法就行。工具调用的参数arguments在 OpenAI 协议里是一个 JSON 字符串，LangChain 里tool_calls的args是字典，映射的时候要json.loads转换，代码里这个细节最容易漏。

bind_tools的实现用了 copy，而不是原地修改。原因是 LangChain 框架经常会在回调、重试时复用同一个模型实例，原地改会污染配置，copy 一份最安全。我在_generate里没有直接把self.tools塞进 payload，实际封装时你应该在_build_payload里检查是否绑定了工具，把tools字段加上。

4.3 参数映射、超时与重试策略

模型服务说自己支持 temperature，不代表它真的按官方语义执行。封装层不要无脑透传所有参数，而是做一个白名单映射。

超时策略我用的是“双超时”：建立连接 5 秒，整体读响应 30 秒。如果模型首 token 慢，整体 30 秒会被首 token 吃掉一大半，所以对交互型 Agent，我还会设置read_timeout=10，首包超过 10 秒直接报错让用户重试，避免一直转圈。

重试策略要区分错误类型。模型服务返回 429 说明限流，退避重试 3 次有效；返回 5xx 说明服务端故障，重试一次看是否恢复；返回 4xx 业务错误，比如参数非法、上下文超长，直接抛错给上层，不要重试，重试也只是浪费配额。我在这块加过一个指数退避，初始 0.5 秒，翻倍，最多 3 次，实测对瞬时抖动效果很好。

5. 实战接入与常见问题排查

5.1 模型返回格式不标准怎么办

自研模型最常见的问题就是输出格式飘忽不定。明明提示词要求返回 JSON，结果它给你包一层 markdown 代码块，或者 JSON 中间多一个逗号，或者在一个文本里混了多个 JSON 对象。

我在封装里放了一个兜底解析函数，放在_generate的响应处理前面。流程是：先尝试验证整个输出是合法 JSON；失败的话，用正则把代码块里的 JSON 提出来；再失败，尝试用修复库做单引号/尾逗号修复；最后才降级为纯文本 AIMessage。这个过程不能太激进，否则会把模型正常的自然语言回复误判成 JSON 然后解析失败，整个 Agent 就断了。

另外一个常见问题是，模型在工具调用场景下会同时返回文本和 tool_calls，内容字段里有“我来帮你查询订单”这种话。框架层对这种情况的处理不一致，有的版本要求 content 为空才能执行工具。我的习惯是保留 content，让上层能展示模型“正在说什么”，然后把 tool_calls 也塞进去，实测 LangGraph 的 ReAct 风格编排可以正常处理双返回值。

5.2 并发场景下的 Token 限制与排队

Agent 的负载模型和普通聊天不一样，一个 Agent 实例在单次任务里可能要调模型七八次，中间穿插多次工具调用。这种高频短请求对模型服务的并发压力非常大，尤其你的封装还是一个团队所有 Agent 共用的时候。

我在封装类里加了一个全局信号量，控制同时打向模型服务的请求数。信号量大小取决于模型服务那边压测过的并发上限，我一般从 4 开始压，逐步往上调，直到 P95 延迟开始明显变差，就停在那。

import asyncio # 全局并发门禁：假设模型服务最多接受 8 个并发 _SEMAPHORE = asyncio.Semaphore(8) async def _agenerate_with_guard(self, messages, stop=None, run_manager=None, **kwargs): async with _SEMAPHORE: return await self._agenerate(messages, stop, run_manager, **kwargs)

除了并发门禁，还要注意 token 配额。有些 Agent 的提示词很长，比如把检索到的文档塞进上下文，如果封装层不做长度检查，模型服务会在跑到一半时返回“context length exceeded”，用户看到的是一句生硬的报错。比较好的做法是把超长上下文截断阈值写进封装，提前在发送前计算估算 token 数，超过 85% 就触发压缩策略，比如只保留最近几轮对话加当前工具结果。

5.3 调试 Agent 链路的三板斧

自定义模型封装一旦出了问题，难点在于你分不清是模型的问题、封装的问题、还是编排层的问题。我调试这类链路有三板斧。

第一板斧，把封装层的输入输出全量打印成结构化日志。在_generate和_stream里记录消息数、总 token 估算、响应耗时、首 token 延迟、是否触发重试。这个日志是定位“慢在哪、错在哪”的第一手材料。

第二板斧，录流回放的测试模式。我会找一个稳定的模型服务请求，把请求 payload 和响应存成 fixture 文件，封装类里留一个_use_fixture开关。测试的时候不连真实服务，直接返回 fixture 内容，这样 Agent 编排逻辑可以完全脱离模型服务做单元测试。谁变了，跑一次测试立刻见分晓。

第三板斧，给每次 Agent 调用链路打 trace_id。封装层收到的每条消息、发出的每个请求，都带上同一个 trace_id 写入日志。多步 Agent 任务里，如果你发现第二次工具调用后的模型回复莫名其妙，拿 trace_id 一捞，所有环节的上下文一目了然。这个能力看着简单，但它的价值在线上排障时巨大。

6. 写在最后：封装这件事，别一步到位

做自定义模型封装，我最深的体会是：第一版千万别追求大而全。我首次做这个封装的时候，一开始只实现了非流式的_generate，图表跑通了，看似万事大吉。结果一挂到真实用户面前，立刻被吐槽“怎么一直转圈”“生成这么慢”。第二天补了流式，才感觉像一个能用的产品。所以建议你按这个顺序推进：先跑通非流式，再做流式，然后处理工具调用，最后才补并发和重试。

工具调用这块，我的建议是优先确认模型服务是否原生支持。原生支持时，封装就是一个参数翻译；不支持时，用提示词模拟方案也能跑，但你要做好心理准备，这个方案对提示词质量极其敏感，上线后会反复调。

另外一个经验：封装类里所有魔法数值，比如超时秒数、重试次数、信号量大小、截断阈值，都要抽到配置里，不要硬编码。我在生产环境里吃过亏，明明压测的时候信号量设为 8 很稳，上线后发现某条业务线峰值流量直接冲到 20 并发，服务被打满。后来把这些参数全部配置化，按业务维度独立调，才消停下来。

最后再分享一个小技巧：封装完成后，写一个“模型协议自检脚本”，每次模型服务方更新接口、或者你升级框架版本之后，跑一遍自检，确认角色映射、流式、工具调用三个核心能力仍然正常。这个脚本花半小时写，之后每一次升级都替你省下几个小时。自定义模型封装不会是你 Agent 项目里最炫的部分，但它的健壮程度，基本决定了整个系统的上限在哪。