做Agent也有段时间了每次换底层模型都让我头大。这个“Agent实践”系列第一篇聊了基础编排今天这篇是第二篇重点放在自定义模型封装。先交代背景我不只是用某个现成平台更多是自己写Python代码或者基于Langflow这类开源框架做二次开发。真正跑起来之后发现最大的瓶颈往往不是Agent的Prompt写得不好而是模型接入层太乱。各家API参数不一样、流式返回格式不一样、错误处理也不一样如果不加一层封装业务代码会被模型SDK绑死换一个模型就得动一堆代码。这篇文章就把这个封装层从设计到落地完整讲一遍。如果你正在搞AI Agent开发想接本地模型或者自定义模型又不想被厂商SDK绑架这篇文章里的方案可以直接抄。1. 为什么需要自定义模型封装1.1 模型接入的真实痛点切换成本与协议分裂大多数Agent项目起步时都会先接OpenAI兼容接口代码写得很顺。等到要接本地部署的模型、企业内部微调模型或者换一个推理服务商麻烦就来了。最直观的问题是参数名对不上同一个“最大生成token数”OpenAI接口里叫max_tokens有的模型服务叫max_new_tokens还有的叫generate_length。温度参数也是有人支持temperature有人只支持top_p还有人两个都支持但取值范围不一样。如果不做归一化每次切换模型都得翻文档然后在业务代码里堆if model xxx这类代码越堆越恶心。还有一个很隐蔽的坑流式返回的格式。OpenAI的流式返回是data: {...}的SSE格式每条消息里带choices[0].delta。但Ollama的流式返回是一个个JSON对象行与行之间没有data:前缀。Claude的流式事件又完全是另一套结构有content_block_delta、message_start等事件类型。如果你在Agent代码里直接解析某一家SDK的返回就等于把产品死死绑在那一家上。我见过一个项目因为底层模型切换改了整整两天最后还有流式对话偶发崩溃的问题。问题不在模型本身而在接入层没有封装。1.2 框架自带模型列表并不够用很多人用Langflow、Dify这类平台打开模型列表一看基本都是几个大厂的官方接口。但实际上项目里要接的往往是内网部署的模型、跑在GPU服务器上的微调模型或者某个第三方兼容服务。平台默认列表里根本没有这些选项所以必须通过“自定义模型”的方式接入。Langflow等框架确实提供User Interface配置项但很多人的第一反应是找官方文档找不到就放弃了。其实这类框架通常支持填一个OpenAI兼容的Base URL只要你的模型服务暴露一个兼容接口就能把自定义模型塞进去。这种“配置地址”的方式适合快速验证但生产环境不够用。你想给不同的模型配不同的超参数、想统计token消耗、想在切换模型时不影响线上流量光靠界面配置是做不到的。所以更稳妥的做法是在框架外面再包一层自己的模型服务或者在框架里用自定义Component做二次开发。这也就是“封装”的第二个原因把模型接入变成项目里一个可替换的组件而不是散落在各个逻辑节点里的硬编码。2. 封装方案设计接口抽象与数据流2.1 核心接口设计Adapter模式封装层最核心的思路是Adapter模式。说得直白一点就是定一套“自己的接口”然后给不同模型各写一个适配器把厂商SDK转换成自己接口。这样业务代码永远只依赖自己的接口不关心底层是哪个模型。我通常这样定义抽象基类from abc import ABC, abstractmethod from typing import AsyncIterator, Optional class BaseModelAdapter(ABC): 所有模型适配器必须实现的接口 property abstractmethod def model_name(self) - str: 返回当前模型名称 abstractmethod async def chat( self, messages: list[dict], stream: bool False, **kwargs ) - dict: 非流式调用返回规范化响应 abstractmethod async def stream_chat( self, messages: list[dict], **kwargs ) - AsyncIterator[dict]: 流式调用产出规范化增量事件这里用async def不是赶时髦Agent场景下并发调用很常见同步接口会卡住事件循环。哪怕你暂时只接一个模型也建议从一开始就用异步接口否则后面做并发会非常痛苦。有了抽象基类每个模型只需要实现两件事一个是普通对话一个是流式对话。比如OpenAI适配器class OpenAIAdapter(BaseModelAdapter): def __init__(self, base_url: str, api_key: str, model: str): from openai import AsyncOpenAI self._client AsyncOpenAI(base_urlbase_url, api_keyapi_key) self._model model property def model_name(self) - str: return self._model async def chat(self, messages, streamFalse, **kwargs): resp await self._client.chat.completions.create( modelself._model, messagesmessages, streamFalse, **kwargs ) return normalize_response(resp) async def stream_chat(self, messages, **kwargs): stream await self._client.chat.completions.create( modelself._model, messagesmessages, streamTrue, **kwargs ) async for chunk in stream: yield normalize_stream_chunk(chunk)你可能会问normalize_response和normalize_stream_chunk是什么这就是封装层的核心——把厂商返回的数据结构转换成自己定义的标准结构。2.2 统一数据协议与参数归一化所有模型适配器返回给上层的数据应该是一套统一的结构。我建议非流式响应统一成类似OpenAI的格式{ id: chatcmpl-xxx, model: gpt-4o, choices: [ { index: 0, message: { role: assistant, content: 回答内容, tool_calls: [] }, finish_reason: stop } ], usage: { prompt_tokens: 100, completion_tokens: 50, total_tokens: 150 } }之所以沿用OpenAI这套结构是因为它已经成了事实上行业标准。下游的Agent逻辑、记忆模块、Tool调度都可以直接读choices[0].message.content。如果接的是Ollama、Claude还是国内模型适配器内部转成这个结构就行。参数归一化也一样。不是简简单单把max_tokens传过来就结束了要做映射。我整理了一个映射表给自己用标准参数OpenAIOllamaClaude说明max_tokensmax_tokensnum_predictmax_tokens生成上限temperaturetemperaturetemperaturetemperature采样温度top_ptop_ptop_ptop_p核采样stopstopstop_sequencesstop_sequences停止符streamstreamstreamstream是否流式看起来很简单但真正麻烦的是默认值。有的服务max_tokens默认很小有的默认无限导致同一个Prompt在两个模型上的输出长度天差地别。封装层必须在调用时显式设置合理的默认值不能信赖第三方SDK的默认行为。2.3 流式与异步Agent交互体验的生命线Agent应用和普通聊天机器人的最大区别是Agent会在一次对话里多次调用模型中间还要插入工具调用、检索、代码执行。如果每次调用都等完整结果再返回用户在界面上看到的就是“一直转圈”非常劝退。流式输出的意义不只是打字机效果更是让Agent的中间状态可被感知。我建议把流式事件也统一成几个类型start: 流开始带上请求ID和模型名delta: 增量内容比如content片段tool_call: 模型要求调用工具包含工具名和参数end: 流结束附带完整统计信息error: 流中途出错在适配器里处理完厂商格式后统一输出这些事件。上层无论是写聊天UI还是做Agent编排都只处理这几种事件。比如OpenAI流式chunk转成标准事件的代码可以这样简化def normalize_stream_chunk(chunk) - dict | None: if not chunk.choices: return None delta chunk.choices[0].delta if delta and delta.content: return {type: delta, content: delta.content} if delta and delta.tool_calls: return {type: tool_call, tool_calls: delta.tool_calls} return None注意上面的normalize_stream_chunk是一次返回一个事件对象但实际流式调用是连续的所以适配器里要配合异步生成器async def wrap_stream(stream): async for raw_chunk in stream: event normalize_stream_chunk(raw_chunk) if event: yield event yield {type: end}这个封装的好处是上层的Agent循环根本不用关心你接的是OpenAI还是本地模型只要迭代拿到事件就行。后续加模型就多写一个适配器Agent主流程一行不用改。3. 实操落地从Langflow到代码级封装3.1 在Langflow中配置自定义模型服务地址Langflow是目前我比较常用的开源Agent编排工具。很多人在Model组件里找不到自定义模型其实Langflow支持通过OpenAI协议接入自定义服务。操作路径一般是在模型组件里选择OpenAI类型的Provider然后在OpenAI API Base或者Base URL字段填你本地服务的地址API Key可以随便填一个占位符很多本地服务不校验但格式必须对再填上模型名称。举个例子你本地用Ollama跑了一个qwen2.5:14bOllama默认兼容OpenAI接口地址通常是http://localhost:11434/v1。在Langflow的OpenAI组件里Base URL填这个地址API Key填ollama或任意非空字符串模型名填qwen2.5:14b就能直接跑起来。如果你有别的服务只要它是OpenAI兼容协议都一样。不过这只是“能用”的阶段。真要上生产不建议直接把内网服务地址暴露给Langflow的各个Flow。团队多人协作的时候模型地址散落在不同Flow里改一个模型就得改一堆节点。所以我更推荐的做法是自己做一层轻量网关把模型统一代理到一个地址Langflow只对接网关。网关后面可以做负载均衡、缓存、token统计将来换模型也不改Flow配置。3.2 代码级封装一个兼容OpenAI协议的处理器Langflow能解决编排问题但自定义逻辑还是得写代码。下面给一个可以照抄的处理器框架我管它叫ModelGateway它使用前面定义的Adapter对外暴露统一调用入口class ModelGateway: def __init__(self, default_model: str local): self._adapters {} self._default_model default_model def register(self, adapter: BaseModelAdapter): self._adapters[adapter.model_name] adapter async def chat(self, messages, model: str | None None, **kwargs): adapter self._get_adapter(model) return await adapter.chat(messages, **kwargs) async def stream_chat(self, messages, model: str | None None, **kwargs): adapter self._get_adapter(model) async for event in adapter.stream_chat(messages, **kwargs): yield event def _get_adapter(self, model: str | None) - BaseModelAdapter: key model or self._default_model if key not in self._adapters: raise KeyError(fmodel adapter not found: {key}) return self._adapters[key]初始化网关的时候把不同模型注册进去gateway ModelGateway() gateway.register(OpenAIAdapter(base_urlhttp://localhost:11434/v1, api_keylocal, modelqwen2.5:14b)) gateway.register(OpenAIAdapter(base_urlhttps://api.openai.com/v1, api_keyos.environ[OPENAI_API_KEY], modelgpt-4o-mini))这样业务代码只要调gateway.chat(messages)或gateway.stream_chat(messages)模型切换完全通过配置文件或注册列表决定。我实测下来这个结构对项目后续维护帮助特别大。新同事接手代码不用关心各种厂商SDK只需要看网关和Adapter。3.3 封装继承多态少写一百个if else前面对Adapter做了抽象但很多人写多模型支持时还是忍不住写if model openai: ... elif model ollama: ...。这种代码为什么不好因为每加一个模型就要改业务逻辑还容易把不同模型的特殊逻辑混在一起。正确姿势是继续使用继承和多态。基类BaseModelAdapter只定义接口各种适配器继承它并实现细节。这样新增模型时你不需要动Agent主流程只需要写一个新类并注册到网关。比如接一个Ollama原生格式的适配器class OllamaAdapter(BaseModelAdapter): def __init__(self, base_url: str, model: str): self._base_url base_url self._model model property def model_name(self) - str: return self._model async def chat(self, messages, streamFalse, **kwargs): # 使用Ollama原生API ...多态的作用在这里体现网关持有的是BaseModelAdapter类型但调用时实际执行的是具体适配器的实现。这个设计就是“开闭原则”——对扩展开放对修改关闭。加模型不用改旧代码风险小很多。有读者可能会问如果某个模型的特殊参数特别多怎么办比如有些模型支持repetition_penalty有些支持beam_search这些参数在标准接口里没有。我建议在chat方法里增加一个extra_body或adapter_params字段把不兼容的参数单独透传但尽量不污染统一接口。把差异隔离在适配器内部上层只关心通用参数。4. 并发、记忆与安全生产环境避坑指南4.1 自定义模型怎么扛并发Agent服务的并发压力跟普通HTTP接口不太一样。普通接口可能几十毫秒就返回了Agent里一次模型调用要几秒甚至几十秒而且流式连接会一直占用。如果不加控制上游随便来十几个并发请求模型服务就被打挂了。我常用的手段是信号量加连接池。信号量控制在途请求数量连接池复用HTTP连接。示例import asyncio from httpx import AsyncClient, Limits class ConcurrencyController: def __init__(self, max_concurrency: int 8): self._semaphore asyncio.Semaphore(max_concurrency) self._client AsyncClient( limitsLimits(max_connections20, max_keepalive_connections10) ) async def call(self, fn, *args, **kwargs): async with self._semaphore: return await fn(*args, **kwargs)这里的max_concurrency不是拍脑袋定的。要看你模型服务的实测并发上限比如本地单卡推理服务可能只能同时处理4个请求那信号量就设4。OpenAI这类云端服务吞吐高一些可以放宽到16或32但也要考虑账号的速率限制。还有一点很重要流式调用的连接占用时间更长并发控制要针对“整个流式过程”生效不能只在请求发起时做限制。否则就变成只限制了开始没限制同时打开多少个流。另外要做超时控制。Agent场景里模型超时是最常见的问题之一。我的经验是给不同接口设置不同超时普通对话15秒流式对话首包5秒、全流程60秒。如果首包5秒还没出来大概率模型服务已经挂了或者队列堵死没必要傻等。4.2 记忆与上下文管理的封装Agent另一个绕不开的问题是记忆。自定义模型封装层最好也管一下上下文因为模型上下文窗口是有限的你不可能把所有历史消息都塞进去。最常见的办法是滑动窗口只保留最近N轮对话。但简单截断会有问题一些早期信息被丢掉后Agent会反复问重复问题。我更推荐“摘要窗口”的组合策略。当历史消息超过一定量时先用模型把早期对话压缩成摘要再把摘要保留在上下文里。封装层可以做成这样一个接口class ContextManager: def __init__(self, max_messages: int 20, max_tokens: int 4000): self._history: list[dict] [] self._summary: str def add_message(self, role: str, content: str): self._history.append({role: role, content: content}) def build_messages(self) - list[dict]: # 组装summary 最近消息 messages [] if self._summary: messages.append({role: system, content: fEarlier conversation summary: {self._summary}}) messages.extend(self._history) return messages注意不要简单把所有历史拼进去。我用过一些Agent框架跑长对话时token急剧膨胀最后模型不愿意回答问题就是因为没做裁剪。封装层做记忆时一定要有token预算。粗略估算中文每个token约1.5个汉字英文约4个字符。比如模型上下文窗口是32k系统提示词占了2k工具定义占了4k那留给记忆的空间大概就是25k左右这个预算要写在配置里。4.3 Agent安全边界与输出校验自定义模型封装最容易被忽略的是安全。Agent有了工具调用能力之后风险指数级上升。模型可能输出一个调用“删除数据库”的指令也有可能在用户输入里注入恶意Prompt覆盖系统设定。封装层应该做几件事第一系统提示词不可被用户消息覆盖。很多模型被越狱就是因为用户说“忽略上面的指示”。封装层可以在组装消息时把用户消息里的“忽略”、“break system”等明显注入模式做一层简单校验或者把系统提示词放在绝对开头并明确要求模型不允许偏离。第二工具调用白名单。Agent需要调用工具时封装层对返回的tool_calls做校验只允许调用已注册的工具并且参数类型要符合预期。我见过一个项目因为工具名来自模型输出一个拼写错误导致去调了不存在的函数最后整个Agent状态崩了。第三输出内容过滤。自定义模型尤其要小心你接的可能是开源模型输出不一定稳定。敏感信息过滤、非法内容拦截这些最好放在封装层统一做不要指望每个Agent节点自己做一遍。我通常会在流式输出阶段对文本块做正则检查命中敏感词的话直接替换为[已过滤]。安全不是可选项尤其当Agent被放在公网服务里的时候。封装层多一道校验就少一类事故。5. 常见问题排查与调试实录5.1 流式解析中断、半截JSON与响应不稳定我实际开发里遇到最多的问题是流式解析。OpenAI兼容接口在流式传输时一个完整事件可能被拆成多个chunk到达也可能一个chunk里塞了好几个事件。如果你按“一行一个事件”去读很容易出现半截JSON解析报错。解决方案是做一个缓冲解析器把所有收到的数据先放进缓冲区然后不断检测缓冲区里是否有完整事件class SSEBuffer: def __init__(self): self._buffer def feed(self, chunk: str): self._buffer chunk while \n in self._buffer: line, self._buffer self._buffer.split(\n, 1) line line.strip() if line.startswith(data:): data line[5:].strip() if data [DONE]: yield None # 结束标记 elif data: yield json.loads(data)注意这里的yield要从feed函数里体现但实际Python里这样写需要把feed变成生成器这只是示例思路。常见做法是维护一个接收循环把解析逻辑放在回调里确保事件边界正确。这个问题我一开始没重视直到生产环境偶发报json.JSONDecodeError排查了半天才发现是chunk被TCP分包切开了。5.2 日志与可观测性别只print排查Agent问题最大的困难是你不知道模型内部到底发生了什么。模型返回的中间步骤多工具调用也多如果只靠print日志会被冲得乱七八糟。我后来把日志分成五层请求层记录request_id、模型名称、消息数量、总token数。适配器层记录调用哪个服务、URL、耗时、状态码。流式事件层记录每次delta的事件类型和长度方便回放。工具调用层记录工具名、入参、出参摘要。安全过滤层记录命中了什么规则过滤了什么内容。一个标准日志行大概长这样[request_idabc123] [agentmain] [modelqwen2.5:14b] [stream_delta len24] [elapsed0.32s]使用结构化日志而不是纯文本能让你在Graphana、Kibana或者云日志平台里直接按request_id聚合查询。我分享一个习惯不管接什么模型都把请求和响应的摘要前200字打到日志里。这样出了问题即使模型配合不积极也可以复盘当时的输入输出。5.3 常见错误排查速查表我把实际踩过的坑整理成一张表可以参考现象可能原因处理方式调用返回401API Key错误或格式不对检查适配器里填的Key很多本地服务要求任意非空字符串返回404Base URL路径不对确认/v1/chat/completions路径是否完整一直无响应直到超时模型服务队列阻塞查看模型推理日志降低并发或增加副本流式中途断开TCP超时或上游关闭连接增加读超时开启心跳或重试机制返回内容截断max_tokens太小在封装层统一设置合理默认值比如1024以上工具调用参数JSON解析失败模型输出有换行或注释解析前先做JSON清理或者约束输出格式Agent突然变成复读机上下文被污染检查历史消息是否混入了工具输出做裁剪排查时要记住封装层把错误类型统一了不代表错误信息丢失。适配器里要把原始错误信息和标准错误信息一起日志输出这样你既能拿到统一接口又不会丢掉诊断线索。还有一个体会不要迷信“模型能力不行”这个结论。很多看起来是模型的问题其实是封装层的参数没有传对。我最开始接本地模型时一直觉得它回答不完整后来发现是num_predict默认值只有256不是模型笨是我的配置把它截断了。这就是封装层要管默认值的原因。最后再分享一个小技巧。自定义模型封装做完之后一定要写一个“模型自检脚本”每次接入新模型先跑一遍固定的测试问题看是否返回标准结构、是否支持流式、工具调用格式是否稳定。五分钟的自检能帮你省下后面一晚上的排查时间。这个脚本本身也可以做成Agent的一个内部工具长期维护很有价值。我现在接新模型第一件事就是把自检跑通再考虑集成到业务里。如果你也在做Agent自定义模型封装建议从这个思路开始先把Adapter接口定清楚再逐步丰富功能。