★ FEATURED ARTICLE 在Agent项目里待过一段时间的人基本都会遇到同一个问题手头有一个非常合适的模型可能是自己微调过的也可能是某个闭源平台独占的又或者是公司内部部署的结果想把它接进LangChain、Dify或者CrewAI时却发现在这些框架里“没有现成入口”。这其实就是模型封装要解决的事。作为“Agent实践”系列的第二篇这篇文章专门聊聊自定义模型封装先讲清楚为什么封装、封装前要想明白什么再给出一套基于LangChain的完整实现顺手解决Dify和CrewAI的接入问题最后把我实际踩过的坑和排查思路摊开讲。这篇文章适合正在搭Agent、想要摆脱“只能用OpenAI”限制的人也适合那种自己部署了模型、想统一暴露成标准接口给多个框架用的开发者。内容偏工程向但我尽量把原理用大白话讲透代码可以直接抄去改。1. 把自定义模型塞进Agent框架为什么这么麻烦1.1 不是每个模型都长OpenAI的样子现在绝大多数Agent框架的核心循环都很相似用户输入进来Agent把任务拆成几步每步调用一次LLM根据返回结果决定下一步是调工具还是继续追问。问题在于框架内部已经预设了一套“模型应该长什么样”的接口。比如LangChain里所有对话模型都要返回ChatResultDify里所有LLM都要能返回某种固定的Response结构CrewAI的Agent则直接接受LangChain的LLM对象。而你手上的自定义模型很可能根本不认识这套接口。它可能是一个HTTP服务只接收{prompt: ..., max_tokens: 512}这种简单的JSON它可能是公司内部的RPC服务方法名都跟OpenAI完全不搭它甚至可能是通过自定义SDK访问的SDK里压根没有“消息列表”的概念只有一个文本输入框。这就是第一个麻烦框架的模型抽象是它自己的你的模型API是另一套中间必须有东西把两边翻译过来。1.2 封装真正解决的是“接缝”问题很多刚接触Agent的人会直接用Requests库在Agent的每个工具函数里去调模型这样确实能跑通但带来的问题很明显每处都得自己处理超时、重试、错误码模型换一个就得把调用代码全部改一遍。更麻烦的是Agent框架内置的很多能力会失效——比如LangChain的Runnable流式输出、Dify的调试面板、CrewAI的记忆机制这些功能全都要依赖框架自带的模型接口才能工作。我在实践中把模型封装理解成“接缝”处理框架那边有一个标准的插座模型这边有一个非标的插头封装层就是一个转接器。转接器做得好后面换模型就是换一个插头的事转接器做得稀烂整个Agent项目都会跟着模型绑死想换都换不下来。1.3 什么信号出现时你该动手封装了不是所有项目都需要专门写一个封装层。我自己的判断标准是出现下面这几种信号之一就值得动手你需要在两个以上不同的Agent框架里使用同一个自定义模型不想每个框架各写一遍适配代码。你的模型API不是OpenAI兼容格式而且你预计后续还要换模型。你希望Agent的流式输出、工具调用、token统计等能力不是靠手写散弹代码而是能被框架原生管理。你有一个自研模型服务之后打算开放给多个团队使用需要一个稳定的接入规范。2. 动手封装前先定好三件事2.1 抽象层选官方还是自研封装自定义模型前先确定“封装到哪里”。大多数情况下我不建议自己重新设计一套全新的LLM抽象接口除非你有特别强的理由——比如公司已经统一了内部模型服务协议所有模型都通过那个协议访问那你确实可以基于那个协议做一层高复用封装。但如果你只是想把一两个模型接入Agent框架直接复用框架自带的抽象是最省事的。拿LangChain举例它已经定义了BaseChatModel、BaseLanguageModel这些抽象类。你写的封装本质上就是去实现这些抽象类的方法然后把实现好的对象传给Agent。这样做的好处是LangChain生态里所有依赖模型对象的组件比如create_react_agent、AgentExecutor、各种memory、回调处理器都能直接识别你这个模型。不要自己去定义一个MyModelInterface然后再写一堆桥接代码那就绕了远路还容易漏掉框架里隐藏的调用契约。2.2 协议栈按OpenAI兼容走省下的都是时间可能有人会问我封装的模型本来就是自定义的为什么要关心OpenAI协议因为现在几乎所有Agent框架都默认支持配置一个OpenAI兼容的base_url和api_key。如果你能把自定义模型包装成一个OpenAI兼容的服务端那么理论上不需要任何Agent框架侧的深度适配只要配置好base_url就能用。所以一个很实用的决策是如果我们可以选择封装位置优先在服务端把模型暴露成OpenAI兼容API而不是去改框架代码。这样LangChain、Dify、CrewAI甚至各种开源客户端都能直接用覆盖面最大。如果模型本身没法跑一个服务端那就只能在客户端写一个适配类思路也相同只是实现位置从FastAPI变成BaseChatModel子类。我自己处理过的最小OpenAI兼容服务就是一个FastAPI应用暴露/v1/chat/completions把传入的消息数组转成内部模型需要的格式再返回指定结构的JSON。由于后续还要在LangChain里搞流式、搞工具调用服务端协议越接近OpenAI客户端封装就越简单。2.3 哪些状态该留在模型层哪些该交给Agent这是封装时容易被忽略的边界问题。有些封装者喜欢把历史消息、会话ID、用户ID全都塞进模型封装里结果会让封装变成“有状态”的用起来非常难受。因为Agent框架本身已经会用MessageHistory、Memory这些组件去管理对话上下文模型封装要做的仅仅是“来一批消息产生一条补全”。如果封装层自己偷偷存历史当Agent框架去做消息裁剪、摘要、多轮切换时两边就会打架。我的经验是封装层应该尽量无状态。像temperature、max_tokens、stop这些参数可以作为调用参数传入而会话ID、用户身份、历史记录这些信息不要进封装的内部状态让Agent层去控制就够了。如果确实有某些字段需要透传到模型服务可以通过kwargs里加额外参数的方式传进去但不要让封装层产生隐藏上下文。3. LangChain实战从零封装一个ChatModel3.1 准备依赖与最小骨架先说下当前LangChain环境。我用的版本是基于langchain-core的你可以理解成LangChain的新架构各种模型类都搬到了核心包里。需要安装的依赖大致是pip install langchain-core pip install openai # 如果你的自定义模型是OpenAI兼容协议我们客户端封装时可以直接借openai库来调用如果你调用的是纯HTTP接口也可以用httpx不过借用openai库有一个额外好处流式、重试、超时这些机制都已经处理好了我们不需要重新造轮子。下面的示例我会假设你的自定义模型服务暴露了一个OpenAI兼容的/v1/chat/completions接口因为这是最通用的场景。LangChain中所有对话模型都需要继承BaseChatModel并且至少实现_generate方法。最小骨架如下from typing import Any, List, Mapping, Optional from langchain_core.callbacks import CallbackManagerForLLMRun from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import AIMessage, BaseMessage from langchain_core.outputs import ChatGeneration, ChatResult class MyCustomChatModel(BaseChatModel): 把自定义模型封装成LangChain ChatModel。 base_url: str http://127.0.0.1:8000/v1 api_key: str local model_name: str my-model property def _llm_type(self) - str: return custom-chat-model def _generate( self, messages: List[BaseMessage], stop: Optional[List[str]] None, run_manager: Optional[CallbackManagerForLLMRun] None, **kwargs: Any, ) - ChatResult: raise NotImplementedError这一段里有几个点要认真对待。_llm_type是给LangChain做标识用的随便起个名字就行但不要和已有的模型类型冲突。stop参数是框架传过来的停止词列表必须透传给模型服务。run_manager是用来上报token使用情况和流式片段的初期可以先不用但流式那节会用上。3.2 实现_generate让模型先跑起来_generate的返回值必须是ChatResult。它的结构就是一个ChatGeneration列表通常只有一个元素每个ChatGeneration里有一个AIMessage。我们实际要做的事就是两件把messages转成OpenAI兼容的请求体把模型服务返回的文本包成AIMessage。def _generate( self, messages: List[BaseMessage], stop: Optional[List[str]] None, run_manager: Optional[CallbackManagerForLLMRun] None, **kwargs: Any, ) - ChatResult: from openai import OpenAI client OpenAI(base_urlself.base_url, api_keyself.api_key) openai_messages [] for msg in messages: if isinstance(msg.content, str): content msg.content else: content # 复杂的content list这里先简化处理 openai_messages.append({role: _convert_role(msg.type), content: content}) payload { model: self.model_name, messages: openai_messages, temperature: kwargs.get(temperature, 0.7), max_tokens: kwargs.get(max_tokens, 2048), } if stop: payload[stop] stop resp client.chat.completions.create(**payload) # 注如果自定义模型返回的字段不完整需要在这里做容错 content resp.choices[0].message.content or return ChatResult( generations[ ChatGeneration( messageAIMessage(contentcontent), ) ] )角色转换_convert_role很简单human-userai-assistantsystem-system其他类型统一转成user避免服务端报错。这一步看着小但很容易漏。假如你的模型服务对role字段有严格校验消息类型不认就会直接拒绝。还有一个容易被忽略的点OpenAI客户端的base_url末尾是否带/v1会影响请求路径。不同版本的openai库对base_url的拼接逻辑有差异我自己建议统一写成到/v1这一层比如http://127.0.0.1:8000/v1这样chat.completions.create会拼出/v1/chat/completions。3.3 实现流式输出别让用户体验倒退回打字机前Agent在回答过程中如果等全部生成完才返回用户的等待感会非常强。LangChain的BaseChatModel支持通过覆写_stream来提供流式能力但这个方法的签名很多人不清楚。新版本的BaseChatModel里_generate是核心_stream是可选的如果你实现了_streamstream()方法就会走你的实现。def _stream( self, messages: List[BaseMessage], stop: Optional[List[str]] None, run_manager: Optional[CallbackManagerForLLMRun] None, **kwargs: Any, ): from openai import OpenAI client OpenAI(base_urlself.base_url, api_keyself.api_key) openai_messages [_to_openai_message(msg) for msg in messages] payload { model: self.model_name, messages: openai_messages, stream: True, temperature: kwargs.get(temperature, 0.7), max_tokens: kwargs.get(max_tokens, 2048), } if stop: payload[stop] stop stream client.chat.completions.create(**payload) for chunk in stream: if not chunk.choices: continue delta chunk.choices[0].delta if delta and delta.content: content delta.content if run_manager: run_manager.on_llm_new_token(content) yield ChatGenerationChunk(messageAIMessageChunk(contentcontent))注意这里不能用ChatGeneration而是要用ChatGenerationChunk和AIMessageChunk因为_stream本身是生成器必须产出“块”对象这样外部调用model.stream(messages)时才能拿到一个迭代器。run_manager.on_llm_new_token是LangChain回调机制的一部分它是回调处理器感知流式输出的关键如果你不看这一句之后想在LangSmith或者自定义回调里统计流式token就会无从下手。从自己封装的角度_stream的返回类型很容易搞错。我一开始把ChatGeneration直接yield出去结果Agent在处理时拿到的消息对象缺少拆分能力后续消息拼接就乱了。一定要用Chunk那组类。3.4 支持function calling让模型真正会用工具Agent要调用工具模型侧必须支持function calling。LangChain里的做法是在创建模型实例后调用.bind_tools(tools)框架会把工具定义传给模型。但你的自定义模型服务不一定原生支持OpenAI的tools参数所以封装层要做一次格式转换。如果你的服务已经支持OpenAI兼容的tools那事情很简单直接在_generate里取kwargs.get(tools)透传给服务就行。但如果你的自定义模型不支持结构化工具调用只能靠提示词来让模型输出JSON那封装层就需要把工具的JSON Schema转换成一个工具说明文本塞进system消息里。我这里给出一个同时支持两种方式的思路def _extract_tools_from_kwargs(self, kwargs: dict) - Optional[List[dict]]: # LangChain在调用bind_tools后tools会出现在kwargs的functions或tools里 tools kwargs.get(tools) or kwargs.get(functions) if tools: return tools return None拿到tools后如果目标服务支持就直接加到请求体如果不支持就生成一段工具说明拼到system消息里同时要求模型输出带action和action_input的JSON。LangChain的JsonOutputFunctionsParser等组件可以帮你解析这种结构化输出但你得保证模型遵循格式。在function calling这一节很多人忽略的是模型服务返回的“工具调用”格式和LangChain期望的格式未必一致。LangChain的AIMessage里工具调用是放在tool_calls字段中的结构是[{name: ..., args: {...}, id: ...}]。所以你的封装代码从模型服务拿到结果后必须把服务端返回的function_call或tool_calls转换成AIMessage(tool_calls[...])才能让Agent正确调度。我一般会在_generate里做这种转换if resp.choices[0].message.tool_calls: tool_calls [] for tc in resp.choices[0].message.tool_calls: tool_calls.append( { name: tc.function.name, args: json.loads(tc.function.arguments or {}), id: tc.id, } ) message AIMessage(contentcontent, tool_callstool_calls) else: content resp.choices[0].message.content or message AIMessage(contentcontent) return ChatResult(generations[ChatGeneration(messagemessage)])这里有个经验如果你的自定义模型没有返回id字段LangChain的一些组件会出问题所以尽量在服务端就生成一个唯一id。如果服务端拿不到id客户端也可以伪造一个uuid但要注意同一个工具调用在流式和非流式下id必须一致否则Agent跟踪会乱。4. 把封装好的模型接到Dify和CrewAI4.1 Dify模型供应商接入要点Dify这类平台的逻辑和LangChain不太一样。LangChain是代码库层面的框架你可以直接写类来接入Dify是一个平台它要求你把自己的模型服务注册成“模型供应商”。好在Dify提供了一套供应商接入机制本质上是让你实现一个符合Dify调用规范的接口。接入前先看你的自定义模型服务能不能给出一套OpenAI兼容协议如果可以最快的办法是走Dify的“OpenAI-API兼容”供应商配置填一个base_url就行。这里要注意的是Dify对base_url的路径要求有的版本需要你填完整到/v1有的版本会自动补填错了会报连通性测试失败。如果你的模型协议完全不同那就要在Dify里做“自定义模型供应商”。基本流程是在Dify后端代码里新增一个供应商目录配置供应商名称、图标等元数据。实现模型类型的接口比如LLM类型需要实现validate_credentials、generate、generate_stream等方法。返回给Dify的response结构必须符合Dify规定的字段text、message_id、usage等。我在实际接入时发现Dify对自定义模型最看重的其实是流式。因为Dify的Agent应用在对话流中会持续把输出增量推送给前端如果你的自定义模型不支持流式Dify为了兼容只能模拟非流式输出效果会打折。所以如果你打算把模型接进Dify务必先确认模型服务能支持stream: true并且在服务端正确返回[DONE]结束标记。4.2 CrewAI直接用LangChain封装CrewAI用起来比Dify更“代码化”而且它对模型的处理方式很有意思大多数时候你不需要自己实现一个CrewAI专属模型类直接传一个LangChain的ChatModel对象进去就行。比如你刚写好的MyCustomChatModel实例可以直接这样用from crewai import Agent, Task, Crew llm MyCustomChatModel(base_urlhttp://127.0.0.1:8000/v1) agent Agent( role研究分析师, goal分析给定的技术主题, backstory你是一名资深技术分析师, llmllm, verboseTrue, )因为CrewAI内部大量复用了LangChain的模型调用规范所以只要你封装出来的BaseChatModel质量达标CrewAI的Agent、Task、Crew这些组件就都能正常工作。这里我还想提醒一点CrewAI对tool_calls的依赖比LangChain裸用更强如果你的自定义模型在function calling的返回结构上不规范CrewAI在调度工具时会出现“工具调用了但Agent没有下一步动作”的假死现象。排查时优先看你模型的tool_calls字段有没有id、name、args这三个完整属性。4.3 不同框架的接入共性适配器模式不管是LangChain、Dify还是CrewAI接入的本质都是同一个模式——适配器。框架那边有它固定的调用协议你的模型服务有自己的协议你写的那一段代码就是一个适配器。不同框架的适配器表现形式不同但核心逻辑高度一致把框架传来的标准消息列表映射成模型服务能理解的输入。把模型服务返回的结果映射回框架标准消息对象。在流式调用中把模型服务吐出的增量块逐片翻译成框架的增量块。在工具调用中把工具的JSON Schema转成模型服务的工具声明再把模型返回的工具参数转成框架标准tool_calls。理解了这一点你再看任何新框架的模型接入文档思路都会清楚很多——你只需要找到一个地方写这四个映射就行不用每个框架都翻一遍源码。5. 我踩过的坑和对应排查链路5.1 流式文本被框架当成完整结果有一次我在LangChain里跑agent.stream()发现最终结果里文本只有最后一个token其它全丢了。排查了一圈发现是因为我在_stream里没有用AIMessageChunk持续累积而是每次yield了一个完整消息的副本。LangChain的流式本质是“增量”每个chunk都是对消息的一部分描述框架会把这些chunk拼起来。如果我在每个chunk里都放完整文本框架会把它拼成重复文本。这提醒我流式chunk的内容只放本次增量不是累计结果。如果你的模型服务返回的是全量文本而不是增量delta那在_stream里要做一次差分计算即把当前全量文本和上一次已发送文本做差只把新增部分放进AIMessageChunk。这个看起来很笨但在很多自研模型服务里真的会遇到因为它们的SSE实现不标准。5.2 function calling参数总是解析失败工具调用解析失败是最让人头疼的问题之一。我遇到过的情况是resp.choices[0].message.tool_calls能拿到但arguments字段是一段不合法JSON导致json.loads直接抛异常。后来我在服务端给模型加了“JSON修正”提示词并且要求模型对参数做一次语法校验才返回。如果你没法改服务端那就在客户端做容错用正则配合json.loads的延迟解析或者直接让大模型重新修复JSON。但后者会增加一次额外调用成本略高只建议在调试时用。还有一个隐藏坑是有些模型服务即使声明支持function calling但它只在tools列表为空时才会返回普通文本一传tools就罢工。我推荐的排查顺序是先不传tools看普通对话是否正常再传一个最简单的tool看返回结构再逐步增加工具数量。不要上来就把五六个工具全塞进去出了问题根本分不清是哪个工具定义惹的祸。5.3 token统计和平台对不上在Dify后台我很早发现自定义模型的token用量经常显示为0或者和模型服务端日志对不上。原因是Dify在自定义模型接口里规定token用量必须由供应商在response里返回usage字段包括prompt_tokens、completion_tokens、total_tokens。很多自研模型服务压根不回这个字段Dify拿不到就只能显示0。解决的办法就是在服务端或封装层估算token。如果你用不了tiktoken那最简单的办法是让服务端根据字符数估算token并填充到usage字段。虽然不准但至少能让你看到趋势不至于成本失控。LangChain这边也有类似问题。BaseChatModel有一个_generate的返回值里可以直接带llm_output里面放usage。我一直建议在llm_output里至少放一个token_usage字典这样LangSmith、Langfuse这类追踪工具才能把成本记录下来。5.4 并发一高就超时Agent场景下并发是常态尤其多个Agent同时跑时自定义模型服务如果只支持单连接或者没做连接池就会出现大量超时。我最早写的封装里每次_generate都新建一个OpenAI客户端没有复用连接。后来并发一上来服务端日志全是连接建立和断开性能数据很难看。解决的方案是把OpenAI客户端作为实例属性或模块级单例让HTTP连接复用class MyCustomChatModel(BaseChatModel): def __init__(self, **kwargs): super().__init__(**kwargs) from openai import OpenAI self._client OpenAI(base_urlself.base_url, api_keyself.api_key)另外给封装层加一个request_timeout参数也很重要。模型服务偶尔变慢会把整个请求链路拖死如果你不设超时Dify和LangChain的请求都悬在那里用户会看到长时间无响应。一般我会把默认超时设置在30到60秒之间Agent调用链较长的场景可以放宽但不要让单个模型请求无限等待。前阵子帮朋友排查一个CrewAI项目现象是第一个Agent正常第二个Agent工具调用后就没了下文。最后定位到问题不是出在Agent逻辑而是自定义模型返回的tool_calls里缺少id字段CrewAI在内部做消息关联时找不到对应关系直接中断了。那个项目改了一行代码补上id就好了。所以封装自定义模型时真正值钱的不是把请求发出去拿到结果而是把每个字段都对齐到框架的预期上。最后分享一个我自己常用的调试技巧封装好后不要直接上Agent先用框架自带的基础调用链测一遍。比如在LangChain里跑一下model.invoke(messages)、model.stream(messages)、model.bind_tools(tools).invoke(messages)这三步都能跑通再接Agent就会顺畅很多。把接口契约先钉死再谈上层业务这是我做了几个项目后最深刻的体会。
阅读完成 · 觉得有帮助？
