前言
智谱 AI 推出了新一代 LLM-GLM4，随之发布了新版本 API SDK v4。在构建智能摘要平台时，除了接入 OpenAI，接入国产大模型 GLM 也是降低 Token 成本的重要方案。目前大模型应用开发框架中，LangChain 是最热门的选择之一。LangChain 经过一年多的开发，已划分为 LangChain、LangChain-Community 和 LangChain-Core 三个部分。第三方实现主要集中在 LangChain-Community 库中。
一、问题分析
根据文档编写测试代码操作：
from langchain_community.chat_models import ChatZhipuAI
model = ChatZhipuAI(
model="chatglm_turbo",
api_key="xxxx",
)
print(model.invoke("hello, what today is today?"))
测试发现报错，提示缺少 model_api 参数。联想到 Zhipu AI 发布了新的 API v4 版本新 SDK，猜测应该是 LangChain-Community 上面的 ChatZhipuAI 适配的是 v3 版本的 SDK。
查看 LangChain 源码：
# file: langchain_community/chat_models/zhipuai.py
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
try:
import zhipuai
self.zhipuai = zhipuai
self.zhipuai.api_key = self.zhipuai_api_key
except ImportError:
raise RuntimeError(
"Could not import zhipuai package. "
"Please install it via 'pip install zhipuai'"
)
这里发现导入 zhipuai 这个库，查看 PyPi 文档，新版本 zhipuai 最新是 2.0.1 版本，调用方法已经发生了变化，明显是 2.0 版本不再兼容之前的 1.0 版本了。所以 LangChain 上的 ChatZhipuAI 不能用了，要想使用只能自己动手适配。
二、适配实现
LangChain 框架对底层的 LLM 已经封装好了，我们自定义的 ChatModel 只要继承 BaseChatModel，实现相关抽象方法即可。
class BaseChatModel(BaseLanguageModel[BaseMessage], ABC):
@abstractmethod
def _generate(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> ChatResult:
"""Top Level call"""
@property
@abstractmethod
def _llm_type(self) -> str:
"""Return type of chat model."""
参考 LangChain 很多已实现的库，如 ChatOpenAI，实现我们的 ChatZhipuAI。
1. 初始化客户端
在 __init__ 的时候，增加 V2 版本的判断，确保使用的是新版 SDK。
class ChatZhipuAI(BaseChatModel):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
try:
from zhipuai import ZhipuAI
if not is_zhipu_v2():
raise RuntimeError(
"zhipuai package version is too low"
"Please install it via 'pip install --upgrade zhipuai'"
)
self.client = ZhipuAI(
api_key=self.zhipuai_api_key,
)
except ImportError:
raise RuntimeError(
"Could not import zhipuai package. "
"Please install it via 'pip install zhipuai'"
)
2. 实现生成逻辑
实现 _generate 方法。LangChain 支持同步异步、流式、批量等多种操作。我们需要根据 zhipuai SDK 说明 和 LangChain 库一一实现。
def _generate(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
stream: Optional[bool] = None,
**kwargs: Any,
) -> ChatResult:
should_stream = stream if stream is not None else self.streaming
if should_stream:
stream_iter = self._stream(
messages, stop=stop, run_manager=run_manager, **kwargs
)
return generate_from_stream(stream_iter)
message_dicts, params = self._create_message_dicts(messages, stop)
params = {
**params,
**({"stream": stream} if stream is not None else {}),
**kwargs,
}
response = self.completion_with_retry(
messages=message_dicts, run_manager=run_manager, **params
)
return self._create_chat_result(response)
3. 消息格式转换
需要实现 _create_message_dicts 方法，将 LangChain 的 BaseMessage 转换为 ZhipuAI 所需的字典格式。这包括处理 System、User、Assistant 等不同角色。
def _create_message_dicts(
self, messages: List[BaseMessage], stop: Optional[List[str]]
) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
message_dicts = []
for m in messages:
role = "user"
if isinstance(m, HumanMessage):
role = "user"
elif isinstance(m, AIMessage):
role = "assistant"
elif isinstance(m, SystemMessage):
role = "system"
# 处理 content 为字符串或列表的情况
content = m.content
if isinstance(content, list):
content = str(content)
message_dicts.append({
"role": role,
"content": content
})
params = {"temperature": self.temperature}
if stop:
params["stop"] = stop
return message_dicts, params
4. 结果解析与重试机制
实现 _create_chat_result 方法，将 API 响应转换为 LangChain 的 ChatResult。同时建议添加简单的重试逻辑以应对网络波动。
from tenacity import retry, stop_after_attempt, wait_exponential
def completion_with_retry(self, **kwargs):
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=4))
def _call():
return self.client.chat.completions.create(**kwargs)
return _call()
def _create_chat_result(self, response: Any) -> ChatResult:
choice = response.choices[0]
content = choice.message.content
finish_reason = choice.finish_reason
generation_info = {"finish_reason": finish_reason}
message = AIMessage(content=content, generation_info=generation_info)
return ChatResult(generations=[[ChatGeneration(message=message)]])
三、依赖冲突处理
在提交进行流水线检查时，可能会发现单元测试没过，遇到版本冲突。
Because no versions of zhipuai match >2.0.1,<3.0.0 and zhipuai (2.0.1) depends on pydantic (>=2.5.2),
zhipuai (>=2.0.1,<3.0.0) requires pydantic (>=2.5.2). And because javelin-sdk (0.1.8) depends on
pydantic (>=1.10.7,<2.0.0)...
这是由于 zhipuai 依赖的 pydantic 版本太高和 LangChain 中集成的三方库有版本依赖冲突。LangChain 推荐采用 pydantic 1.0+。如果遇到此问题，建议优先升级 zhipuai 到最新版，或者在本地项目中隔离依赖环境，使用虚拟环境单独管理。
四、完整使用示例
以下是一个完整的集成脚本示例，展示了如何实例化并调用模型。
from langchain_core.messages import HumanMessage
from your_custom_chat_model import ChatZhipuAI
zhipuai_chat = ChatZhipuAI(
temperature=0.5,
api_key="your-api-key",
model_name="glm-3-turbo",
)
messages = [
HumanMessage(content="你好，请介绍一下你自己")
]
response = zhipuai_chat.invoke(messages)
print(response.content)
五、总结
截止发文前，版本依赖冲突还没有完全解决，观察后续的进展，有需要的同学可以自己先自定义本地集成进项目。通过上述步骤，可以成功将 LangChain 适配至最新的 ZhipuAI v2 SDK，实现国产大模型的无缝集成。开发者在集成过程中需注意 SDK 版本兼容性，并根据实际业务需求调整温度、停止符等参数。
