★ FEATURED ARTICLE 1. 为什么我要自己封装一个 ChatModel先说结论如果你打算把 Agent 真正落到业务里早晚会走到“自定义模型封装”这一步。原因很简单市面上的模型太多了接口协议、鉴权方式、流式返回格式、工具调用Tool Calling的字段结构各家都不一样。LangChain 虽然内置了一大批集成包但总有覆盖不到的场景——比如公司内部私有化部署的推理服务、某个刚发布还没进官方集成列表的新模型、又或者你想在模型调用链路上加一层自己的审计和限流逻辑。我这次的项目背景是这样的手上有一个基于 LangChain 搭建的 Agent 应用原本用的是官方集成好的某个云端模型跑得挺顺。后来业务方要求把推理切换到自研的模型服务上这个服务对外暴露的是标准的 HTTP 接口返回结构跟 OpenAI 的格式有几分相似但又不完全一样尤其是流式输出的分片格式和工具调用的字段命名差异不小。最省事的做法当然是写个适配层把响应转成 OpenAI 格式然后继续用ChatOpenAI硬套。但我试过之后发现两个问题一是转换层越写越厚二是ChatOpenAI内部有些默认行为比如对tool_choice的处理、对finish_reason的映射跟我们的服务对不上改起来很别扭。所以最终我选择了一条更彻底的路直接继承 LangChain 的BaseChatModel从零封装一个自定义的 ChatModel。这篇文章就把整个封装过程、踩过的坑、以及一些设计上的取舍完整地讲一遍。适合已经用过 LangChain、想深入定制模型层的同学也适合刚开始接触 Agent 开发、想搞清楚“模型封装到底在封装什么”的朋友。哪怕你用的是别的框架底层的思路也是通用的。2. 封装之前必须搞清楚的几个核心概念2.1 BaseChatModel 到底规定了什么LangChain 里所有聊天模型都继承自BaseChatModel它本身又继承自BaseLanguageModel。这个抽象基类定义了一套契约你只要把这几个方法实现好你的模型就能无缝接入 LangChain 的整个生态——包括 Chain、Agent、回调系统、缓存、流式输出等等。核心要实现的其实就两个东西_generate同步生成接收一批消息返回一个ChatResult。_stream可选但强烈建议流式生成逐块 yieldChatGenerationChunk。另外还有几个属性需要声明_llm_type一个字符串标识用于日志和调试。_identifying_params返回一个字典描述这个模型的配置用于缓存 key 的生成。这里有个容易被忽略的点BaseChatModel已经帮你实现了generate、invoke、stream、batch这些对外方法它们内部会做参数校验、回调触发、缓存查询等一堆杂事然后才调用你实现的_generate或_stream。所以你的活儿其实很聚焦——只管把消息发出去、把结果拿回来、组装成 LangChain 认识的格式。2.2 消息格式与转换的坑LangChain 的消息体系是BaseMessage的子类SystemMessage、HumanMessage、AIMessage、ToolMessage等等。你的模型服务大概率不认这套对象需要转成它认识的 JSON。反过来服务返回的结果也要转回AIMessage。这里最麻烦的是工具调用。LangChain 里工具调用挂在AIMessage.tool_calls上结构是{name: ..., args: {...}, id: ...}。而不同模型服务返回的字段可能是function_call、可能是tool_calls数组、可能参数是字符串需要二次解析。我在项目里就遇到过服务返回的arguments是一个 JSON 字符串而不是对象如果不做兼容处理Agent 在执行工具时会直接报参数解析错误。还有一个坑是多模态内容。如果你的消息里包含图片HumanMessage.content可能是一个列表而不是字符串。转换的时候要判断类型否则会拼出乱七八糟的东西。2.3 流式输出为什么必须实现很多人封装模型时偷懒只实现_generate流式靠框架兜底。这样能用但体验差很多。Agent 场景下用户对首字延迟非常敏感流式输出能让用户感觉“模型在思考”而不是干等十几秒。实现_stream的关键在于你要把服务返回的 SSE 分片逐个转成ChatGenerationChunk。每个 chunk 里包含一个AIMessageChunk它的content是这一片的增量文本。如果服务在流式过程中还会返回工具调用的增量信息比如先返回函数名再返回参数片段那你还要处理tool_call_chunks的拼接。这块逻辑不复杂但很琐碎后面我会给出具体代码。3. 动手封装从零实现一个可用的 ChatModel3.1 环境准备与依赖清单我用的环境是 Python 3.10LangChain 版本是 0.2.x。这里提醒一句LangChain 的版本迭代非常快0.1 和 0.2 在部分接口上有 breaking change封装前最好先确认你项目里的版本。pip install langchain-core0.2.0 pip install httpx0.27.0 pip install pydantic2.0我选httpx而不是requests原因是它原生支持异步而且流式读取的 API 更顺手。虽然这次只实现同步版本但留着以后扩展异步的余地。3.2 定义模型类与初始化参数先定义类的骨架。我用 Pydantic 的Field来声明配置项这样能自动获得校验和文档。from typing import Any, Dict, Iterator, List, Optional from langchain_core.callbacks import CallbackManagerForLLMRun from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import AIMessage, AIMessageChunk, BaseMessage from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult from pydantic import Field import httpx import json class CustomChatModel(BaseChatModel): model_name: str Field(aliasmodel) api_base: str Field(defaulthttp://localhost:8000/v1) api_key: str Field(default) temperature: float 0.7 max_tokens: int 2048 timeout: int 60 property def _llm_type(self) - str: return custom-chat-model property def _identifying_params(self) - Dict[str, Any]: return { model_name: self.model_name, api_base: self.api_base, temperature: self.temperature, max_tokens: self.max_tokens, }这里有个细节model_name我用了aliasmodel这样初始化时既可以写CustomChatModel(modelxxx)也可以写CustomChatModel(model_namexxx)兼容性更好。_identifying_params一定要把影响输出的参数都放进去否则 LangChain 的缓存会串味——两个不同 temperature 的调用可能命中同一个缓存那就出大问题了。3.3 消息转换把 LangChain 消息转成服务认识的格式这是整个封装里最需要细心的一环。我写了一个_convert_messages方法把List[BaseMessage]转成服务需要的字典列表。def _convert_messages(self, messages: List[BaseMessage]) - List[Dict[str, Any]]: result [] for msg in messages: role self._map_role(msg) content msg.content # 处理多模态内容 if isinstance(content, list): content self._convert_multimodal(content) item {role: role, content: content} # 处理工具调用 if isinstance(msg, AIMessage) and msg.tool_calls: item[tool_calls] [ { id: tc[id], type: function, function: { name: tc[name], arguments: json.dumps(tc[args], ensure_asciiFalse), }, } for tc in msg.tool_calls ] # 处理工具返回 if msg.type tool: item[tool_call_id] msg.tool_call_id result.append(item) return result def _map_role(self, msg: BaseMessage) - str: mapping { system: system, human: user, ai: assistant, tool: tool, } return mapping.get(msg.type, user)注意arguments这里我做了json.dumps因为大多数服务端期望的是字符串而不是对象。如果你的服务期望对象那就去掉这层序列化。这个细节一定要对着服务文档确认我当初就是没注意导致工具调用一直失败排查了半天。3.4 实现 _generate同步调用的完整链路_generate是核心方法它要完成构造请求体、发 HTTP 请求、解析响应、组装ChatResult。def _generate( self, messages: List[BaseMessage], stop: Optional[List[str]] None, run_manager: Optional[CallbackManagerForLLMRun] None, **kwargs: Any, ) - ChatResult: payload { model: self.model_name, messages: self._convert_messages(messages), temperature: self.temperature, max_tokens: self.max_tokens, } if stop: payload[stop] stop # 处理工具绑定 if tools in kwargs: payload[tools] kwargs[tools] if tool_choice in kwargs: payload[tool_choice] kwargs[tool_choice] headers { Content-Type: application/json, Authorization: fBearer {self.api_key}, } with httpx.Client(timeoutself.timeout) as client: resp client.post( f{self.api_base}/chat/completions, jsonpayload, headersheaders, ) resp.raise_for_status() data resp.json() return self._parse_response(data)_parse_response负责把服务返回的 JSON 转成ChatResultdef _parse_response(self, data: Dict[str, Any]) - ChatResult: choice data[choices][0] message choice[message] content message.get(content) or tool_calls [] if message.get(tool_calls): for tc in message[tool_calls]: args tc[function][arguments] if isinstance(args, str): try: args json.loads(args) except json.JSONDecodeError: args {_raw: args} tool_calls.append({ name: tc[function][name], args: args, id: tc.get(id, ), }) ai_message AIMessage(contentcontent, tool_callstool_calls) generation ChatGeneration(messageai_message) return ChatResult(generations[generation])这里我特意对arguments做了容错如果解析失败就包一层_raw至少不让整个流程崩掉。实际生产中你可能会想记录这种异常方便后续排查。3.5 实现 _stream让首字延迟降下来流式实现稍微复杂一点核心是逐行读取 SSE 数据解析出增量内容。def _stream( self, messages: List[BaseMessage], stop: Optional[List[str]] None, run_manager: Optional[CallbackManagerForLLMRun] None, **kwargs: Any, ) - Iterator[ChatGenerationChunk]: payload { model: self.model_name, messages: self._convert_messages(messages), temperature: self.temperature, max_tokens: self.max_tokens, stream: True, } if stop: payload[stop] stop if tools in kwargs: payload[tools] kwargs[tools] headers { Content-Type: application/json, Authorization: fBearer {self.api_key}, } with httpx.Client(timeoutself.timeout) as client: with client.stream( POST, f{self.api_base}/chat/completions, jsonpayload, headersheaders, ) as resp: resp.raise_for_status() for line in resp.iter_lines(): if not line or not line.startswith(data:): continue data_str line[len(data:):].strip() if data_str [DONE]: break chunk_data json.loads(data_str) chunk self._parse_chunk(chunk_data) if chunk is not None: if run_manager: run_manager.on_llm_new_token(chunk.text) yield chunk_parse_chunk负责把单个分片转成ChatGenerationChunkdef _parse_chunk(self, data: Dict[str, Any]) - Optional[ChatGenerationChunk]: choices data.get(choices) if not choices: return None delta choices[0].get(delta, {}) content delta.get(content) or tool_call_chunks [] if delta.get(tool_calls): for tc in delta[tool_calls]: tool_call_chunks.append({ name: tc.get(function, {}).get(name), args: tc.get(function, {}).get(arguments, ), id: tc.get(id), index: tc.get(index, 0), }) message_chunk AIMessageChunk( contentcontent, tool_call_chunkstool_call_chunks, ) return ChatGenerationChunk(messagemessage_chunk)工具调用的流式拼接是个难点。服务通常会把一个工具调用拆成多个分片第一个分片带name和id后续分片只带arguments的片段。LangChain 的AIMessageChunk支持通过运算符合并框架会自动帮你把tool_call_chunks拼成完整的tool_calls。所以你在_parse_chunk里只要如实传递每个分片就行不用自己拼接。4. 工具绑定与 Agent 接入的实战细节4.1 bind_tools 的正确实现方式Agent 要用工具模型必须支持工具绑定。LangChain 的BaseChatModel提供了一个默认的bind_tools实现但它依赖模型本身支持tools参数。如果你直接继承默认行为是把工具 schema 塞进kwargs然后传给_generate。我在_generate里已经处理了kwargs[tools]所以理论上能跑通。但更稳妥的做法是重写bind_tools把工具 schema 的转换逻辑显式写出来def bind_tools(self, tools, **kwargs): from langchain_core.utils.function_calling import convert_to_openai_tool formatted [convert_to_openai_tool(t) for t in tools] return self.bind(toolsformatted, **kwargs)convert_to_openai_tool是 LangChain 提供的工具能把各种形式的工具函数、Pydantic 模型、字典统一转成 OpenAI 风格的 schema。这样你的模型服务只要认这套 schema就能直接对接。4.2 接入 Agent 时的一个隐蔽问题把自定义模型塞进 Agent 后我遇到一个很隐蔽的问题Agent 在执行完工具后会把ToolMessage追加到消息列表里再次调用模型。这时候消息列表里同时有AIMessage带 tool_calls和ToolMessage。如果我的_convert_messages没有正确处理tool_call_id的对应关系服务端会报错说“找不到对应的工具调用”。解决办法就是前面代码里那行item[tool_call_id] msg.tool_call_id。同时要确保AIMessage的tool_calls里的id和ToolMessage的tool_call_id是一致的。LangChain 内部会维护这个一致性你只要如实转换就行。4.3 参数校验与错误处理生产环境里模型服务不可能永远稳定。我在封装时加了几个防护超时控制httpx.Client(timeoutself.timeout)默认 60 秒可通过初始化参数调整。HTTP 错误resp.raise_for_status()会抛出异常LangChain 的回调系统会捕获并记录。响应格式异常_parse_response里对choices为空、message缺失等情况做了防御性判断避免直接KeyError。如果你想让错误更友好可以自定义异常类并在_generate里捕获后重新抛出附带更多上下文信息。5. 常见问题排查与避坑经验5.1 流式输出中断或乱码我遇到过一次流式输出到一半就断了日志里没有任何报错。排查后发现是服务端的 SSE 分片里有些行不是以data:开头而是空行或者注释行。我的解析逻辑里if not line or not line.startswith(data:)已经过滤了这些但问题出在iter_lines()的默认行为上——它会把\r\n也当作换行导致某些分片被截断。解决办法是显式指定iter_lines()的解码方式或者在解析前先strip()。另一个常见问题是中文乱码。SSE 传输的是 UTF-8 字节流httpx默认会按响应头的 charset 解码。如果服务端没正确设置Content-Type: text/event-stream; charsetutf-8就可能出现乱码。这时候可以手动指定编码resp.iter_lines(decode_unicodeTrue)配合resp.encoding utf-8。5.2 工具调用参数解析失败前面提过arguments可能是字符串也可能是对象。我建议统一按字符串处理然后尝试json.loads失败就包一层。另外要注意有些服务返回的arguments是空字符串这时候json.loads会报错需要特殊处理成空字典。还有一个坑是tool_calls的id字段。OpenAI 的格式里每个工具调用都有唯一id但有些自研服务不返回这个字段。如果缺失LangChain 在后续匹配ToolMessage时会出问题。我的做法是如果服务没返回id就用name index生成一个稳定的标识。5.3 缓存导致的“灵异现象”LangChain 默认会对模型调用做缓存如果配置了 cache。我遇到过一次改了 temperature 后输出还是老样子。排查半天才发现是_identifying_params里没把 temperature 放进去导致缓存 key 没变。所以再次强调所有影响输出的参数都必须放进_identifying_params。5.4 常见问题速查表问题现象可能原因排查方向工具调用报参数错误arguments 是字符串未解析检查_parse_response里的 json.loads流式输出中断SSE 分片格式不标准打印原始 line检查过滤逻辑中文乱码编码未指定手动设置 resp.encoding缓存不更新_identifying_params缺参数补全所有影响输出的字段Agent 找不到工具调用tool_call_id 不匹配检查 AIMessage 和 ToolMessage 的 id 对应首字延迟高未实现_stream确认_stream被正确调用6. 一些设计上的取舍与个人体会封装完这个模型后我最大的感受是不要为了封装而封装。如果你的模型服务跟 OpenAI 格式高度兼容直接用ChatOpenAI加个base_url是最省事的。只有当差异大到适配层变得臃肿时才值得自己继承BaseChatModel。另一个体会是流式实现越早做越好。我一开始只做了同步版本后来补流式的时候发现消息转换、工具调用拼接这些逻辑都要重新过一遍返工成本不低。如果一开始就把_stream和_generate一起写很多公共逻辑可以抽出来复用。还有一点关于工具调用的流式拼接LangChain 的AIMessageChunk合并机制很好用但前提是你传的tool_call_chunks结构要对。index字段很关键它决定了不同工具调用的分片能不能正确归组。如果服务不返回index你要自己按顺序生成。最后说个实际效果封装完成后Agent 的首字延迟从原来的 3 秒多降到了 800 毫秒左右工具调用的成功率也从之前的 70% 提升到了接近 100%之前失败主要是因为参数解析问题。这个投入是值得的。如果你也在做类似的封装建议先把消息转换和响应解析这两块写扎实再考虑流式和工具调用。这两块是地基地基不稳上面盖什么都会晃。
阅读完成 · 觉得有帮助？
