如果你做过Agent项目大概率遇到过这种场面框架装好了编排逻辑跑通了结果到了选模型这一步卡住——内置选项翻来覆去就是那几家官方大模型而项目真正要用的是一个自己微调过、部署在内网、接口还不太标准的模型。把这样的模型接进Agent流程并不是写个HTTP请求那么简单它需要你对齐一套“Agent框架认为模型应该长什么样”的接口契约。本文就是我这次自定义模型封装实践的全过程包括契约设计、适配层实现、并发与流式处理以及不得不踩的坑。适合正在自研Agent编排层也就是常说的Agent harness、或者想把本地模型接入Agent项目的开发者。1. 为什么“模型封装”是Agent项目的隐形地基1.1 一个从“框架只支持官方大模型”开始的真实场景我接手的项目是一个偏私域场景的智能体系统核心诉求是把AI Agent的能力嵌进客户内部流程里数据的合规性要求很高模型不能直接往公网大模型API上传。所以方案很明确在自己的机房或私有云里部署一个量化过的开源模型用vLLM或者类似推理框架起服务。当时团队的Agent编排层是自研的原因很简单——业务流程里定制化逻辑太多比如特定工单系统的工具调用、多轮流程里的人工接管节点通用框架在这些地方反而要绕很多弯。但自研有个代价所有模型接入都要自己维护。我给这个编排层起个名字就叫harness后面都用这个词。harness负责任务拆解、工具执行、记忆管理但真正“生产文字”的地方还是模型。问题马上来了harness从设计之初就假设后端模型输出的是标准的chat格式但内部微调模型的服务接口返回的是一个自定义JSON结构字段叫result_text版本号还嵌套在meta里。这其实就是“自定义模型封装”最朴素的起点——不是让模型本身做什么而是在模型和Agent编排层之间插入一层适配转换模块。1.2 封装不是调API而是对齐一份能力契约很多人会把封装理解成“写一个函数把请求发出去拿到响应再处理”。这种理解太浅了。Agent场景里模型不是被调用一次而是被调度几十上百次每次都要携带不同的上下文、工具定义、状态约束。封装层要做的不是一次请求转发而是把“模型能力”稳定地暴露成一套Agent可以长期依赖的接口这叫能力契约。能力契约至少包括几件事多轮对话消息列表的输入输出格式稳定角色信息不能丢。参数透传温度、最大长度、停止符这些采样参数要能按一次调用粒度传递。工具调用模型输出的结构化工具请求能被Agent正确解析并执行。流式输出部分场景需要逐字返回封装层不能把流式能力阉割掉。错误语义超时、限流、服务不可用这些错误要被识别并转化而不是抛一个底层HTTP异常让harness无从处理。一旦封装层做不好问题不会立刻暴露在功能上而是暴露在所有“凭空消失的回复”和“偶发的超时重试”上排查起来非常头痛。1.3 三条主流封装路线与我的选型目前把自定义模型接进Agent系统主流做法有三条路线适用场景优点缺点OpenAI兼容代理模型本身有HTTP接口但格式不统一接入快、生态成熟、工具多协议天花板明显高级能力难透传继承框架基类深度使用LangChain等框架与框架能力无缝衔接与特定框架强绑定升级成本高自研harness协议适配自己维护Agent编排层灵活、契约完全可控需要自己做设计坑要自己趟我这次选的是第三条路线。坦白说如果项目一开始就是基于LangChain搭的我大概率会反着选——好好继承它的BaseChatModel接口省心。但自研harness意味着上层全是我们自己的代码再引入一个框架的抽象层意义不大反而增加了包体积和升级风险。自己设计一套轻量的适配层是最自然的选择。2. Agent框架到底向模型要什么先把契约拆开看2.1 最小闭环多轮对话补全先看最基础的能力。一个Agent任务跑起来harness会把系统提示词、用户问题、历史消息拼成一个消息列表发给模型。这看起来简单但模型封装层经常会在这里犯两个错第一个错是不保留系统消息。真实Agent场景里系统提示词承载了任务边界、工具使用规则、输出格式要求。如果封装的适配层内部对消息做了裁剪把system消息过滤掉了模型的回答会瞬间“失去灵魂”。我见过同事写的适配器为了省事直接只拿最后一条用户消息效果就是Agent频繁地“忘记规则”。第二个错是角色映射混乱。自定义模型服务可能不叫“user/assistant/system”而是用“human/bot/context”这种命名或者干脆用数字role编码。封装层必须做角色映射而且要保证tool类型的消息也能正确传递。很多模型服务不支持tool角色需要你把它转成普通user消息同时带上工具执行结果的前缀说明。2.2 工具调用Agent的“手”能不能伸出去Agent和普通聊天机器人的分水岭就是工具调用能力。模型输出一个工具调用的结构化请求harness解析后执行对应函数再把函数结果返回给模型模型基于结果生成最终回复。这个闭环跑通Agent才算“能干活的Agent”。封装层在工具调用上的责任是什么是让模型输出的工具请求变得稳定可解析。理想情况下模型服务原生支持OpenAI风格的tool_calls数组封装层直接透传。但自定义模型往往不是这样很多微调过的模型只会以自然语言形式“说自己想调用某个工具”这时候封装层要做的就是把自然语言输出转成标准结构或者反过来把标准的工具定义转换成模型能理解的自然语言描述注入到提示词里。这部分细节后面第4节单独展开这里先说一个结论工具调用协议的稳定性决定了你的封装层价值的80%。2.3 结构化输出与推理内容除了工具调用Agent还经常要求模型输出严格的JSON结构比如“从用户问题里抽取三个关键字段以JSON返回”。很多自定义模型在JSON这件事上不够稳定动不动就夹带一段解释文字或者把JSON包在markdown代码块里。封装层可以在后处理阶段做修复剥离多余的markdown标识、定位第一个和最后一个花括号、修复末尾的逗号。但更稳妥的做法是在请求时就约束把输出格式要求写进系统提示词并且在采样参数上把temperature调低甚至设置为0附近减少发散。另外现在很多推理模型会区分reasoning_content草稿思维链和content正式回答。如果你的自定义模型也输出推理内容封装层最好单独透出一个字段给harness而不是把推理和回答混在一起。否则harness把推理过程也当成回答内容展示给用户体验会非常奇怪。2.4 上下文长度与系统提示词的隐性要求最后一类契约要求往往被忽略上下文长度。Agent的提示词很膨胀原因在于工具定义会占据大量token。每多一个工具描述文本就多几百token。如果你的模型上下文窗口只有4k而harness又不知道这个限制它会把所有历史消息一股脑塞进请求里后果就是模型只看到后半段对话前半段信息被截断丢失。封装层应该显式暴露一个max_context_length属性harness在组装上下文之前先询问封装层超长时做摘要压缩或者定向裁剪。这一步不在封装层里实现但封装层必须把能力边界告诉上层。我见过太多项目在这上面吃暗亏表现就是Agent“忘性大”明明之前提过的事后面就忘了其实是被截断了。3. 自研harness路线一个完整的模型适配层实现3.1 为什么我不直接继承LangChain的BaseChatModel先说结论团队没有深度使用LangChain所以不值得为了一个模型接入就把整套LangChain抽象引入。LangChain的BaseChatModel设计得很完善提供了_generate、_stream、bind_tools等一系列钩子继承它可以直接获得框架的缓存、重试、回调等能力。如果你的Agent整体就是基于LangChain/LangGraph搭建的那我强烈建议走这条路——框架已经为你处理了大量边界问题省心。但自研harness的情况下引入LangChain的抽象层会出现两个别扭的地方抽象层隐含的类型假设和生命周期管理与自研代码不同调试时总是隔着一层。为了一个适配器把整个LangChain核心库拽进依赖树维护成本高。所以我的选择是定义一套最简的接口只覆盖harness真正需要的能力其他一概不做。这让适配层非常轻量任何同事看一眼代码就能完全理解不依赖框架文档。3.2 统一请求/响应数据模型先定义数据模型。这里的信息量不只是“字段定义”更是整个系统的契约锚点。我直接用Python的dataclass做后续如果要跨语言比如把适配层用Rust重写一版也可以基于同样的JSON schema。from dataclasses import dataclass, field from typing import Literal, Optional Role Literal[system, user, assistant, tool] dataclass class ToolCall: id: str name: str arguments: str # JSON字符串由harness层解析 dataclass class ChatMessage: role: Role content: Optional[str] None tool_calls: Optional[list[ToolCall]] None tool_call_id: Optional[str] None dataclass class ModelRequest: messages: list[ChatMessage] temperature: float 0.7 max_tokens: int 2048 stop: Optional[list[str]] None stream: bool False dataclass class ModelResponse: content: str reasoning_content: Optional[str] None tool_calls: Optional[list[ToolCall]] None usage: Optional[dict] None finish_reason: Optional[str] None这个模型看起来简单但设计时我特意做了几件事ToolCall.arguments存JSON字符串而不是dict。原因是模型输出的工具参数可能包含非法JSON封装层不应该负责完整校验harness解析失败时能够看到原始原料方便排查。ModelRequest带完整的采样参数。有些Agent框架封装模型时就传一个prompt字符串温度、最大长度全靠全局配置这在多任务场景下是不够的不同任务的偏好多有不同。finish_reason保留。它把“模型是正常结束还是因为长度上限被打断”告诉上层这个信息在Agent场景里很重要后面会讲。3.3 适配层核心超时、重试、鉴权一个都不能少数据模型定了以后适配层的工作就是“把这个模型转换成自定义推理服务能吃的格式发出去再把返回转回ModelResponse”。听起来简单实际最容易翻车的都在边缘逻辑上。先看超时。Agent场景里模型请求不是一两秒就能结束的长上下文加上复杂推理响应可能拖到30秒甚至更久。如果超时时间设短了大量请求被误判为超时设长了模型服务卡死的时候harness也要干等。我的做法是把超时分成两个层次连接超时用5秒读取超时用请求体量的一个比例比如min(300, 30 max_tokens / 20)秒实测更接近真实需要。再看重试。重试逻辑最忌讳“一刀切”请求失败了就重试三次。正确做法是先区分错误类别不可重试的直接抛出比如400参数错误、401鉴权失败、请求体格式不对重试一万次结果也一样可重试的才重试比如超时、429限流、5xx服务端错误。最后是鉴权。很多内部模型服务不鉴权或者用简单的token头但封装层还是应该把鉴权参数做进去。因为封装层要被多个Agent任务复用有些任务访问的模型服务部署在不同环境鉴权方式可能完全不同参数化处理减少后续改代码的频率。import httpx import time class ModelAdapter: def __init__(self, endpoint: str, api_key: str, max_retries: int 3): self._endpoint endpoint self._api_key api_key self._max_retries max_retries self._client httpx.Client(timeout60.0) def chat(self, req: ModelRequest) - ModelResponse: payload self._build_payload(req) headers {Authorization: fBearer {self._api_key}} retryable_exc (httpx.TimeoutException, httpx.NetworkError) for attempt in range(self._max_retries): try: resp self._client.post(self._endpoint, jsonpayload, headersheaders) except retryable_exc: if attempt self._max_retries - 1: raise time.sleep(self._backoff(attempt)) continue if resp.status_code 500: if attempt self._max_retries - 1: resp.raise_for_status() time.sleep(self._backoff(attempt)) continue resp.raise_for_status() # 4xx这样不可重试的错误直接抛 return self._parse_response(resp.json()) raise RuntimeError(unreachable) def _backoff(self, attempt: int) - float: return 0.5 * (2 ** attempt) # 0.5s, 1s, 2s注意resp.raise_for_status()放在4xx场景下是直接抛异常的不会进入重试循环只有超时和5xx才重试。这个分类是整个重试策略的关键。3.4 接入自定义推理服务的示例上面是通用骨架现在看具体接入。假设内部服务是一个自定义推理接口请求格式长这样{ session: { prompt: 一段拼接好的文本, max_new_tokens: 512 } }它不像标准chat接口那样收消息数组而是收一个拼接好的字符串。很多内部模型服务都是这种形态因为它们没有多轮对话的概念需要调用方自己管理历史。封装层就要负责把ChatMessage列表拼成这么一个prompt字符串。拼接的规则不是简单地用换行符连起来而是遵循一套角色分隔的模板system: 【系统提示词内容】 user: 【用户问题内容】 assistant: 【上一轮助手回答】这个拼接逻辑在工具场景下还需要特殊处理。tool消息拼接时要在前面加上工具执行结果tool_call_id: xxx这种前缀让模型知道这段内容不是用户说的而是工具返回的观测结果。解析响应也一样自定义服务返回的可能只是{ data: { result_text: 模型的回复文本 } }那_parse_response就很简单取data.result_text填进ModelResponse.content其他字段留空。后面如果服务升级、支持了工具调用格式也只需要改这个解析函数harness侧完全不需要动。这就是封装的价值变化被隔离在一层之内。4. 封装层扛不扛得住真实场景就看这三块4.1 并发控制信号量、连接池与排队Agent项目一旦上线模型封装层几乎立刻会成为并发瓶颈。原因很简单一个Agent任务在推理过程中会多次调用模型多路Agent同时跑瞬间叠加出来的并发数非常可观。而内部模型服务通常承载能力有限GPU推理或者小型的CPU推理服务能同时处理的请求数也就那么几个。直接在封装层做并发控制是必要的不能指望调用方自觉限流。我的做法是引入asyncio信号量和httpx的AsyncClient连接池。import asyncio import httpx class AsyncModelAdapter: def __init__(self, endpoint: str, api_key: str, max_concurrency: int 8): self._semaphore asyncio.Semaphore(max_concurrency) self._client httpx.AsyncClient(timeout60.0, limitshttpx.Limits(max_connectionsmax_concurrency)) async def chat(self, req: ModelRequest) - ModelResponse: async with self._semaphore: payload self._build_payload(req) resp await self._client.post(self._endpoint, jsonpayload) return self._parse_response(resp.json())max_concurrency这个参数很关键。它不是越大越好而是要根据模型服务端的实际承载量来定。怎么测跑一个简单的并发脚本逐步加大并发数观察模型服务的响应延迟和错误率找到拐点留20%余量就是比较合理的值。如果并发数超过信号量上限请求会发生等待。等待本身不是坏事它让模型服务不会被突发流量打崩。但等待时间也不能无限长所以信号量获取的地方建议配合一个带超时的获取方式比如asyncio.wait_for(self._semaphore.acquire(), timeout20)避免任务在排队中卡死。4.2 流式输出SSE的真实形态与解析很多Agent场景需要带流式输出最典型的就是对话型Agent用户界面上文字要一个字一个字蹦出来。模型服务走的是SSEServer-Sent Events也就是HTTP响应内容按data:前缀一行一行推过来。封装层如果只是等全部结果出来再返回那流式体验就全没了。实现流式封装核心是定义一个回调函数或者异步生成器。我的做法是让chat方法在streamTrue时变成异步生成器逐段产出文本增量。async def chat_stream(self, req: ModelRequest): payload self._build_payload(req, streamTrue) async with self._client.stream(POST, self._endpoint, jsonpayload) as resp: async for line in resp.aiter_lines(): if not line.startswith(data:): continue data json.loads(line[5:].strip()) if data.get(type) delta: yield data[delta], None # (文本增量, 工具增量) elif data.get(type) tool_call_delta: yield None, data真实环境里SSE的格式五花八门有些服务每行前缀是data: {json}有些是data:{json}冒号后面有没有空格都随缘。解析时千万不要用split(:)[1]这种硬切方式用line[5:]的前提是你确认前缀固定更稳的做法是从line.find( )或者line.find({)开始截取。这个细节我踩过后面具体说。流式场景下还有一个问题token统计。模型服务在流式结束后一般在最后一行返回usage封装层要把这个信息保留下来传给需要计费或者日志统计的上层模块。4.3 工具调用从“输出JSON”到“稳定可解析”工具调用是封装层里最让人头疼的部分没有之一。一个大厂的商业模型官方API工具调用格式是严格约束的即使格式有偏差你也能依赖它的稳定性。但自定义模型可不惯着你它输出工具请求的方式千奇百怪。我遇到过的情况模型输出一段自然语言“我需要查询工单系统参数是12345”。模型输出一个JSON但包在markdown代码块里。模型输出JSON但字段名跟约定好的不一致比如用args代替arguments。模型输出一个完整的JSON数组但数组外面还包了一层解释文本。封装层的职责是尽量把这些情况“归一化”成标准ToolCall结构。我实现的解析策略是层层递进先直接在回复文本里做JSON解析提取name和arguments。失败的话用正则剥离json ... 这样的markdown代码块标记再解析。再失败的话定位第一个{和最后一个}截取中间部分尝试容错解析。利用工具名称字典做模糊匹配在回复文本里搜索已知工具名找到对应位置后截取参数部分。import json, re def parse_tool_call_from_text(text: str, available_tools: dict) - Optional[ToolCall]: candidates [text] candidates.append(re.sub(r(?:json)?, , text)) for cand in candidates: try: obj json.loads(cand) if isinstance(obj, list): obj obj[0] if name in obj: return ToolCall(id, nameobj[name], argumentsjson.dumps(obj.get(arguments, obj.get(args, {})))) except json.JSONDecodeError: continue # 兜底工具名模糊匹配 for name in available_tools: if name in text: args_start text.find({) args_end text.rfind(}) if args_start ! -1 and args_end args_start: return ToolCall(id, namename, argumentstext[args_start:args_end 1]) return None这段代码不是完美的但它在真实项目里把工具调用的成功率从70%拉到了95%以上。剩下的5%失败场景harness可以做一层兜底向模型返回“工具调用解析失败请重新输出”或者直接放弃工具调用让模型用文本回答。这个兜底逻辑不复杂但一定要有否则一个解析失败就会让整个Agent任务卡死。5. 这个改造项目里我踩过的几个坑5.1 max_tokens写死Agent任务被拦腰截断我第一次适配内部模型时图省事把max_tokens固定成了512。单轮问答场景完全没问题但Agent一旦涉及多步推理模型输出很容易超过512结果就是回复被截断成一个半截的句子工具调用请求如果正好落在截断边界整个结构化JSON被切碎harness怎么都解析不出来。排查时第一反应是模型问题后来看finish_reason才发现全是length而不是stop。修正方案是不同任务设定不同的max_tokens跟工具调用相关的任务给足余量甚至直接给到模型上下文上限的一半。同时在封装层日志里记录finish_reason一旦出现大量length就发出告警逼迫自己关注长度问题而不是等它在线上爆。5.2 工具结果回传时丢了role标签harness在模型输出工具调用后执行完工具要把工具结果回传给模型。回传时消息列表里多了一条roletool的消息。问题出在里面有不少自定义模型服务不认tool角色要么返回400要么干脆忽略这条消息。我最初的修法是遇到tool角色就把它转成user角色内容前面加个前缀。这在单次工具调用场景下没问题但轮次一多就出bug——模型无法区分“用户说的话”和“工具返回的结果”偶尔会误以为工具结果是用户的需求回答方向完全跑偏。后来我改用了一个更显式的方案系统提示词里固化一段说明解释“带有【工具结果】前缀的消息是环境观测结果不是用户的指令”同时在转换后的消息内容里保留tool_call_id让模型能够对上号。效果好了很多但这是不得已而为之的下策。如果你用的模型服务真的不支持tool角色这种措辞上的补偿几乎不可避免。5.3 超时重试没分类把“参数错误”也重试了三遍我早期写的重试逻辑很粗暴任何异常都重试三次指数退避。结果有一次部署了新配置比如把max_tokens设置成了模型服务不接受的值服务一直返回400。我的代码傻乎乎地重试了三次每次间隔越拉越长用户看着页面转圈圈转了很久最后才吐出一个错误。问题的根源在于我把不可重试的4xx错误也放进了重试循环。这个坑好解决但它的教训值钱重试分类不是性能优化而是正确性问题。不可重试的错误应当立刻抛给上层让harness及时走失败分支比如回复用户“当前请求参数有问题”而不是让用户干等几十秒最后等来同一个错误。5.4 并发参数不是越大越好我把并发数从8调到32想着内部模型服务性能不错能扛得住。结果模型服务延迟从200毫秒直接飙到2秒错误率也上来了。原因很简单并发过大导致推理任务在GPU上排队单个请求反而变慢系统总吞吐量并没有提升多少但响应延迟完全不可接受。这个调优过程很实际并发往上加观察延迟和错误率找到性能曲线的拐点退回20%。我最后定在了12比最初的8高一些但远不是32。这个参数没有任何公式可以直接算出来只能在你的模型服务上用真实负载跑测试。6. 封装完以后冒烟用例集与后续演进6.1 一套能挡80%回归问题的冒烟用例封装层改完之后最怕的是某次调整把原有的行为悄悄改坏了。我总结了一套冒烟用例集每条都很短但覆盖面足够任何一次改动后跑一遍能挡掉大多数回归问题。单轮问答发一条user消息检查返回的content非空且不含谐波噪音。多轮上下文连续三轮对话检查模型是否记住第一轮提到的关键信息比如名字、编号。工具调用给一个明确的工具描述让模型调用检查解析出的ToolCall参数完整。工具结果回传工具返回结果后模型能够继续回答不把工具结果误认为用户新指令。流式输出检查生成器能够产出多个delta且在最后一行返回usage。超时模拟把模型服务地址改成一个不存在的端口检查重试与最终报错是否符合预期。并发回归用10路并发跑单轮问答检查失败率和延迟是否在阈值内。这套用例我建议你写在自建CI或脚本里而不是靠人工点点点。封装层的改动通常很小但影响面很大自动化验证能节省大量时间。6.2 模型升级时怎么做回归模型服务升级很频繁微调模型换个checkpoint或者推理框架升个版本都可能影响输出行为。做回归时除了重跑上面的冒烟用例还要做两个对比一是输出格式对比。升级前把一组固定问题的输出保存下来升级后再跑一遍做文本相似度对比。不是要求一模一样而是要人工检查差异是否只在“合理范围”内特别是工具调用格式的稳定性这个必须严格对比。二是延迟与token消耗对比。模型的回复长度变化会让Agent的总token消耗和响应延迟出现明显波动。如果新模型回答变得又长又亢长Agent任务的整体成本和用户体验都会受影响。6.3 后续演进多模型路由与轻量化重写封装层稳定后下一步我规划做两件事。一是多模型路由。因为封装层已经把模型抽象成统一接口了harness完全可以在运行时根据任务类型把请求路由到不同模型简单分类任务走小模型省成本复杂推理走大模型保证质量内部模型不可用时自动降级到备用的云端模型。这个能力在目前架构下实现成本很低核心就是把“模型选择”从配置项升级成路由策略。二是用Rust重写一份适配层。目前Python版本跑得很稳但并发场景下Python异步模型的性能还是有天花板。团队里有把适配层下沉到网关进程的计划用Rust实现同样的接口契约用更少的资源扛更大并发。数据模型不变协议不变只是换一层实现这也是当初明确接口契约的最大红利。说句实在的整个改造做完最大的体会不是技术上的而是认知上的模型不是Agent项目的终点而是起点。模型封装这件事本质上是在回答一个非常根本的问题——你的Agent系统到底依赖模型的哪些能力以及你如何确保这些能力被稳定地、通用地使用。把这个契约想清楚比换一个更强的模型重要得多。以后不管底层模型怎么换、服务怎么升级封装层都会是项目最稳固的那一块地基。