上回我们跑通了一个最小可用的 Agent 之后我本以为接下来的重点是写更多工具、调更复杂的工作流。结果真正卡了我两天的是一件特别不起眼的小事Agent 框架死活不认我自己封装的模型。框架预设的模型厂商就那么几个本地部署的模型服务虽说已经起来了请求发过去要么直接报错要么返回了但框架根本不解析结果。后来我把框架源码里请求和响应的结构翻了一遍才明白所谓自定义模型封装本质上就是给你的模型做一层协议翻译让它在 Agent 循环里看起来像一个标准模型。这篇文章就把这件事从头到尾讲清楚为什么要封装、有哪几条路可以走、工具调用怎么接、流式和并发怎么处理以及最容易踩的几个坑。这篇内容更适合两类读者一类是刚把最小 Agent 跑通、正准备接入自有模型或第三方模型 API 的开发者另一类是本地部署了开源模型、想把它接进 LangChain、CrewAI、Dify 这些框架的人。我默认你看过 Agent 的基本概念知道 harness、消息循环、工具调用这些词大概是什么但不需要对某个框架特别熟——我会把关键结构直接展开讲。1. Agent框架为什么不认你的自定义模型1.1 框架对模型的要求远比生成一段文本多Agent 框架的角色本质上是一个循环调度器它把系统提示词、用户输入、历史消息打包给模型模型返回一个决策框架再根据这个决策决定是直接给出最终答案还是去执行某个工具、把结果回填后再问模型一次。这个循环里框架对模型的要求远远不只是给我一段文字那么简单。至少还包括这么几项需要知道本轮生成了多少 token也就是 usage 字段。框架要用它算成本、控制上下文长度以及判断某个任务是不是快把上下文撑爆了。需要知道模型为什么停止生成也就是 finish_reason。是正常说完了stop还是因为打算调用工具tool_calls还是因为触达了最大长度length。这三种情况在 Agent 循环里的下一步动作完全不同。需要模型支持把工具列表和工具选择策略传进去也就是 tools 和 tool_choice。模型只有在接收工具定义后才可能在回复里返回我要调用哪个工具、参数是什么。需要按约定的消息结构来回传历史包括 system、user、assistant、tool 四种角色的消息。很多自定义模型只习惯你问我答根本没考虑过工具结果回填这件事。这就是为什么你不能像拨数据库一样直接把一个模型接进框架里。不同厂商的模型接口路径不同、字段名不同、能力协商方式不同框架不可能为每个模型单独做适配所以它们制定了一套统一协议让模型侧自己来适配这套协议。这就是封装存在的根本原因。用个不恰当的类比不同国家的插座标准不一样你不能拿欧标插头硬怼国标插座适配器就是你要写的这一层。1.2 OpenAI兼容协议实际上已经成了通用插座市面上大部分 Agent 框架无论是 LangChain、LlamaIndex、CrewAI、Dify还是各种个人项目最后都收敛到了同一套接口规范OpenAI 的 Chat Completions 格式。原因不复杂当年生态扩张的时候大家图省事都直接照着/v1/chat/completions这个 HTTP 接口实现接入层后来自己部署模型服务的团队为了让模型能直接被这些框架使用也照着这个接口去实现。久而久之框架的支持列表里就出现了OpenAI compatible这个分类。一个标准的请求体长得这个样子{ model: my-model, messages: [ {role: system, content: 你是一个有用的助手}, {role: user, content: 帮我查一下北京的天气} ], tools: [ { type: function, function: { name: query_weather, description: 查询城市天气, parameters: { type: object, properties: { city: {type: string, description: 城市名} }, required: [city] } } } ], tool_choice: auto, temperature: 0.7, stream: false }响应体的核心字段则包括choices[0].message.content、choices[0].message.tool_calls、choices[0].finish_reason以及usage里的 prompt_tokens、completion_tokens、total_tokens。Agent 框架拿到这些字段后才能判断下一步该干什么。我见过不少本地推理服务比如用 vLLM、SGLang、Ollama 部署开源模型的那几个主流程方案基本都提供了兼容端点所以很多人会误以为只要是 OpenAI 兼容的就能直接被 Agent 框架用。但实测下来兼容从来不是完全兼容有的实现把 tool_calls 里的 arguments 直接塞成对象而不是字符串有的把 usage 整个省掉有的流式输出最后不发 finish_reason。这些细节差异就是后面一系列奇怪报错的根源。我自己后来干脆养成了一个习惯拿到任何一个自称兼容的服务第一件事就是用脚本打一遍标准请求逐字段核对响应结构而不是直接丢给框架去试。2. 两条路二选一服务侧包装 vs 框架侧适配2.1 路径A在服务侧包装成OpenAI兼容API如果你的模型是通过自有服务对外提供推理能力的比如公司内部部署了一版微调模型或者你写了一个自定义推理脚本最直接的封装方式是在模型前面加一个 HTTP 适配层伪造出一个/v1/chat/completions端点。这个适配层要做的事情很纯粹把进来的 OpenAI 格式请求翻译成你模型服务能理解的格式再把模型服务的响应翻译回 OpenAI 格式。下面是一个用 FastAPI 写的最简适配服务重点看字段映射的逻辑from fastapi import FastAPI from pydantic import BaseModel import uuid, time app FastAPI() class ChatMessage(BaseModel): role: str content: str class ChatRequest(BaseModel): model: str my-model messages: list[ChatMessage] temperature: float 0.7 stream: bool False def my_infer(messages: list[dict]) - str: # 这里替换成你实际的模型推理调用 # 注意如果你的模型服务原生支持工具调用需要在 messages # 之外接收 tools 参数这里用最简逻辑演示字段映射 return 这是本地模型生成的回复 app.post(/v1/chat/completions) async def chat_completions(req: ChatRequest): reply my_infer([m.dict() for m in req.messages]) return { id: fchatcmpl-{uuid.uuid4().hex}, object: chat.completion, created: int(time.time()), model: req.model, choices: [{ index: 0, message: {role: assistant, content: reply}, finish_reason: stop }], usage: { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 } }这不是完整方案只是核心映射逻辑。生产环境你至少还得多做几件事鉴权、请求日志、超时控制、限流以及把内部实际消耗的 token 数填进 usage 字段。尤其是 usage很多框架拿它做成本统计和上下文控制你漏了它框架可能不报错但统计全乱有的框架解析时甚至直接抛异常。2.2 路径B在框架侧实现模型适配器另一种思路是反过来不改模型服务而是在 Agent 框架这边写一个适配器把自有模型包装成框架认识的模型对象。以 LangChain 为例核心是继承BaseChatModel实现_generate方法如果要做流式还得实现_stream方法from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.outputs import ChatGeneration, ChatResult from langchain_core.messages import AIMessage, BaseMessage import requests class MyChatModel(BaseChatModel): endpoint: str http://127.0.0.1:8000/v1/chat/completions def _generate(self, messages, stopNone, run_managerNone, **kwargs): payload { model: my-model, messages: [{role: m.type, content: str(m.content)} for m in messages], } resp requests.post(self.endpoint, jsonpayload) content resp.json()[choices][0][message][content] return ChatResult( generations[ChatGeneration(messageAIMessage(contentcontent))] ) property def _llm_type(self): return my-chat-model这段代码同样是最简版但你应该能看出重点你在框架侧写适配器时需要关心的不只是怎么传文本这一件事而是框架内部所有依赖模型协议的行为——工具绑定、流式输出、token 用量统计、stop 序列传递——都会落到你实现的这些方法上。比如只实现_generate不实现_stream框架一开流式可能报错或退化成普通模式。换个生态也一样JVM 上用 Spring AI 的团队本质也是实现一个 ChatModel 接口CrewAI 允许你传一个自定义 LLM 实例Dify 的自定义模型接入则是走它自己的模型提供商接口。虽然 API 长相不同但背后的适配逻辑完全一致都是把框架期待的模型协议翻译成你实际模型的真实能力。2.3 选择标准改哪一侧更划算我遇到过不少人纠结走哪条路这里直接给出我的判断标准对比维度路径A服务侧包装路径B框架侧适配器改造位置模型服务前面加一层HTTP网关Agent框架内部实现模型类影响范围所有通过HTTP访问的框架都能用只对当前框架生效维护成本一次性做好多框架复用换框架就要重写一遍适用场景自有模型要供多个项目/团队使用快速实验只服务一个框架典型坑字段映射不全、鉴权/限流缺失流式方法没实现、工具协议未处理我个人的偏好是只要模型有服务化部署的可能优先做路径A。因为 Agent 生态变化太快今天你用 LangChain明天可能想试新框架路径A做一次之后接哪里都是同一套地址。路径B更适合你在本地调试、快速验证某个想法或者模型本身只能在某个框架内部运行。3. 工具调用封装Agent能不能动手干活就看这一层3.1 原生tool calling的封装要点Agent 的手和脑靠 tool calling 协议连在一起。框架把工具列表以 JSON Schema 的形式传给模型模型不用真的执行工具只需要在回复里返回我想调用哪个工具、参数是什么。如果模型本身原生支持 tool calling封装工作相对轻松。你只需要把请求里的 tools 参数原样透传给模型推理服务然后把模型返回的决策映射成标准响应结构。一个带工具调用的响应长这样{ choices: [{ index: 0, message: { role: assistant, content: null, tool_calls: [{ id: call_abc123, type: function, function: { name: query_weather, arguments: {\city\: \北京\} } }] }, finish_reason: tool_calls }], usage: { prompt_tokens: 120, completion_tokens: 40, total_tokens: 160 } }这里有一个极其容易踩的细节arguments必须是字符串而不是解析好的对象。我见过好几个自建推理服务在适配时图省事直接塞了一个 JSON 对象进去框架侧解析器如果不做额外容错就直接抛类型错误。框架拿到这个响应后会去调用对应的工具函数再把工具结果以role: tool的消息回传给模型继续下一轮循环。如果你的模型服务原生支持 tool calling封装层通常不需要管工具执行和结果回填框架自己会处理。封装层只需要确保两件事请求里的 tools 原样透传响应里的 tool_calls 按标准格式返回。3.2 模型不认tool calling时的提示词内联方案现实情况是不是所有模型都原生支持 tool calling。很多开源模型、自训练模型、以及一些第三方 API压根没有在协议层实现这个能力。这时候就需要在封装层做提示词内联把工具列表序列化成文本塞进系统提示词逼模型按约定格式输出选择。做法大概是这样。把工具定义转成下面这种文本可用工具列表 1. query_weather查询天气。参数{city: 城市名} 2. search_web联网搜索。参数{query: 搜索关键词} 请严格输出如下JSON格式不要输出任何其他内容 {name: 工具名, arguments: {参数名: 参数值}} 如果你不需要调用工具直接输出 {name: , arguments: {}}然后模型返回的原始文本在封装层做一个解析器提取 JSON校验工具名是否在清单里再把参数透传给工具执行环节。流程上这一步是纯文本进、结构化出所以对模型本身没有任何特殊要求。这个方案特别适合先跑通再说的阶段市面上不少轻量 Agent 项目其实都是这么干的。这个方案有几个明显的局限你提前知道就不至于措手不及。第一工具列表一旦长了会疯狂挤占上下文五十个工具定义塞进去对话质量肉眼可见地下降。第二模型经常不严格遵守只输出 JSON的指令会多打字、夹带解释解析器必须做容错。第三从速度和稳定性来看它都不如原生 tool calling。所以它适合做保底方案不适合作为长期生产路径。3.3 工具调用结果回填封装层要不要插手工具执行完结果要以role: tool的消息回传模型让模型基于工具结果生成下一步回答。这一步框架通常自己会拼装不需要封装层处理。但在一种特殊情况下你会遇到麻烦如果你的封装层不严格等工具调用结束就往下走或者你在流式模式下把 tool_calls 拆碎了返回框架那边拿不到完整的 tool_calls 列表就会卡在循环里。另外提醒一点工具结果是一段不可控长度的文本。一个搜索工具可能返回十几 KB 的内容你在封装层做使用者时最好对工具结果做一次截断或者摘要。实测下来不加控制的话一轮 Agent 任务跑完工具结果塞回上下文导致的 token 膨胀可能比对话本身多出一个数量级。这不是夸张是我确实在日志里看到过 4 轮工具调用之后上下文从 3K token 涨到 33K token 的场景。4. 流式、并发、错误分类能跑和能扛差得远4.1 流式输出比你想的更苛刻Agent 场景里流式几乎是标配因为用户需要实时看到模型在想什么。但流式这里的坑非常密集。标准的 OpenAI 兼容流式响应是 SSE 格式每一块以data: {...}\n\n的形式返回最后以data: [DONE]结束。流式模式下最大的坑在 tool_calls当模型打算调用工具时框架常常会收到一堆碎片其中每个 delta 块里的message.tool_calls参数是分片的。比如arguments字段可能被拆成{city: 北和京}两块需要按 index 拼接起来。如果你的封装层在流式模式下没做增量拼接直接把每个分片单独返回框架拿到的一定是残缺的 JSON工具执行直接失败。另一个高频问题就是丢 finish_reason。SSE 流式响应中最后一个 chunk 通常带有finish_reason: tool_calls或stopAgent 框架拿它来决定是否进入工具执行环节。有的推理服务实现漏掉这个字段所有块都是finish_reason: nullAgent 就会一直等下一个块直接挂起。这个问题的排查也很烦因为它不会报错只是不结束。4.2 Agent请求的并发特征与控制手段聊到并发必须先承认一个现实Agent 请求对模型的压力比普通聊天大得多。普通聊天基本是一问一答Agent 一个任务可能循环调用模型四五次甚至更多且每次循环都要把工具结果、历史对话一起塞回去token 用量翻好几倍。所以给我一个 Agent 怎么扛并发这类问题本质上不是单点的模型请求有多重而是你要不要为一批大请求做削峰。我的建议是分层控制。模型服务侧用好 continuous batching 这类推理优化能力让 GPU 尽量跑满应用侧用信号量或者消息队列控制并发上限避免一瞬间把模型服务打挂。用 Python 的 asyncio.Semaphore 做应用层限流是最简单的做法import asyncio sem asyncio.Semaphore(10) # 同时最多10个Agent任务 async def run_agent_task(task_input: str): async with sem: # 在这里执行Agent任务循环 return await agent_runner.run(task_input)并发压测时一定要分清两个指标首 token 延迟和端到端延迟。Agent 场景里用户感知更强的是首 token 时间而框架侧的稳定性更依赖端到端时间。如果发现端到端延迟随着并发数快速恶化大概率是模型服务的 batch 推理能力达到瓶颈了而不是请求排队的问题。4.3 按错误类型设计的重试与熔断策略Agent 循环中单次模型调用出错处理方法不能一概而论。我按错误类型整理了一张策略表实测下来比较管用错误现象典型原因处理策略429并发超限、TPM/RPM超限指数退避重试同时限制应用层并发数5xx推理服务不稳定、模型加载中短重试两三次仍失败就熔断该端点context_length_exceeded上下文超过模型实际上限截断历史、对中间步骤做摘要而不是直接重试请求超时模型推理慢、流式无数据connect超时调短read超时调长超时后丢弃整个循环非法响应字段缺失、JSON解析失败记录请求快照降级为纯文本对话重试一次有一点特别重要不是所有错误都适合直接重试。如果错误是 context_length_exceeded那说明上下文已经太长了你直接重发同样的请求只会再报一次错。这时候正确的做法是在封装层做一层历史压缩——把早先的对话步骤压缩成摘要再重发。这也算是一种最基础的记忆管理策略。我在生产里给 Agent 加长任务自动摘要功能其实就是从这个错误处理逻辑出发的。5. 实测踩坑与四层验证法5.1 三个花钱买来的教训第一个坑漏了 usage 字段。我最早写适配层的时候觉得 usage 没什么用直接返回了全零。结果接进框架后每次调用后成本统计全是 0倒不是不能用但所有依赖 token 数去做上下文策略的功能全部失效。更麻烦的一种情况是某些框架解析代码直接读 usage.total_tokens没有就抛 KeyError。这个坑排查也不难但会让你怀疑人生。第二个坑流式模式下丢 finish_reason。我有一个模型服务在非流式状态下一切正常一开流式Agent 就卡在最后一步像是等什么东西永远不来。后来抓了原生接口的响应对比发现我实现的流式端点最后少发了带 finish_reason 的 chunk。补上之后一切正常。这个坑提醒我封装层写完不是看起来在流式就行字段完整性和顺序都要严格对照标准实现。第三个坑上下文长度元数据配错。框架一般会根据模型声明的上下文长度来决定什么时候截断历史、什么时候允许请求发出。我一度把元数据配得比模型实际能力大了一倍结果 Agent 跑到一半模型服务报 context_length_exceeded而框架侧根本不知道发生了什么就一直在循环里重试。后来把模型的真实上下文长度写进适配器的元数据里这个报错就消失了。5.2 四层验证法从单聊到并发压测被坑过几次之后我总结了一套四层验证法现在每接一个新的自定义模型都按这个顺序跑一遍。别跨层跳每层通过再进下一层。第一层纯文本对话验证。直接 curl 打你的兼容端点确认最基本的内容生成、usage 字段、finish_reason 都是正常的curl http://127.0.0.1:8000/v1/chat/completions \ -H Content-Type: application/json \ -d {model:my-model,messages:[{role:user,content:你好}]}第二层带工具调用的单轮验证。发一个带 tools 的请求确认模型能返回 tool_calls并且 arguments 是合法 JSON 字符串。第三层流式验证。加stream: true逐块观察 SSE 数据确认最后有[DONE]若中间有 tool_calls 分片要确认 arguments 能拼接成完整 JSON。第四层并发压测。写个脚本用 10、20、50 的并发度各跑一轮记录成功率、P95 延迟、错误类型分布。这一步不是为了看性能上限而是为了暴露前面三层发现不了的竞态问题比如并发下请求上下文互相污染、超时重试打爆服务等。5.3 日志快照是调试Agent的黑匣子最后说一个我强烈建议你在封装层就做好的事请求/响应快照日志。每进来一个请求给它分配一个 trace_id把完整的请求体、响应体、耗时、token 用量、错误堆栈全部记录下来按 trace_id 聚合成一个可回放的文件。这个习惯在调试 Agent 时几乎救命。因为 Agent 任务出了诡异问题你很难判断是模型本身变傻了、工具结果太脏、还是框架拼消息拼错了。有快照日志你就能像回放录像一样把每一轮模型调用拿出来看。我见过太多人排查 Agent 问题时全靠猜就是因为日志只有报错了三个字没有上下文。这个成本很低但收益极大。最后聊几句个人体会封装模型这种事平时没人夸你做得好但只要漏一个字段框架立刻给你颜色看。我自己吃过几次亏之后总结下来就一句话先别追求炫技把纯文本对话、一次工具调用、流式输出这三条基础链路完完整整地跑通再谈并发、记忆、多 Agent 这些进阶内容。基础链路通了后面加什么都顺基础链路里藏着一堆隐性问题直接上并发只会让你面对一堆混合在一起的疑难杂症。如果你正准备给自己的 Agent 项目做模型封装我的建议是第一版就老实做路径A用 FastAPI 包一个 OpenAI 兼容端点因为它是所有框架的共同语言。把所有字段补齐、日志留好、验证用完剩下的就是选 Agent 框架的事了。