LangChain 设计与实现
第17章 Partner 集成架构
开篇引言
LangChain 的真正威力不在于它自身的代码量，而在于它能够统一地对接数十个甚至上百个 AI 服务提供商。OpenAI、Anthropic、Google、Mistral、Groq -- 每个提供商都有自己的 API 格式、认证方式和功能特性，但在 LangChain 中，它们都通过 BaseChatModel.invoke() 这一个接口被调用。
这种统一性的背后，是 LangChain 精心设计的 Partner 集成架构。每个服务提供商对应一个独立的 Python 包（如 langchain-openai、langchain-anthropic），这些包遵循统一的结构规范，实现统一的抽象接口，通过统一的测试套件验证。
本章将以 langchain-openai 为主要案例，深入剖析 Partner 包的标准结构、ChatOpenAI 的实现细节、标准测试套件的工作原理，以及如何开发自己的 Partner 包。
本章要点
- Partner 包的标准目录结构与 pyproject.toml 配置
- ChatOpenAI 如何实现 BaseChatModel 的核心方法
- bind_tools 的工具 Schema 转换机制
- lc_secrets 与 secret_from_env 的密钥管理模式
- langchain-tests 标准测试套件的设计与使用
- 开发自定义 Partner 包的完整步骤
17.1 Partner 生态全景
截至源码快照（langchain-core 1.2.26 对应的 commit），LangChain 的 libs/partners/ 目录包含 15 个官方 Partner 包：
libs/partners/
anthropic/ # Anthropic (Claude)
chroma/ # Chroma 向量数据库
deepseek/ # DeepSeek
exa/ # Exa 搜索
fireworks/ # Fireworks AI
groq/ # Groq
huggingface/ # HuggingFace
mistralai/ # Mistral AI
nomic/ # Nomic 嵌入
ollama/ # Ollama 本地模型
openai/ # OpenAI
openrouter/ # OpenRouter
perplexity/ # Perplexity
qdrant/ # Qdrant 向量数据库
xai/ # xAI (Grok)每个 Partner 包都是独立发布的 PyPI 包，有自己的版本号和依赖管理。它们通过 langchain-core 提供的抽象接口与 LangChain 生态连接。
这种分布式的包管理模式是 LangChain 生态能够快速扩展的关键。当一个新的 AI 服务提供商出现时，只需要开发一个新的 Partner 包，实现 langchain-core 定义的抽象接口，就可以立即与 LangChain 的所有上层功能（Agent、Chain、LCEL 管道等）无缝配合。新包的开发和发布不需要修改任何现有代码，也不会影响其他 Partner 包的稳定性。
从依赖管理的角度看，这种架构解决了一个核心矛盾：框架需要支持尽可能多的服务提供商，但每增加一个集成就多一个外部依赖。如果所有集成都在一个包中，安装一个只需要 OpenAI 的项目可能需要下载 Anthropic、Groq 等所有 SDK。独立包模式让用户只安装需要的依赖，保持了项目的精简。
flowchart TB
subgraph "langchain-core (抽象层)"
A[BaseChatModel]
B[Embeddings]
C[BaseTool]
D[BaseRetriever]
E[Serializable]
end
subgraph "Partner 包 (实现层)"
F[langchain-openai<br/>ChatOpenAI]
G[langchain-anthropic<br/>ChatAnthropic]
H[langchain-groq<br/>ChatGroq]
I[langchain-mistralai<br/>ChatMistralAI]
J[langchain-ollama<br/>ChatOllama]
end
subgraph "langchain-tests (验证层)"
K[ChatModelUnitTests]
L[ChatModelIntegrationTests]
M[EmbeddingsUnitTests]
end
A --> F
A --> G
A --> H
A --> I
A --> J
K --> F
K --> G
K --> H
L --> F
L --> G
17.2 Partner 包的标准结构
以 langchain-openai 为例，一个 Partner 包的标准结构如下：
langchain-openai/
langchain_openai/
__init__.py
chat_models/
__init__.py
base.py # ChatOpenAI 主实现
azure.py # AzureChatOpenAI
_client_utils.py # httpx 客户端工具
_compat.py # 兼容性处理
embeddings/
__init__.py
base.py # OpenAIEmbeddings
llms/
__init__.py
base.py # OpenAI (传统补全)
tools/
...
output_parsers/
...
data/
_profiles.py # 模型配置
middleware/
...
tests/
unit_tests/
integration_tests/
pyproject.toml17.2.1 pyproject.toml 配置
# langchain-openai/pyproject.toml
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "langchain-openai"
version = "1.1.12"
requires-python = ">=3.10.0,<4.0.0"
dependencies = [
"langchain-core>=1.2.21,<2.0.0", # 核心依赖
"openai>=2.26.0,<3.0.0", # 提供商 SDK
"tiktoken>=0.7.0,<1.0.0", # token 计数
]
[dependency-groups]
test = [
"pytest>=7.3.0,<8.0.0",
...
"pytest-asyncio>=0.21.1,<1.0.0",
...
"langchain",
"langchain-core",
"langchain-tests", # 标准测试套件
]
[tool.uv.sources]
langchain-core = { path = "../../core", editable = true }
langchain-tests = { path = "../../standard-tests", editable = true }
langchain = { path = "../../langchain_v1", editable = true }这份配置文件中有几个值得深入讨论的设计要点。
首先是依赖的精简原则。Partner 包只依赖 langchain-core 和提供商的官方 SDK（在这个例子中是 openai 和 tiktoken），运行时完全不依赖 langchain 主包（langchain 只出现在 test 依赖组里）。这种设计确保了 Partner 包可以在最小依赖环境下工作。用户如果只需要调用 OpenAI 的 API，只需安装 langchain-openai，不需要拉入整个 LangChain 框架。Partner 包之间一般也互不依赖，例外是 langchain-deepseek 与 langchain-xai：ChatDeepSeek、ChatXAI 都直接继承 langchain-openai 的 BaseChatOpenAI，所以两者的运行时依赖里都有 langchain-openai（见各自的 pyproject.toml）。
其次是版本约束的策略。对 langchain-core 的约束是 >=1.2.21,<2.0.0，这意味着只要 langchain-core 保持在 1.x 大版本内，Partner 包就应该保持兼容。这个约束遵循语义化版本规范：大版本号变更代表不兼容的接口修改，小版本号变更代表向后兼容的新功能，补丁版本号变更代表向后兼容的问题修复。下界 1.2.21 则指定了 Partner 包所依赖的最低接口版本。
第三是测试依赖的配置。langchain-tests 被列为测试依赖而非运行时依赖，确保了终端用户安装 Partner 包时不会附带测试框架。[tool.uv.sources] 部分将 langchain-core、langchain-tests 以及测试用的 langchain（libs/langchain_v1）链接到本地源码，使得在单仓库（monorepo）环境下开发时可以直接使用本地修改，无需频繁发布和安装。
最后是代码质量工具的配置。ruff 作为代码检查和格式化工具被配置了详细的规则，包括用 select = ["ALL"] 启用全部检查规则、忽略已知的误报、以及针对测试文件放宽限制。多数 Partner 包沿用了同样的 select = ["ALL"] 起点，但并非全部——例如 perplexity 只选了 E、F、I、T201、UP、S 几组规则，exa、fireworks、huggingface 也是各自列举的规则集。
17.3 ChatOpenAI 实现剖析
ChatOpenAI 是最成熟的 Partner 实现，也是其他 Partner 包的参考范本。下面的源码出自 langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/base.py（该目录里 langchain-openai 自身的版本是 1.1.12，与 langchain-core 1.2.26 各自独立发版）。
17.3.1 类定义与字段
# langchain_openai/chat_models/base.py（节选，字段顺序有调整）
class BaseChatOpenAI(BaseChatModel):
"""ChatOpenAI / AzureChatOpenAI 共用的基类"""
model_name: str = Field(default="gpt-3.5-turbo", alias="model")
temperature: float | None = None
max_tokens: int | None = Field(default=None)
request_timeout: float | tuple[float, float] | Any | None = Field(
default=None, alias="timeout",
)
max_retries: int | None = None
openai_api_key: (
SecretStr | None | Callable[[], str] | Callable[[], Awaitable[str]]
) = Field(
alias="api_key",
default_factory=secret_from_env("OPENAI_API_KEY", default=None),
)
openai_api_base: str | None = Field(default=None, alias="base_url")
openai_organization: str | None = Field(default=None, alias="organization")
# ... 更多字段
class ChatOpenAI(BaseChatOpenAI):
"""OpenAI Chat 模型包装器"""
max_tokens: int | None = Field(default=None, alias="max_completion_tokens")
# ... 覆写与补充几个设计亮点：
- 两级类结构：真正的实现在
BaseChatOpenAI（langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/base.py:554），ChatOpenAI（同文件:2351）与AzureChatOpenAI（langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/azure.py:37）都是它的子类 - alias：字段内部名带
openai_前缀（openai_api_key见langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/base.py:581，同文件里openai_api_base在:635、openai_organization在:638、request_timeout在:646），对外别名则是简洁的api_key/base_url/organization/timeout，所以ChatOpenAI(model="gpt-4", api_key=...)能直接写 - SecretStr：API 密钥使用 Pydantic 的
SecretStr类型，在打印或日志中自动掩码；它还接受同步/异步可调用对象，用于动态取密钥 - secret_from_env：密钥用工厂函数从环境变量读取默认值；
openai_organization/openai_api_base则默认为None，在校验器里再回落到环境变量OPENAI_ORG_ID（兼容旧名OPENAI_ORGANIZATION）/OPENAI_API_BASE（langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/base.py:983） - 默认值大多是
None：temperature、max_retries（langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/base.py:671）、max_tokens（langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/base.py:705）都不在 LangChain 侧硬编码默认值，而是交给 OpenAI SDK / 服务端的默认行为
17.3.2 密钥管理模式
# 密钥声明 -- 覆写 Serializable 上的同名属性
@property
def lc_secrets(self) -> dict[str, str]:
return {"openai_api_key": "OPENAI_API_KEY"}注意这个 lc_secrets 声明在 ChatOpenAI 这一层，而不是基类 BaseChatOpenAI 上（langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/base.py:3113）。这个属性告诉序列化系统：openai_api_key 字段（别名 api_key）对应环境变量 OPENAI_API_KEY。Serializable.to_json 沿 MRO 逐个合并 lc_secrets 时会顺带把字段别名也登记进去（langchain-core-1.2.26/libs/core/langchain_core/load/serializable.py:258），这样即使 kwargs 里出现 api_key 这个别名形式的键，也会被替换成 secret 标记。不过 to_json 收集 kwargs 时用的是字段名，所以实际输出的键是 openai_api_key。序列化时，密钥值会被替换为（实跑 dumpd(ChatOpenAI(model="gpt-4", api_key="sk-test")) 的 kwargs 片段）：
{
"openai_api_key": {
"lc": 1,
"type": "secret",
"id": ["OPENAI_API_KEY"]
}
}反序列化时，Reviver 会从 secrets_map 中恢复密钥值，只有显式传入 secrets_from_env=True（默认 False）时才会再去读环境变量（langchain-core-1.2.26/libs/core/langchain_core/load/load.py:293，取值分支在 :415）。
17.3.3 _generate -- 核心生成方法
BaseChatModel 要求子类实现 _generate 方法。BaseChatOpenAI._generate 的实现在 langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/base.py:1445，简化后（省略了响应头、Responses API 结构化解析等细节）的关键流程是：
def _generate(
self,
messages: list[BaseMessage],
stop: list[str] | None = None,
run_manager: CallbackManagerForLLMRun | None = None,
**kwargs: Any,
) -> ChatResult:
self._ensure_sync_client_available()
# 1. 构建请求：合并 _default_params（model、temperature 等）、stop 与 kwargs，
# 并用 _convert_message_to_dict 把 LangChain 消息转换为 OpenAI 格式
payload = self._get_request_payload(messages, stop=stop, **kwargs)
try:
# 2. 按请求形态分三路调用 OpenAI SDK
if "response_format" in payload: # 原生结构化输出
payload.pop("stream")
response = self.root_client.chat.completions.with_raw_response.parse(
**payload
).parse()
elif self._use_responses_api(payload): # Responses API，直接组装结果返回
response = self.root_client.responses.with_raw_response.create(
**payload
).parse()
return _construct_lc_result_from_responses_api(response, ...)
else: # 普通 Chat Completions
# self.client 即 root_client.chat.completions
response = self.client.with_raw_response.create(**payload).parse()
except openai.BadRequestError as e:
_handle_openai_bad_request(e) # 上下文溢出等错误的翻译，见 17.11 节
except openai.APIError as e:
_handle_openai_api_error(e)
# 3. 将 OpenAI 响应转换为 LangChain 格式
return self._create_chat_result(response)反向的消息转换函数 _convert_dict_to_message（langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/base.py:173）处理了各种角色的映射，下面是省略了 name / id 透传、function_call / audio 等 additional_kwargs 分支、developer 角色的 __openai_role__ 标记以及 function 角色（转成 FunctionMessage）分支后的骨架：
def _convert_dict_to_message(_dict: Mapping[str, Any]) -> BaseMessage:
role = _dict.get("role")
if role == "user":
return HumanMessage(content=_dict.get("content", ""))
if role == "assistant":
content = _dict.get("content", "") or ""
tool_calls = []
invalid_tool_calls = []
if raw_tool_calls := _dict.get("tool_calls"):
for raw_tool_call in raw_tool_calls:
try:
tool_calls.append(
parse_tool_call(raw_tool_call, return_id=True)
)
except Exception as e:
invalid_tool_calls.append(
make_invalid_tool_call(raw_tool_call, str(e))
)
return AIMessage(
content=content,
tool_calls=tool_calls,
invalid_tool_calls=invalid_tool_calls,
)
if role in ("system", "developer"):
return SystemMessage(content=_dict.get("content", ""))
if role == "tool":
return ToolMessage(
content=_dict.get("content", ""),
tool_call_id=_dict.get("tool_call_id"),
)
return ChatMessage(content=_dict.get("content", ""), role=role)17.3.4 bind_tools -- 工具绑定
bind_tools 是 Tool Calling Agent 的基础。BaseChatOpenAI.bind_tools（langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/base.py:1948）将 LangChain 工具转换为 OpenAI 的工具格式：
def bind_tools(
self,
tools: Sequence[dict[str, Any] | type | Callable | BaseTool],
*,
tool_choice: dict | str | bool | None = None,
strict: bool | None = None,
parallel_tool_calls: bool | None = None,
response_format: _DictOrPydanticClass | None = None,
**kwargs: Any,
) -> Runnable[LanguageModelInput, AIMessage]:
formatted_tools = [
convert_to_openai_tool(tool, strict=strict) for tool in tools
]
# 省略：parallel_tool_calls、defer_loading 透传、response_format 转换
if tool_choice:
# 归一化：工具名 -> {"type": "function", "function": {"name": ...}}，
# 内置工具名 -> {"type": ...}，"any" 与 True -> "required"
...
kwargs["tool_choice"] = tool_choice
return super().bind(tools=formatted_tools, **kwargs)convert_to_openai_tool（langchain-core-1.2.26/libs/core/langchain_core/utils/function_calling.py:515）将 BaseTool 的 args_schema（Pydantic 模型）转换为 OpenAI 的 JSON Schema 格式。返回值是一个 RunnableBinding，将工具参数绑定到每次 LLM 调用。
作为对照，ChatAnthropic.bind_tools（langchain-core-1.2.26/libs/partners/anthropic/langchain_anthropic/chat_models.py:1559）改用本包自己的 convert_to_anthropic_tool（同文件 :1934）：它先调 convert_to_openai_tool，再把结果改写成 Anthropic 要求的 {name, description, input_schema} 形状；已经是这种形状的 dict 和内置工具则原样透传。
flowchart LR
A["BaseTool<br/>(name, description, args_schema)"] --> B["convert_to_openai_tool()"]
B --> C["OpenAI Tool Format<br/>{type: 'function',<br/>function: {name, description,<br/>parameters: JSON Schema}}"]
C --> D["llm.bind(tools=[...])"]
D --> E["RunnableBinding<br/>(每次调用自动附带工具描述)"]
17.3.5 with_structured_output -- 结构化输出
BaseChatOpenAI 还实现了 with_structured_output（langchain-core-1.2.26/libs/partners/openai/langchain_openai/chat_models/base.py:2050，ChatOpenAI 在 :3193 又做了一层覆写），让模型直接输出 Pydantic 模型或 TypedDict：
# 使用示例
class Person(BaseModel):
name: str
age: int
structured_model = model.with_structured_output(Person)
result = structured_model.invoke("Tell me about Alice who is 30")
# result 是 Person(name="Alice", age=30) 实例内部实现按 method 分三路："function_calling" 通过 bind_tools 将 Pydantic Schema 作为工具绑定，再用 PydanticToolsParser（langchain-core-1.2.26/libs/core/langchain_core/output_parsers/openai_tools.py:306）解析输出；"json_schema" 则 bind(response_format=...) 走 OpenAI 原生的 Structured Outputs，用 _oai_structured_outputs_parser（Pydantic schema）或 JsonOutputParser 解析；"json_mode" 绑定 {"type": "json_object"}。两层的默认值不同：BaseChatOpenAI.with_structured_output 默认 method="function_calling"，ChatOpenAI 那层覆写只把默认值改成 "json_schema" 后调回父类；ChatAnthropic 的默认则是 "function_calling"。
17.4 标准测试套件
langchain-tests（位于 libs/standard-tests）提供了一套标准化的测试基类，确保所有 Partner 包的行为一致。
17.4.1 测试类层次
# langchain_tests/unit_tests/chat_models.py
class ChatModelTests(BaseStandardTests):
"""Chat 模型测试基类"""
@property
@abstractmethod
def chat_model_class(self) -> type[BaseChatModel]:
"""要测试的模型类"""
...
@property
def chat_model_params(self) -> dict[str, Any]:
"""模型初始化参数"""
return {}
@property
def standard_chat_model_params(self) -> dict[str, Any]:
"""标准参数"""
return {
"temperature": 0,
"max_tokens": 100,
"timeout": 60,
"stop": [],
"max_retries": 2,
}
@pytest.fixture
def model(self, request: Any) -> BaseChatModel:
"""模型 fixture"""
extra_init_params = getattr(request, "param", None) or {}
return self.chat_model_class(
**{
**self.standard_chat_model_params,
**self.chat_model_params,
**extra_init_params,
},
)17.4.2 能力声明
测试基类通过属性方法声明模型的能力，测试用例根据能力自动跳过不适用的测试：
# libs/standard-tests/langchain_tests/unit_tests/chat_models.py:94
@property
def has_tool_calling(self) -> bool:
"""模型是否支持工具调用"""
return self.chat_model_class.bind_tools is not BaseChatModel.bind_tools
@property
def has_tool_choice(self) -> bool:
"""模型的 bind_tools 是否接受 tool_choice 参数"""
bind_tools_params = inspect.signature(
self.chat_model_class.bind_tools
).parameters
return "tool_choice" in bind_tools_params
@property
def has_structured_output(self) -> bool:
"""模型是否支持结构化输出"""
return (
self.chat_model_class.with_structured_output
is not BaseChatModel.with_structured_output
) or self.has_tool_calling注意 has_tool_choice 用的不是"方法是否被覆盖"，而是 inspect.signature 检查 bind_tools 的形参里有没有 tool_choice；has_structured_output 则多了一条 or self.has_tool_calling -- 只要能绑工具，就可以用 bind_tools + 工具解析器凑出结构化输出，因此也算支持。
这种基于方法覆盖检测的能力发现机制非常巧妙：如果 Partner 类覆盖了 bind_tools 方法，就意味着它支持工具调用；如果没覆盖，使用的还是基类的（抛 NotImplementedError 的）方法，则认为不支持。不过这种检测只用于 has_tool_calling、has_tool_choice、has_structured_output 三项；supports_image_inputs、supports_json_mode、returns_usage_metadata 等其余能力仍是布尔属性，默认值写在基类里（多数为 False，returns_usage_metadata 与 supports_model_override 为 True），需要 Partner 按实际情况覆写。
这种"检测而非声明"的设计有两个优势。第一，对上述三项能力，它消除了声明与实现不一致的风险。如果使用布尔属性声明能力（如 supports_tools = True），开发者可能声明了支持但忘记实现方法，或者实现了方法但忘记声明。通过直接检测方法是否被覆盖，声明和实现永远是一致的。
第二，它实现了向后兼容的能力扩展。当标准测试新增了一个能力检测（比如 has_structured_output），所有现有的 Partner 包不需要做任何修改。如果某个 Partner 包确实实现了 with_structured_output 方法，新测试会自动检测到并运行相关的测试用例。如果没有实现，测试会自动跳过。整个过程对 Partner 开发者完全透明。
17.4.3 单元测试与集成测试
标准测试分为两层：
单元测试（不需要 API 密钥）：
- 测试模型的序列化/反序列化
- 测试用标准参数初始化与
_get_ls_params返回的标准参数 - 测试
bind_tools/with_structured_output能否用各种 schema 构造
集成测试（需要真实 API 调用）：
- 测试基本的 invoke/ainvoke
- 测试流式输出
- 测试工具调用
- 测试结构化输出
- 测试多模态输入
classDiagram
class BaseStandardTests {
<<abstract>>
}
class ChatModelTests {
<<abstract>>
+chat_model_class: type
+chat_model_params: dict
+has_tool_calling: bool
+has_structured_output: bool
+model: fixture
}
class ChatModelUnitTests {
+test_init()
+test_init_from_env()
+test_serdes()
+test_bind_tool_pydantic()
}
class ChatModelIntegrationTests {
+test_invoke()
+test_stream()
+test_tool_calling()
+test_structured_output()
+test_image_inputs()
}
BaseStandardTests <|-- ChatModelTests
ChatModelTests <|-- ChatModelUnitTests
ChatModelTests <|-- ChatModelIntegrationTests
class MyModelUnitTests {
+chat_model_class = MyModel
+chat_model_params : dict
}
ChatModelUnitTests <|-- MyModelUnitTests
17.4.4 在 Partner 包中使用标准测试
# tests/unit_tests/test_chat_models.py
from langchain_openai import ChatOpenAI
from langchain_tests.unit_tests.chat_models import ChatModelUnitTests
class TestChatOpenAIUnit(ChatModelUnitTests):
@property
def chat_model_class(self):
return ChatOpenAI
@property
def chat_model_params(self):
return {"model": "gpt-4", "api_key": "test-key"}只需继承测试基类，声明模型类和初始化参数，就自动获得一整套标准测试（pin 版 ChatModelUnitTests 有 8 个测试方法，ChatModelIntegrationTests 有 39 个）。这极大地降低了 Partner 开发者的测试负担。
这种测试设计体现了"继承用于共享行为"的原则。测试基类不仅仅是一组测试方法的集合，它还包含了模型 fixture 的创建逻辑、标准参数的定义、以及能力检测的机制。Partner 开发者通过继承和覆盖属性来定制这些行为，而不需要了解测试的内部实现。
标准测试覆盖的范围非常广泛。单元测试包括初始化测试（用标准参数 temperature、max_tokens、timeout 等能否构造出实例）、序列化往返测试（确保 dumpd 然后 load 能恢复原始对象）、工具绑定测试（bind_tools 能否接受 BaseTool、函数和多种 Pydantic 模型并返回 RunnableBinding）、以及快照测试（确保序列化输出在版本更新后保持稳定）。集成测试则包括基本的文本生成、流式输出、工具调用、结构化输出、多模态输入等功能性测试。
通过标准测试，LangChain 建立了一个"可信赖的集成"标准。当一个 Partner 包通过了所有标准测试，用户就可以相信它的行为符合 LangChain 的预期，可以安全地在 Agent、Chain 和 LCEL 管道中使用。
17.5 Partner 包的命名空间与序列化
Partner 包的命名空间设计与序列化系统紧密关联。每个 Partner 包中的可序列化类都有一个由 get_lc_namespace 返回的命名空间标识。
早期的 Partner 包（如 langchain-openai、langchain-anthropic）覆盖了 get_lc_namespace，返回的是模拟旧模块路径的命名空间，如 ["langchain", "chat_models", "openai"]。这是为了与从 langchain 主包迁移之前的序列化数据保持兼容。
新开发的 Partner 包不应该覆写 get_lc_namespace。默认实现从 cls.__module__ 自动生成命名空间，例如 langchain_groq.chat_models.ChatGroq 的命名空间就是 ["langchain_groq", "chat_models"]。这种基于实际模块路径的命名空间更加准确，也更容易维护。
如果未来需要在包之间迁移类（比如将一个实验性的类从社区包迁移到官方 Partner 包），可以通过在 SERIALIZABLE_MAPPING 映射表中添加重定向条目来保持兼容性，而不需要在新位置覆盖 get_lc_namespace 来模拟旧路径。
命名空间的选择还影响反序列化的安全性。只有 DEFAULT_NAMESPACES 列表中的命名空间才被认为是可信的，其余命名空间要通过 valid_namespaces 参数显式放行。此外 load 还有一道类路径白名单：默认 allowed_objects="core" 只放行 langchain-core 自己的类，"all" 也只放行映射表（SERIALIZABLE_MAPPING 等）里登记过的类路径，不在其中的类要用 allowed_objects=[...] 显式列出。所以一个新的第三方 Partner 包要被默认反序列化，需要框架更新把它登记进这两处；否则只能由调用方显式放行。这确保了只有经过审核的 Partner 包才能默认参与反序列化过程。
17.6 开发自定义 Partner 包
假设你要为一个名为 "MyLLM" 的 AI 服务开发 Partner 包，完整步骤如下：
17.6.1 创建包结构
langchain-myllm/
langchain_myllm/
__init__.py
chat_models.py
tests/
unit_tests/
test_chat_models.py
integration_tests/
test_chat_models.py
pyproject.toml17.6.2 实现 Chat 模型
# langchain_myllm/chat_models.py
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, BaseMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from pydantic import Field, SecretStr
from langchain_core.utils.utils import secret_from_env
class ChatMyLLM(BaseChatModel):
"""MyLLM 聊天模型"""
model: str = "myllm-base"
temperature: float = 0.7
api_key: SecretStr | None = Field(
default_factory=secret_from_env("MYLLM_API_KEY", default=None),
)
@property
def _llm_type(self) -> str:
return "myllm"
@property
def lc_secrets(self) -> dict[str, str]:
return {"api_key": "MYLLM_API_KEY"}
@classmethod
def is_lc_serializable(cls) -> bool:
return True
def _generate(
self,
messages: list[BaseMessage],
stop: list[str] | None = None,
run_manager: CallbackManagerForLLMRun | None = None,
**kwargs,
) -> ChatResult:
# 1. 转换消息格式
formatted = self._format_messages(messages)
# 2. 调用 MyLLM API
response = self._call_api(formatted, stop=stop, **kwargs)
# 3. 转换响应格式
message = AIMessage(content=response["text"])
generation = ChatGeneration(message=message)
return ChatResult(generations=[generation])
def _format_messages(self, messages):
"""将 LangChain 消息转为 MyLLM 格式"""
...
def _call_api(self, messages, **kwargs):
"""调用 MyLLM HTTP API"""
...17.6.3 配置 pyproject.toml
[project]
name = "langchain-myllm"
version = "0.1.0"
dependencies = [
"langchain-core>=1.0.0,<2.0.0",
"httpx>=0.25.0", # HTTP 客户端
]
[dependency-groups]
test = [
"langchain-tests>=0.3.0",
"pytest>=7.0.0",
"pytest-asyncio>=0.21.0",
]17.6.4 编写标准测试
# tests/unit_tests/test_chat_models.py
from langchain_myllm import ChatMyLLM
from langchain_tests.unit_tests.chat_models import ChatModelUnitTests
class TestChatMyLLMUnit(ChatModelUnitTests):
@property
def chat_model_class(self):
return ChatMyLLM
@property
def chat_model_params(self):
return {"api_key": "test-key"}在 pin 版 langchain-tests 下实跑这份单元测试（配上 17.6.2 的 ChatMyLLM），有两点要注意。一是 test_serdes 做快照比对，第一次运行必须加 --snapshot-update 生成快照。二是 test_serdes 反序列化时用的是 allowed_objects="all"，而 ChatMyLLM 的类路径 ('langchain_myllm', 'chat_models', 'ChatMyLLM') 不在映射表里（见 17.5 节），于是报 ValueError: Deserialization of ('langchain_myllm', 'chat_models', 'ChatMyLLM') is not allowed.；其余 7 项通过，test_init_from_env 因未设置 init_from_env_params 而跳过。不打算进入 langchain-core 映射表的第三方包，可以干脆不覆写 is_lc_serializable（test_serdes 会随之跳过），或按 BaseStandardTests 的规定，用带 reason 的 @pytest.mark.xfail 覆写该测试。
# tests/integration_tests/test_chat_models.py
from langchain_myllm import ChatMyLLM
from langchain_tests.integration_tests.chat_models import (
ChatModelIntegrationTests,
)
class TestChatMyLLMIntegration(ChatModelIntegrationTests):
@property
def chat_model_class(self):
return ChatMyLLM
@property
def chat_model_params(self):
return {"api_key": "real-key-from-env"}17.6.5 逐步增加功能
在基本的 _generate 实现之后，可以逐步添加高级功能：
class ChatMyLLM(BaseChatModel):
# 1. 流式输出
def _stream(self, messages, stop=None, run_manager=None, **kwargs):
for chunk in self._call_api_stream(messages, **kwargs):
yield ChatGenerationChunk(
message=AIMessageChunk(content=chunk["text"])
)
# 2. 工具调用
def bind_tools(self, tools, **kwargs):
formatted_tools = [self._format_tool(t) for t in tools]
return super().bind(tools=formatted_tools, **kwargs)
# 3. 结构化输出
def with_structured_output(self, schema, **kwargs):
# 利用 bind_tools + PydanticToolsParser
...flowchart TD
A["开发 Partner 包"] --> B["1. 创建包结构"]
B --> C["2. 实现 BaseChatModel._generate()"]
C --> D["3. 配置 pyproject.toml"]
D --> E["4. 继承标准测试基类"]
E --> F["5. 运行单元测试"]
F --> G{测试通过?}
G -->|否| C
G -->|是| H["6. 添加流式支持 _stream()"]
H --> I["7. 添加工具调用 bind_tools()"]
I --> J["8. 添加结构化输出"]
J --> K["9. 运行集成测试"]
K --> L["10. 发布到 PyPI"]
17.7 Partner 架构的设计决策
独立包 vs 单一包
LangChain 最初把所有集成写在 langchain 主包里，2023-12 拆出 langchain-community（#14463），同月起又陆续把主要提供商拆成独立的 Partner 包（langchain-openai 在 2024-01，#15503）。独立包模式的好处是：
- 依赖隔离：用户只需安装用到的 SDK，不需要拖入数十个不相关的依赖
- 版本独立：OpenAI 的更新不影响 Anthropic 的版本号
- 维护责任：部分 Partner 包由服务提供商自行维护
- 安装体积：只解析和下载一家提供商的依赖树，安装明显更快
langchain-core 作为稳定锚点
Partner 包在运行时只依赖 langchain-core（langchain-deepseek、langchain-xai 额外依赖 langchain-openai，见 17.2.1 节），不依赖 langchain 主包。这意味着 langchain-core 的 API 是整个生态的稳定锚点。BaseChatModel、BaseMessage、Runnable 等核心抽象一旦定义，就不能轻易修改，否则会破坏所有 Partner 包。
这也是为什么 langchain-core 对公共 API 的改动非常谨慎，而 Partner 包可以频繁发版。
secret_from_env 模式
需要 API 密钥的 Partner 包（openai、anthropic、groq、mistralai、fireworks、deepseek、xai、perplexity、openrouter 等）都使用 secret_from_env 作为 API 密钥的默认值工厂：
api_key: SecretStr | None = Field(
default_factory=secret_from_env("OPENAI_API_KEY", default=None),
)这个模式有三层含义：
- 显式优先：直接传入的密钥值优先级最高
- 环境变量回退：未传入时自动从环境变量读取
- None 兜底：环境变量也不存在时返回 None，由运行时验证处理
标准测试作为契约
标准测试套件不仅是测试工具，更是一种契约。它定义了"一个合格的 Chat 模型应该如何行为"。通过继承测试基类，Partner 开发者无需阅读大量文档就能知道需要实现什么功能、达到什么标准。
测试的能力检测机制（检查方法是否被覆盖，或读取默认多为 False 的能力开关）也确保了向后兼容性：新增的测试用例会自动跳过不支持相关功能的模型。
flowchart LR
subgraph "契约关系"
A["langchain-core<br/>(定义抽象接口)"] -->|"BaseChatModel<br/>Embeddings<br/>BaseTool"| B["Partner 包<br/>(实现接口)"]
C["langchain-tests<br/>(定义行为标准)"] -->|"验证行为<br/>是否符合预期"| B
end
subgraph "用户视角"
D["应用代码"] -->|"统一调用<br/>model.invoke()"| B
end
17.8 ChatOpenAI 消息转换的复杂性
消息转换是 Partner 实现中最复杂的部分。ChatOpenAI 的消息转换分散在三个函数里——出方向的 _convert_message_to_dict（内容由 _format_message_content 处理）、入方向的 _convert_dict_to_message、流式的 _convert_delta_to_message_chunk——需要处理大量边界情况：
- 工具调用解析：
tool_calls可能格式错误，需要 try-catch 后放入invalid_tool_calls - 多模态内容（Content Block 翻译）：出方向上，
_format_message_content把标准数据块（is_data_content_block判定）交给 langchain-core 的convert_to_openai_data_block（langchain_core/messages/block_translators/openai.py）翻译成 OpenAI 格式，把 Anthropic 风格的image块（base64 / url 两种source）改写成image_url块（base64 拼成data:URL），并丢掉tool_use、thinking等 OpenAI 不认识的块；AIMessage在发出前还要经_compat.py的_convert_from_v1_to_chat_completions从 v1 标准块降回 Chat Completions 格式 - 角色映射：OpenAI 的
developer角色映射到 LangChain 的SystemMessage（并在additional_kwargs里记下__openai_role__） - 空值处理：
content可能是 None（纯工具调用时），需要替换为空字符串 - 流式分块：
_convert_delta_to_message_chunk把每个增量转成AIMessageChunk，工具调用片段转成带index的tool_call_chunk，再由AIMessageChunk的加法合并
这种复杂性正是 Partner 包存在的意义 -- 它将提供商特定的数据格式转换封装在一个地方，让上层应用代码完全不需要关心这些细节。
消息转换的质量直接影响了 Agent 系统的可靠性。一个不完善的消息转换可能导致工具调用参数丢失、消息角色映射错误、或者多模态内容被忽略。ChatOpenAI 的消息转换代码虽然冗长，但每一个条件分支都对应着一个真实的边界情况。
例如，invalid_tool_calls 的处理就是一个很好的例子。当模型返回的工具调用参数不是有效的 JSON 时，简单的实现可能会直接抛出异常，导致整个请求失败。而 ChatOpenAI 的实现会将无效的工具调用放入 invalid_tool_calls 列表，让上层代码有机会进行错误恢复。但这需要调用方自己检查 message.invalid_tool_calls：pin 版 langchain-classic 的工具调用 Agent 解析器 parse_ai_message_to_tool_action 并不读这个字段，一条只含无效工具调用的消息会被当成 AgentFinish(return_values={'output': ''})（实跑），handle_parsing_errors 也就无从介入。
另一个重要的细节是空值处理。OpenAI API 在模型进行工具调用时可能返回 content: null（纯工具调用，没有文本内容）。如果不将 null 替换为空字符串，AIMessage(content=None) 在构造时就会触发 Pydantic 的 ValidationError（实跑）。这种防御性的空值处理在消息转换代码中随处可见，反映了与真实 API 长期交互中积累的经验教训。
17.9 Partner 包的异步实现
现代 AI 应用框架必须支持异步操作，否则在高并发场景下会成为性能瓶颈。Partner 包的异步支持是通过 _agenerate 和 _astream 方法实现的。
BaseChatModel 提供了默认的异步实现：将同步方法放入线程池中执行。这个默认行为确保了即使 Partner 包没有实现原生异步，异步调用也不会阻塞事件循环。但线程池的开销是存在的 -- 每个异步调用会占用一个线程池线程，在高并发场景下可能成为瓶颈。
因此，成熟的 Partner 包通常会实现原生的异步方法。以 ChatOpenAI 为例，它使用 OpenAI SDK 的 AsyncOpenAI 客户端进行异步调用，完全避免了线程切换的开销。异步客户端的初始化和管理也需要特别注意 -- ChatOpenAI 在初始化校验阶段就创建好 AsyncOpenAI 客户端并保存在实例上，底层的 httpx 客户端还会按 (base_url, timeout) 用 lru_cache 在实例之间共享，避免重复创建连接池。
异步支持不仅关乎性能，还关乎与现代 Web 框架的兼容性。FastAPI、Starlette 等异步 Web 框架要求请求处理函数是协程。如果 Partner 包只支持同步调用，在这些框架中使用时就需要额外的线程池包装，增加了复杂性和延迟。原生异步支持使得 Partner 包可以直接在异步 Web 应用中使用，无需中间层。
另一个需要考虑的细节是连接管理。HTTP 连接的创建是昂贵的操作，优秀的 Partner 实现会通过 httpx 或类似的 HTTP 客户端库维护连接池。ChatOpenAI 的实现中，_get_default_httpx_client 和 _get_default_async_httpx_client 函数负责创建客户端实例：它们只设置 base_url 与 timeout，其余（连接池等）沿用 openai.DefaultHttpxClient / DefaultAsyncHttpxClient 的默认值；timeout 可哈希时走 lru_cache 缓存的实例，否则（如传入 httpx.Timeout）每次新建。
17.10 Partner 包中的流式输出实现
流式输出是现代 AI 应用的基本需求。用户不愿意等待数秒才看到回复的第一个字。在 Partner 包中，流式输出通过 _stream 方法实现。
流式方法签名
def _stream(
self,
messages: list[BaseMessage],
stop: list[str] | None = None,
run_manager: CallbackManagerForLLMRun | None = None,
**kwargs: Any,
) -> Iterator[ChatGenerationChunk]:
...实现这个方法需要处理几个关键问题。首先是分块消息的合并。每个 API 返回的分块只包含增量内容（例如一个 token 的文本），但 LangChain 需要维护一个累积的消息状态。AIMessageChunk 的 __add__ 方法负责将多个分块合并为完整的消息，包括文本内容的拼接、工具调用参数的增量组装、使用量统计的累加等。
其次是工具调用的流式处理。当模型进行工具调用时，工具名称和参数不是一次性返回的，而是以 JSON 片段的形式逐步生成。Partner 包只需把每个片段转成带 index 的 tool_call_chunk；分块相加时，langchain-core 的合并逻辑按 index 把片段拼到同一个调用上，AIMessageChunk 再用 parse_partial_json 对尚不完整的参数做尽力解析，得到随流逐步增长的 tool_calls。
最后是回调集成。每收到一个分块，_stream 方法需要通过 run_manager.on_llm_new_token 回调通知观察者。这使得 LangSmith 等追踪工具可以实时记录生成过程，前端应用可以逐字显示回复。
异步流式
除了同步的 _stream，Partner 包通常还实现 _astream 异步版本。异步流式在 Web 应用中尤为重要，它不会阻塞事件循环，可以同时处理多个用户的流式请求。大部分提供商的 SDK 都原生支持异步迭代器，Partner 包只需做格式转换。
17.11 Partner 包的错误处理与重试
在生产环境中，API 调用不可避免地会遇到各种错误：网络超时、速率限制、服务暂时不可用等。Partner 包需要提供健壮的错误处理。
重试机制
大多数 Partner 包通过提供商 SDK 内置的重试机制来处理临时性错误。例如 ChatOpenAI 的 max_retries 参数会传递给 OpenAI SDK，由 SDK 负责指数退避重试。LangChain 额外提供了 with_retry() 方法，可以在 Runnable 层面添加重试：
model_with_retry = ChatOpenAI(model="gpt-4").with_retry(
stop_after_attempt=3,
wait_exponential_jitter=True,
)上下文窗口溢出
一个特别值得关注的错误类型是上下文窗口溢出。langchain_core.exceptions 定义了统一的 ContextOverflowError，而 ChatOpenAI 在 _handle_openai_bad_request / _handle_openai_api_error 里把 OpenAI SDK 抛出的原始异常翻译过来：当报错文本里出现 context_length_exceeded、Input tokens exceed the configured limit 或 exceeds the context window 时，重新抛出 OpenAIContextOverflowError（同时继承 openai.BadRequestError 和 ContextOverflowError）或 OpenAIAPIContextOverflowError。这样上层代码可以只 catch 与提供商无关的 ContextOverflowError，而不必去匹配各家 SDK 的错误字符串；同时因为它仍然是原始 openai 异常的子类，已有的 except openai.BadRequestError 代码不会被破坏。ChatOpenAI 另外用 tiktoken 提供 get_num_tokens 这类客户端计数能力，但溢出判定本身来自服务端返回。
速率限制处理
速率限制（Rate Limit）是使用 AI API 时最常遇到的问题之一。LangChain 的 Runnable 层面提供了 max_concurrency 配置来控制并发数量，但真正的速率限制处理通常依赖提供商 SDK 的内置机制。主动限速则在 langchain-core 一侧：BaseChatModel 有 rate_limiter 字段，可以传入基于令牌桶算法的 InMemoryRateLimiter（langchain_core.rate_limiters）来平滑请求速率，Partner 包本身不需要各自实现。
17.12 ChatOpenAI 的模型配置系统
ChatOpenAI 接入了 langchain-core 的模型配置（profile）系统，通过 ModelProfile 和 ModelProfileRegistry 管理不同模型的能力信息。
# langchain_core/language_models/model_profile.py
class ModelProfile(TypedDict, total=False):
max_input_tokens: int # 上下文窗口
max_output_tokens: int # 最大输出
text_inputs: bool
image_inputs: bool
pdf_inputs: bool
reasoning_output: bool
tool_calling: bool # 是否支持工具调用
tool_choice: bool # 是否支持强制工具选择
structured_output: bool # 是否有原生结构化输出
temperature: bool # 是否接受 temperature 参数
# ... 还有 name / release_date / open_weights 等元数据
ModelProfileRegistry = dict[str, ModelProfile]
# langchain_openai/data/_profiles.py（自动生成，数据源 models.dev）
_PROFILES: dict[str, dict[str, Any]] = {
# ...
"codex-mini-latest": {
"name": "Codex Mini",
"release_date": "2025-05-16",
"max_input_tokens": 200000,
"max_output_tokens": 100000,
# ...
},
# ...
}ModelProfile 本身定义在 langchain_core.language_models.model_profile（标注为 beta，格式可能变动），BaseChatModel 上有一个 profile: ModelProfile | None 字段，由模型校验器调用 _resolve_model_profile() 填充。Partner 包只需覆写 _resolve_model_profile -- ChatOpenAI 的实现就是拿 self.model_name 去查 _PROFILES。_profiles.py 是由工具从 models.dev 的数据自动生成的，文件头明确写着"DO NOT EDIT THIS FILE MANUALLY"，这意味着模型能力表的维护成本被外部化了，不需要人工跟踪每个新模型。
需要澄清的是，在当前版本里 profile 是"信息暴露"而非"强制校验"：bind_tools 并不会读 profile 去拦截不支持工具调用的模型，langchain-tests 也还没有用 profile 来决定跳过哪些用例 -- 标准测试用的仍然是前面讲的方法覆盖检测和签名检查。profile 的价值目前主要在于让上层应用（如中间件、路由器）可以在调用前查询上下文窗口大小和能力开关。
17.13 从 langchain-community 到独立 Partner 包的迁移
LangChain 的集成生态经历了从集中式到分布式的重大迁移。早期所有集成都放在 langchain 主包中，2023-12 拆出后集中在 langchain-community 包里，它要为上百个第三方服务提供适配代码，虽然大部分 SDK 被声明为可选依赖（extras），但依赖解析的复杂度和版本冲突的概率都随之上升。
迁移到独立 Partner 包后，每个包只包含一个提供商的代码和依赖。langchain-community 本身也在 2025-04 迁出主仓库，改在独立的 langchain-ai/langchain-community 仓库维护（#31060），所以 pin 树的 libs/ 下已经没有它。这个迁移过程本身就是一个巨大的工程挑战，因为需要保持序列化的向后兼容性 -- 旧的序列化数据中的类路径指向 langchain.chat_models.openai，现在实际类在 langchain_openai.chat_models.base。
这就是第 16 章讨论的 SERIALIZABLE_MAPPING 映射表的核心用途。映射表记录了从旧路径到新路径的对应关系，使得迁移对用户透明。Reviver 在反序列化时自动将旧路径重定向到新路径，无需用户修改任何持久化的数据。
这次迁移的经验也影响了新 Partner 包的设计规范：get_lc_namespace 方法应该返回基于实际模块路径的命名空间（而非手动构造的"兼容路径"），以避免未来再次迁移时的额外工作。
小结
本章详细剖析了 LangChain 的 Partner 集成架构。Partner 包是 LangChain 生态的核心扩展点，它们通过实现 langchain-core 定义的抽象接口，将各种 AI 服务标准化地接入 LangChain 体系。
以 langchain-openai 为例，我们深入分析了一个成熟 Partner 包的方方面面：从 pyproject.toml 的依赖配置和版本管理，到 ChatOpenAI 的 _generate 和 _stream 方法实现，再到 bind_tools 的工具绑定机制和模型配置系统。密钥管理通过 lc_secrets 和 secret_from_env 实现了安全与便利的平衡。错误处理和重试策略确保了生产环境的可靠性。
标准测试套件 langchain-tests 是这个架构的验证层。通过继承测试基类、声明模型能力，Partner 开发者可以快速验证自己的实现是否符合 LangChain 的行为契约。从 langchain-community 到独立 Partner 包的迁移经验则展示了如何在大规模重构中保持向后兼容性。
下一章是本专栏的最后一章，我们将从 LangChain 的具体实现中抽身出来，总结那些可迁移的设计模式和架构决策，为你构建自己的 AI 应用框架提供指引。