企业官网建设流程全解析

主流的 Agent 框架用起来顺手，前提是模型得“听话”——能聊、能流式输出、能规规矩矩地返回 JSON、还能在工具调用时给出正确格式的参数。可一旦你想接入的不是 OpenAI 或 Anthropic 这种现成大厂模型，而是公司内部私有化部署的开源模型、自己微调过的业务模型、甚至某个小众厂商的 API，各种别扭就冒出来了。框架不认你的模型，工具调用乱码一样返回，Token 统计对不上，流式输出直接卡死。

这一类问题，本质上就是自定义模型封装没做对。这篇实战记录是我在 Agent 项目里接入非标准模型时踩坑和总结的经验，覆盖从“协议兼容层”到“框架原生扩展”再到“工具调用适配”“多模型路由降级”的完整链路，适合正在做 Agent 开发、被模型接入搞到头秃的工程师参考。

1. 为什么 Agent 框架对模型这么“挑剔”：先搞懂框架的模型抽象层

很多人第一次接入自定义模型失败，第一反应是“这个框架太死板，不支持我的模型”。其实不是框架不支持，而是你还没理解框架对模型到底有哪些默认期待。

1.1 框架眼中“一个合格的模型”长什么样

拿 LangChain、LlamaIndex、Dify 这类 Agent 框架来说，无论底层实现差多远，它们对模型的要求基本可以收敛成五件事：

1. 对话补全能力：输入多轮消息，输出一条回复。
2. 流式输出支持：按 token 逐步返回内容，而不是一次性吐完。
3. Token 计算能力：能估算输入输出的 token 数，方便控制上下文窗口。
4. 工具调用支持：能根据系统提示中的工具列表，返回结构化的工具调用请求。
5. 多轮消息格式：能正确处理 system / user / assistant / tool 四种角色消息。

框架内部的编排逻辑——比如 ReAct 循环、Plan-and-Execute、多 Agent 协作——全部建立在这五个能力之上。模型封装做的事情，说穿了就是把某个具体模型的输入输出格式，翻译成框架能识别的统一格式。

1.2 你引入一个“异形”模型时，框架内部发生了什么

假设你的模型只支持最简单的单轮对话文本，其他什么都不支持。那在 Agent 框架里跑起来会是这个效果：

• 框架发给模型的请求中带了 tools 参数，模型直接忽略，输出普通文本。
• 框架期待返回{"tool_calls": [...]}结构，结果拿到一段聊天内容，解析直接失败。
• 框架为了控制上下文，调用模型的 token 统计接口，发现模型压根没有这个接口，只能退化成字符数估算，上下文管理立刻失真。
• 流式场景下，框架按 SSE 格式解析数据流，你的模型返回的是自定义协议，前端直接白屏。

这些问题不是“小事”，而是每一件都会让 Agent 的核心循环跑不下去。封装自定义模型，本质上就是把这五个能力逐一补上，让“异形”模型在框架面前表现得像个标准玩家。

1.3 一个容易被忽略的事实：模型格式兼容比你想的值钱

我在多个项目里感受过：很多团队费劲开发 Agent 的编排逻辑、工具系统、记忆模块，最后却卡在模型接入上，浪费大量时间做“翻译工作”。如果一开始就把模型封装当成 Agent 项目的第一优先级，后面会省掉非常多麻烦。另外顺带回答一个常见困惑：harness 和 agent 区别到底是什么？harness 是承载 Agent 运行的“外壳”，负责输入解析、工具注入、结果输出这些环节；而 agent 本身是决策与执行的核心。自定义模型封装，恰好就落在 harness 层——你在给 Agent 做一件合身的“外壳适配器”。

2. 兼容层方案：把自定义模型伪装成 OpenAI 兼容协议

如果时间紧、模型本身能力不差，但只是接口格式和框架不匹配，最快的路径不是去实现框架的原生模型接口，而是给你的模型套一层 OpenAI 兼容协议。这是当前开源生态里性价比最高的做法。

2.1 为什么 OpenAI 协议成了事实标准

哪怕你不喜欢 OpenAI，也得承认它的 API 设计已经成了行业默认接口。vLLM、llama.cpp、Ollama、TensorRT-LLM 这些主流推理引擎，全都默认提供/chat/completions和/completions兼容端点。连不少商业模型厂商也在兼容 OpenAI 协议，就是因为客户端生态已经深度绑定。

这意味着：如果你的自定义模型部署在 vLLM 上，那么你告诉框架“这是一个 OpenAI 模型，只是 base_url 指向我自己”，框架就能以最小改动用起来。你花在兼容层上的工作量，基本只是写一个路由转发配置。

2.2 自家模型快速接入的检查清单

在动手之前，先把下面几个问题确认清楚：

• 有没有 OpenAI 兼容端点？启动 vLLM 时是否开启了--api-key和 OpenAI 服务；Ollama 的/v1端点能否访问。
• 流式输出支持是否走 SSE？框架默认按text/event-stream解析，如果你的代理层是普通 HTTP 完整返回，流式体验会退化成“一次性加载”。
• tools 参数是否透传？有些兼容层虽然实现了对话补全，但 tools 参数直接被丢掉了。这会导致框架的 tool calling 静默失效。
• model 字段是否校验严格？部分兼容端点会严格校验模型名，而框架可能默认传一个固定的模型 ID，两边对不上就会 400。

2.3 一个直接可抄的接入示例

假设你在内网用 vLLM 部署了一个 Qwen 模型，地址是http://10.0.0.20:8000，模型名是qwen2.5-14b-instruct。这时你在 LangChain 里完全可以用标准的ChatOpenAI类去接，只要改两个参数：

from langchain_openai import ChatOpenAI llm = ChatOpenAI( base_url="http://10.0.0.20:8000/v1", # 指向 vLLM 的 OpenAI 兼容端点 api_key="EMPTY", # vLLM 默认不校验 key，随便填 model="qwen2.5-14b-instruct", temperature=0.7, streaming=True, )

接完之后，这个llm对象可以直接嵌入 Agent：

from langchain.agents import create_react_agent from langchain.tools import tool @tool def query_stock(symbol: str) -> str: """根据股票代码查询当前价格""" return "mock price: $100" agent = create_react_agent(llm=llm, tools=[query_stock])

这样算下来，整个接入花了不到十分钟。很多模型“接入不了”其实是“没找到正确的端点”，而不是兼容性真的差。

2.4 兼容层方案的天花板在哪里

兼容层不是万能的。以下几个场景，兼容层方案的体验会迅速恶化：

• 模型的返回格式不稳定，比如工具调用偶尔用function_call字段，偶尔用tool_calls字段，无法直接在协议层归一到 OpenAI 格式。
• 模型有特殊的推荐调用方式，比如必须走某种私有 RPC，或需要额外传递业务上下文 ID。
• 你在模型中做了业务逻辑定制，比如内置了鉴权、限流、缓存，这些逻辑不适合暴露为纯 OpenAI 协议。

出现这些情况时，就该考虑第三章的原生扩展方案了。

3. 原生扩展方案：按框架接口实现真正的自定义模型封装

兼容层方案适合“快速跑通”，但生产环境中你会碰到越来越多协议外的东西。这时候需要做真正的模型封装：实现框架定义好的模型抽象接口，把自定义模型的逻辑写进去。

3.1 什么时候必须写原生封装

我的判断标准很简单：如果这个模型在兼容层方案下需要打的补丁超过两个，就值得直接写原生扩展。典型的例子包括：

• 模型需要多阶段推理，先做一次检索，再根据检索结果生成回答。
• 模型返回的内容自带置信度、来源引用、业务字段，需要在 Agent 链路上透传。
• 模型工具调用格式与 OpenAI 差异大，想在内部做专门的解析和格式化。
• 模型需要特殊的鉴权逻辑，比如每次请求都要临时申请令牌。

这些需求在兼容层里只能靠“外挂中间件”实现，而在原生封装里可以干干净净地做进模型类内部。

3.2 LangChain 的 BaseChatModel 子类化实战

LangChain 里自定义模型最规范的方式是继承BaseChatModel，并实现它的核心抽象方法。下面是一个最小可用的实现骨架：

from typing import Any, Dict, List, Optional, Iterator from langchain_core.callbacks import CallbackManagerForLLMRun from langchain_core.language_models import BaseChatModel from langchain_core.messages import ( AIMessage, AIMessageChunk, BaseMessage, HumanMessage, SystemMessage, ToolMessage, ) from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult from pydantic import Field class MyCustomChatModel(BaseChatModel): """ 自定义模型的 LangChain 封装。 假设你的模型只暴露一个 predict(messages) -> str 的同步接口。 """ endpoint_url: str = Field(description="模型服务地址") model_name: str = Field(default="my-model", description="模型标识") max_tokens: int = Field(default=1024, description="最大生成长度") temperature: float = Field(default=0.3, description="采样温度") # ---------- 核心对话补全 ---------- def _generate( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> ChatResult: # 1. 把框架消息列表转成自定义模型的输入格式 payload = self._convert_messages_to_payload(messages) # 2. 调用模型服务 raw_response = self._call_model(payload) # 3. 把模型输出转回 AIMessage，塞进 ChatResult ai_message = AIMessage(content=raw_response["text"], additional_kwargs={"score": raw_response["score"]}) return ChatResult(generations=[ChatGeneration(message=ai_message)]) # ---------- 流式输出 ---------- def _stream( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> Iterator[ChatGenerationChunk]: payload = self._convert_messages_to_payload(messages) # 逐 token 迭代模型返回 for token in self._stream_model(payload): chunk = AIMessageChunk(content=token) yield ChatGenerationChunk(message=chunk) # ---------- Token 估算 ---------- def _get_num_tokens_from_messages(self, messages: List[BaseMessage]) -> int: # 如果没有真实 tokenizer，退化用字符串长度估算 total_chars = sum(len(m.content) for m in messages) return max(1, total_chars // 2) # ---------- 模型不原生支持工具调用时需要声明 ---------- @property def bind_tools(self): # 如果你的模型本身支持函数调用，可以重写此方法： # return super().bind_tools(tools, **kwargs) # 如果不支持，让框架走 prompt 注入方式 raise NotImplementedError( "此模型不支持工具绑定，请使用 prompt 注入方案或工具调用适配层" ) # ---------- 辅助方法 ---------- def _convert_messages_to_payload(self, messages: List[BaseMessage]) -> Dict: """把框架消息格式转成模型服务能接受的格式""" converted = [] for msg in messages: if isinstance(msg, SystemMessage): converted.append({"role": "system", "content": msg.content}) elif isinstance(msg, HumanMessage): converted.append({"role": "user", "content": msg.content}) elif isinstance(msg, AIMessage): converted.append({"role": "assistant", "content": msg.content}) elif isinstance(msg, ToolMessage): converted.append({"role": "tool", "content": msg.content}) return {"messages": converted, "max_tokens": self.max_tokens, "temperature": self.temperature} def _call_model(self, payload: Dict) -> Dict: """实际调用模型服务，这里用 requests 简单示例""" import requests resp = requests.post(self.endpoint_url, json=payload, timeout=60) resp.raise_for_status() return resp.json() def _stream_model(self, payload: Dict) -> Iterator[str]: """实际流式调用模型服务""" import requests with requests.post(self.endpoint_url, json=payload, stream=True, timeout=60) as resp: resp.raise_for_status() for line in resp.iter_lines(decode_unicode=True): if line and line.startswith("data:"): token = line[5:].strip() if token and token != "[DONE]": yield token @property def _llm_type(self) -> str: return "my_custom_chat_model"

实现完这个类之后，用起来和普通的 ChatModel 完全一致：

model = MyCustomChatModel( endpoint_url="http://10.0.0.30:8080/predict", model_name="my-model", max_tokens=2048, ) from langchain.agents import create_tool_calling_agent agent = create_tool_calling_agent(llm=model, tools=[query_stock], prompt=...)

注意：_get_num_tokens_from_messages如果只用字符数估算，在上下文窗口接近上限时会出偏差。生产环境建议接一个真实的 tokenizer，或者让模型服务端返回 token 统计，否则框架的上下文管理会失准。

3.3 LlamaIndex / Spring AI 等其他框架的适配思路

不只是 LangChain，其他框架的封装思路是共通的。比如 LlamaIndex 里对应的基类是LLMPredictor和CustomLLM，你需要实现complete、achat、stream_chat等方法。Spring AI 里则是实现ChatModel接口。核心要点永远是那五个能力：对话、流式、Token、工具、多轮格式。你只要把一套模型能力摸清楚，换框架的成本其实比想象中低。

4. 工具调用适配：自定义模型最容易翻车的地方

谈到 Agent 开发，工具调用（Tool Calling / Function Calling）几乎决定了一个 Agent 能不能真正干活。但这也是自定义模型封装最常翻车的位置。

4.1 从“聊天模型”到“工具调用模型”的鸿沟

普通模型只需要回答文本，工具调用模型在回答之前必须经历一个隐藏步骤：从用户的问题中识别意图，匹配可用工具，并为工具生成符合 schema 的参数 JSON。这个步骤对模型的指令遵循能力和结构化输出能力要求很高。

开源模型中，Qwen 系列、GLM 系列对工具调用的支持相对成熟，但仍有不少模型“声称支持，实际拉胯”——要么工具名幻觉，要么参数类型不符合，要么在多个工具并行调用时直接把 JSON 写崩。封装时要对自己的模型做一次“工具调用能力体检”，不要默认它行。

4.2 三条工具调用接入路径，按顺序选

我整理了三档方案，从强到弱，你可以按模型的真实能力选择：

第一档：模型原生支持 tools 协议。此时只需要在请求中按 OpenAI 格式传 tools，解析返回的tool_calls即可。封装最简单。

第二档：模型不支持 tools 协议但能严格遵循指令。此时走 prompt 注入方案：把工具列表转成 JSON 说明塞进 system prompt，要求模型只输出一条 JSON，里面包含选中的工具名和参数。封装的关键是写清“输出格式约束”：

你是一个工具调用助手。工具列表如下： [{"name": "query_stock", "description": "查询股票价格", "parameters": {"symbol": "string"}}] 根据用户问题，输出如下 JSON（不要输出其他内容）： {"tool": "query_stock", "args": {"symbol": "AAPL"}}

第三档：模型连指令遵循都不稳定。此时必须在后处理层做兜底：用正则或解析器从模型输出里提取 JSON 片段，对参数做类型修复，甚至匹配“最接近的工具名”。这部分逻辑可以参考 LangChain 的_convert_output_to_tool_call思路自己实现，也可以用json_repair库修复残缺 JSON。

4.3 实测踩坑：三种典型故障与排查链路

下面几类问题我在实测中反复遇到，每一条都给出根因和排查思路。

故障一：模型工具调用成功，但参数里混进了多余字段。比如工具只接受symbol，模型却输出了{"symbol": "AAPL", "exchange": "NASDAQ"}。框架默认做严格校验，直接报错。我的处理是在后处理层“白名单过滤”：解析出 args 后，只保留工具 schema 里声明过的字段。这不算作弊，这是在生产里必要的宽容。

故障二：并行工具调用时模型输出格式崩坏。例如应该输出{"tool_calls": [{"name": "tool_a"}, {"name": "tool_b"}]}，模型输出成了两段互相独立的 JSON。后来我改成强制要求模型“每次只调用一个工具，需要多个时输出tool_calls数组”，并且把数组元素个数限制说清楚，情况立刻改善。这里你一定要在自己模型上多试几个 prompt 写法，找到它最擅长的格式。

故障三：流式模式下工具调用被截断。有次模型在流式输出到一半时，工具调用的 JSON 还没吐完，框架就以为结束了。根因是流式切分点刚好落在参数中间。解决方案：在流式解析端做一个“等 JSON 闭合再提交”的缓冲逻辑，或者干脆在工具调用场景下禁用流式。生产环境里，工具调用不流式是完全可以接受的。

5. 多模型路由与降级：封装之外的生产级考量

模型封装做扎实之后，下一个要面对的是生产环境的多模型问题。一个 Agent 项目只用一家模型供应商的方案几乎走不远——成本、限流、可用性，总有一个会逼你做路由。

5.1 为什么要做模型路由：三个现实理由

第一，成本。大模型按 token 计价，不同任务的模型单价能差出几十倍。一个“总结一句话”的需求没必要动用旗舰模型。第二，限流。主流模型 API 都有 RPM / TPM 限制，单一模型扛不住突发流量时，需要把请求分散到多个后端。第三，可用性。任何一家供应商都可能出问题，降级策略决定了你的服务是“缓慢但可用”还是“直接挂掉”。

5.2 路由封装的落地方式

模型路由听起来复杂，落地方式其实非常灵活。最简单的做法是在上一章封装好的模型类外面再包一层 Router：

class AgentRouter: def __init__(self): self.primary_model = MyCustomChatModel(endpoint_url="...", max_tokens=2048) self.backup_model = MyCustomChatModel(endpoint_url="...", max_tokens=2048) self.cheap_model = MyCustomChatModel(endpoint_url="...", max_tokens=512) def get_model_for_task(self, task_type: str, is_streaming: bool = False): if task_type == "quick_summary": return self.cheap_model if is_streaming: return self.primary_model return self.primary_model def get_model_with_fallback(self, task_type: str): model = self.get_model_for_task(task_type) try: return model except Exception: return self.backup_model

如果业务规模没到自研路由的程度，直接用现成的网关组件也可以。比如 LiteLLM 就提供统一的 OpenAI 兼容入口，可以在上游配置多个模型源，并设置故障转移。你做的事情本质上就是：在我的模型类和框架之间，再插一层转发器。这种分层思想在 Agent 架构里非常有用，它让你可以随时换模型而不动业务代码。

5.3 重试与超时的实操经验

最后分享几条和模型调用相关的实战细节：

• 超时不能只设一个值。普通生成请求 30 秒，流式请求首次 token 等待设 10 秒，但总时长放宽到 5 分钟。否则流式长文很容易被误杀。
• 重试要区分错误类型。429 限流可以退避重试，400 参数错误重试一万次都是浪费时间。封装函数里务必把错误分类。
• 日志里一定要记模型名、token 数、耗时。出问题时你第一件事就是看“这次是谁家的哪个模型在什么状态下挂了”，没有这些日志，排查就是大海捞针。
• 模型版本号要进配置。线上看到“同样的代码，昨天正常今天抽风”，八成是上游模型悄悄换了版本。把版本号写进请求参数，或者至少在配置中心锁定。

6. 我自己的一点收尾心得

我陆陆续续接过的自定义模型有七八种了，从接入速度上排个经验：协议兼容层永远最快，原生扩展层最灵活，路由层决定生产稳定性。如果你刚起步，先别急着写一堆抽象类，找一条 OpenAI 兼容端点跑通 Agent 闭环，感受一下框架和模型之间的交互日志，再决定要不要原生封装。

还有一条小技巧：在进行自定义模型封装时，强烈建议先写一组“模型体检脚本”，一次性验证对话、流式、token 统计、工具调用、多轮消息五个能力。这样你每接一个新模型，都能在五分钟内知道它的短板在哪，而不是等 Agent 跑挂了才从一堆日志里找原因。

Agent 项目的瓶颈往往不在框架，而在模型接入的质量和灵活性。把模型封装这层做扎实，后面搭建记忆模块、工具系统、多 Agent 协作都会顺畅很多。