企业官网建设流程全解析

LangChain 自定义 LLM Wrapper：封装自部署模型为标准接口的工程实践

一、深度引言与场景痛点

大家好，我是赵咕咕。

三月份的时候老板下了死命令："所有涉及客户数据的场景，不允许调用外部 API。"这意味着我们之前基于 GPT-4 搭建的 Agent 系统，需要全部切到自部署的模型上。公司内部跑着一套 vLLM 集群，Host 着 Qwen2-72B 和 Llama-3-70B。

问题来了：LangChain 原生支持 OpenAI API 格式，但我们的 vLLM 暴露的是兼容 OpenAI 风格的 HTTP 接口，参数名略有不同（比如top_kvstop_p），而且错误处理机制完全不一样。更麻烦的是——我们的 Agent 代码里，ChatOpenAI已经嵌入了 37 个地方。

重写 37 处？不可能。最好的方案是写一个符合 LangChain BaseLLM 接口的自定义 Wrapper，让所有现有代码零改动切换。

二、底层机制与原理深度剖析

2.1 LangChain 的 LLM 抽象层

LangChain 的 LLM 体系核心是BaseLLM（纯文本补全）和BaseChatModel（聊天补全）。所有 LLM 实现都继承这两个基类，只要实现了它们要求的方法，你就能无缝接入 LangChain 的 Chain、Agent、Tool 等所有上层抽象。

2.2 BaseChatModel 的继承契约

继承BaseChatModel需要实现三个核心方法：

• _generate()：接收 messages，返回 LLM 生成结果。
• _stream()（可选）：流式输出。
• _llm_type：返回字符串标识，用于日志和追踪。

关键是_generate()的输入输出类型：输入是list[BaseMessage]，输出是ChatResult。只要你的自部署模型能接收并返回符合这个类型的数据，LangChain 上层完全感知不到底层换了一个模型。

2.3 自定义 Wrapper 的调用时序

调用时序的核心设计思路：

1. 格式转换在 Wrapper 内部完成：外部传入的BaseMessage自动转为 vLLM 需要的 OpenAI Chat 格式，响应也自动转回ChatResult。
2. 连接池复用：使用httpx.AsyncClient的连接池，避免每次请求重新建立 TCP 连接。
3. 指数退避重试：2s → 4s → 8s，最多 3 次重试。
4. 降级链路：vLLM 集群完全不可用时，自动切换到本机 Ollama 运行的小模型，保住基础可用性。

三、生产级代码实现

import asyncio import logging import time from typing import Any, Iterator import httpx from langchain_core.callbacks import CallbackManagerForLLMRun from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import ( AIMessage, BaseMessage, HumanMessage, SystemMessage, ) from langchain_core.outputs import ChatGeneration, ChatResult from pydantic import Field, PrivateAttr logger = logging.getLogger(__name__) class CustomVLLMWrapper(BaseChatModel): """自部署 vLLM 模型的自定义 Wrapper。 继承 BaseChatModel，实现 LangChain 标准接口， 封装 vLLM OpenAI-compatible API 的调用细节。 """ model_name: str = Field(default="Qwen2-72B-Instruct") api_base: str = Field(default="http://localhost:8000/v1") temperature: float = Field(default=0.0, ge=0.0, le=2.0) max_tokens: int = Field(default=4096, gt=0) max_retries: int = Field(default=3, ge=0) request_timeout: float = Field(default=120.0, gt=0) # 降级配置 fallback_enabled: bool = Field(default=True) fallback_api_base: str = Field(default="http://localhost:11434") # 内部状态 _client: httpx.AsyncClient = PrivateAttr() _fallback_client: httpx.AsyncClient | None = PrivateAttr(default=None) def __init__(self, **data: Any): super().__init__(**data) self._init_clients() def _init_clients(self) -> None: """初始化 HTTP 客户端（含连接池）。""" limits = httpx.Limits( max_keepalive_connections=50, max_connections=200, keepalive_expiry=30.0, ) self._client = httpx.AsyncClient( base_url=self.api_base, timeout=httpx.Timeout(self.request_timeout), limits=limits, ) if self.fallback_enabled: self._fallback_client = httpx.AsyncClient( base_url=self.fallback_api_base, timeout=httpx.Timeout(60.0), limits=limits, ) @property def _llm_type(self) -> str: return f"custom_vllm::{self.model_name}" @property def _identifying_params(self) -> dict[str, Any]: return { "model_name": self.model_name, "api_base": self.api_base, "temperature": self.temperature, "max_tokens": self.max_tokens, } def _generate( self, messages: list[BaseMessage], stop: list[str] | None = None, run_manager: CallbackManagerForLLMRun | None = None, **kwargs: Any, ) -> ChatResult: """同步入口（包装异步实现）。""" return asyncio.run(self._agenerate(messages, stop, run_manager, **kwargs)) async def _agenerate( self, messages: list[BaseMessage], stop: list[str] | None = None, run_manager: CallbackManagerForLLMRun | None = None, **kwargs: Any, ) -> ChatResult: """异步生成——核心逻辑。""" payload = self._build_payload(messages, stop, **kwargs) # 尝试主模型（带重试） for attempt in range(self.max_retries + 1): try: response = await self._client.post( "/chat/completions", json=payload ) response.raise_for_status() data = response.json() return self._parse_response(data) except httpx.HTTPStatusError as e: if e.response.status_code == 429: # 限流 wait = min(2 ** (attempt + 1), 30) logger.warning( "vLLM 限流 (429)，等待 %ds 后重试 (%d/%d)", wait, attempt + 1, self.max_retries, ) await asyncio.sleep(wait) continue elif e.response.status_code >= 500: # 服务端错误 if attempt < self.max_retries: wait = min(2 ** (attempt + 1), 16) logger.warning( "vLLM 服务端错误 (%d)，%ds 后重试 (%d/%d)", e.response.status_code, wait, attempt + 1, self.max_retries, ) await asyncio.sleep(wait) continue raise else: # 4xx 客户端错误不重试 logger.error("vLLM 请求错误: %s", e) raise except (httpx.TimeoutException, httpx.ConnectError) as e: if attempt < self.max_retries: wait = min(2 ** (attempt + 1), 16) logger.warning( "vLLM 连接超时，%ds 后重试 (%d/%d)", wait, attempt + 1, self.max_retries, ) await asyncio.sleep(wait) continue raise RuntimeError(f"vLLM 连接失败（超过 {self.max_retries} 次重试）") from e # 全部重试耗尽 → 降级 if self.fallback_enabled and self._fallback_client: logger.warning("vLLM 不可用，降级到 Ollama 本地模型") return await self._fallback_generate(messages, stop) raise RuntimeError("vLLM 不可用且降级未启用") async def _fallback_generate( self, messages: list[BaseMessage], stop: list[str] | None = None, ) -> ChatResult: """降级链路：使用本地 Ollama 模型。""" try: prompt = "\n".join( f"{self._role_name(m)}: {m.content}" for m in messages ) response = await self._fallback_client.post( # type: ignore[union-attr] "/api/generate", json={ "model": "llama3:8b", "prompt": prompt, "stream": False, }, timeout=30.0, ) response.raise_for_status() data = response.json() return ChatResult( generations=[ChatGeneration( message=AIMessage(content=data.get("response", "")) )], llm_output={ "model": "fallback::llama3:8b", "is_fallback": True, }, ) except Exception as e: logger.error("降级模型也失败了: %s", e) raise RuntimeError("主模型和降级模型均不可用") from e def _build_payload( self, messages: list[BaseMessage], stop: list[str] | None, **kwargs: Any, ) -> dict[str, Any]: """构建 OpenAI-compatible 请求体。""" chat_messages = [] for msg in messages: role = self._role_name(msg) content = str(msg.content) if msg.content else "" chat_messages.append({"role": role, "content": content}) payload: dict[str, Any] = { "model": self.model_name, "messages": chat_messages, "temperature": kwargs.get("temperature", self.temperature), "max_tokens": kwargs.get("max_tokens", self.max_tokens), } if stop: payload["stop"] = stop return payload @staticmethod def _role_name(msg: BaseMessage) -> str: """BaseMessage → OpenAI role 映射。""" if isinstance(msg, SystemMessage): return "system" elif isinstance(msg, HumanMessage): return "user" elif isinstance(msg, AIMessage): return "assistant" return "user" @staticmethod def _parse_response(data: dict[str, Any]) -> ChatResult: """解析 vLLM 响应 → ChatResult。""" choices = data.get("choices", []) if not choices: logger.warning("vLLM 返回空 choices: %s", data) return ChatResult( generations=[ChatGeneration(message=AIMessage(content=""))] ) choice = choices[0] content = choice.get("message", {}).get("content", "") finish_reason = choice.get("finish_reason", "stop") return ChatResult( generations=[ChatGeneration( message=AIMessage(content=content), generation_info={"finish_reason": finish_reason}, )], llm_output={ "model": data.get("model", "unknown"), "usage": data.get("usage", {}), "is_fallback": False, }, ) async def aclose(self) -> None: """关闭 HTTP 客户端连接。""" await self._client.aclose() if self._fallback_client: await self._fallback_client.aclose() # ─── Agent 集成示例 ─── async def main(): from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder llm = CustomVLLMWrapper( model_name="Qwen2-72B-Instruct", api_base="http://vllm-cluster.internal:8000/v1", temperature=0.0, max_retries=3, fallback_enabled=True, ) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个智能助手。"), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), ]) agent = create_openai_tools_agent(llm, [], prompt) executor = AgentExecutor(agent=agent, tools=[], verbose=False) try: result = await executor.ainvoke({"input": "解释什么是 RAG"}) print(result["output"]) finally: await llm.aclose() if __name__ == "__main__": asyncio.run(main())

代码中值得注意的几个点：

• 连接池复用：httpx.AsyncClient的max_keepalive_connections配置了 50 个长连接池，避免高并发下频繁 TCP 握手。
• 分级重试：429（限流）和 5xx（服务端错误）走指数退避重试，4xx（客户端错误如 400 Bad Request）直接抛出，不浪费重试次数。
• 降级透明：降级结果通过llm_output["is_fallback"]标记，上层可以感知到"这次是降级回答，质量可能下降"。
• BaseMessage 自动映射：_role_name()方法把 LangChain 的消息类型自动映射到 OpenAI 的 role 格式，上游完全不需要关心。

四、边界分析与架构权衡

4.1 同步 vs 异步

BaseChatModel._generate()是同步接口，但底层的 HTTP 调用必然是异步的。这里用asyncio.run()包装同步入口，实际逻辑全在_agenerate()中。如果你的 Agent 系统全部用ainvoke()链路，_agenerate()会被直接调用，不会触发asyncio.run()的开销。

4.2 降级的代价

降级链路的问题在于回答质量可能显著下降。从 Qwen2-72B 降到 Ollama 的 Llama-3-8B，推理能力下降是必然的。建议在降级时通过generation_info标记，让上游的监控系统能区分"主模型回答"和"降级回答"，分别计算满意度。

4.3 什么时候封装 Wrapper，什么时候直接用 OpenAI 兼容模式？

4.4 Token 计数问题

vLLM 的 OpenAI 兼容端点不一定返回usage信息。如果你的 LangChain 上层依赖get_num_tokens()做上下文窗口管理，需要在 Wrapper 里自己实现 token 计数——最简单的方式是用tiktoken做近似估算，精度损失在 5% 以内。

五、总结

LangChain 的BaseChatModel继承体系为自部署模型提供了一个干净的接入点。通过实现_agenerate()方法，你可以把任何自部署模型（vLLM、Ollama、甚至是自研的推理服务）无缝接入 LangChain 的 Agent 生态。

而一个好的自定义 Wrapper 不只是"把请求转发过去"——它应该包含：

• 连接池管理（减少 TCP 握手开销）
• 智能重试（429 退避、5xx 重试、4xx 放过）
• 降级链路（主模型挂了还有备胎）
• 格式映射（BaseMessage 和 API 格式的自动转换）

这套方案我们线上跑了三个月，日均 50 万次 LLM 调用，vLLM 集群的可用性从裸调时的 99.5% 提升到了 99.95%（重试+降级贡献的 0.45 个百分点）。

自部署模型的封装不是技术难题，是工程耐心题。把边角处理好了，稳定性自然上来。

下一篇预告：Ruff、mypy、pytest 在 RAG 项目中的协作配置，打造 CI 质量防线。