本文介绍如何使用 LangChain 框架自定义封装大语言模型接口。核心在于继承 LLM 基类并实现_call 与_identifying_params 方法。通过讯飞星火 API 示例，展示了如何通过 WebSocket 进行鉴权、发送请求及接收流式响应，并将其封装为 LangChain 兼容的 LLM 实例。该方法支持灵活切换不同模型，无需修改上层应用代码，适用于文本生成、翻译等任务。同时强调了生产环境中 API Key 的安全管理策略，包括环境变量使用和异常处理机制，确保应用稳定性。
在构建基于大语言模型（LLM）的应用时，直接调用厂商 API 往往会导致代码耦合度高、切换模型困难以及性能损耗等问题。LangChain 框架提供了一套标准化的抽象接口，允许开发者将任意大模型封装为统一的 LLM 对象，从而实现业务逻辑与底层模型的解耦。
LLM（Large Language Model）是 LangChain 平台与各种大模型进行交互的核心抽象类。它定义了一个黑盒接口，输入为字符串，输出也为字符串。通过继承 langchain.llms.base.LLM 基类，开发者可以隐藏不同模型 API 的差异（如 HTTP 请求、WebSocket 连接、鉴权方式等），专注于提示词工程和应用逻辑。
要实现一个自定义 LLM，主要需要实现以下两个核心部分：
_call 方法：这是核心逻辑入口。接收用户输入的字符串（prompt），返回模型生成的文本。支持可选的停止词（stop words）和回调管理器（CallbackManager）。_identifying_params 属性：用于描述该 LLM 实例的配置参数，便于调试和日志记录。此外，通常还需要实现 _llm_type 属性以标识模型类型。
为了演示结构，我们首先实现一个仅返回输入前 N 个字符的简易 LLM：
from langchain.llm import LLM
from typing import Optional, List, Mapping, Any
class CustomLLM(LLM):
n: int = 10 # 返回字符数
@property
def _llm_type(self) -> str:
return "custom_simple"
def _call(
self,
prompt: str,
stop: Optional[List[str]] = None,
run_manager=None,
**kwargs: Any,
) -> str:
# 模拟处理逻辑
if stop:
for word in stop:
prompt = prompt.replace(word, "")
return prompt[:self.n]
@property
def _identifying_params(self) -> Mapping[str, Any]:
return {"n": self.n}
在实际生产中，我们需要对接真实的 WebSocket 或 HTTP 接口。以下以讯飞星火大模型为例，展示如何封装完整的鉴权与通信逻辑。
pip install websocket-client requests讯飞星火使用 WebSocket 协议，需按照官方文档生成签名（Signature）。以下是封装后的完整代码示例，整合了鉴权、连接建立及消息处理：
import json
import ssl
import time
import hashlib
import hmac
import base64
from urllib.parse import urlencode, urlparse
from datetime import datetime
from wsgiref.handlers import format_date_time
from threading import Thread
from typing import Optional, List, Mapping, Any
from langchain.callbacks.manager import CallbackManagerForLLMRun
from langchain.llms.base import LLM
import websocket
class SparkLLM(LLM):
appid: Optional[str] = None
api_secret: Optional[str] = None
api_key: Optional[str] = None
model: Optional[str] = None
answer: str = ""
@property
def _llm_type(self) -> str:
return "SparkLLM"
def _call(
self,
prompt: str,
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> str:
# 重置答案
self.answer = ""
# 配置模型版本与地址
modal_dict = {
"general": "ws://spark-api.xf-yun.com/v1.1/chat",
"generalv2": "ws://spark-api.xf-yun.com/v2.1/chat",
"generalv3": "ws://spark-api.xf-yun.com/v3.1/chat"
}
domain = self.model or "generalv3"
spark_url = modal_dict.get(domain, modal_dict["generalv3"])
try:
wsParam = Ws_Param(self.appid, self.api_key, self.api_secret, spark_url)
wsUrl = wsParam.create_url()
ws = websocket.WebSocketApp(
wsUrl,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=lambda ws: self.run(ws, prompt, domain)
)
ws.appid = self.appid
ws.question = prompt
ws.domain = domain
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
except Exception as e:
return f"Error occurred during connection: {str(e)}"
return self.answer
def on_message(self, ws, message):
data = json.loads(message)
code = data['header']['code']
if code != 0:
print(f'Request error: {code}, {data}')
ws.close()
else:
choices = data["payload"]["choices"]
status = choices["status"]
content = choices["text"][0]["content"]
self.answer += content
if status == 2:
ws.close()
def on_error(self, ws, error):
print(f"WebSocket Error: {error}")
def on_close(self, ws, one, two):
pass
def run(self, ws, question, domain):
data = json.dumps(self.gen_params(ws.appid, domain, question))
ws.send(data)
def gen_params(self, appid, domain, question):
return {
"header": {"app_id": appid, "uid": "user_123"},
"parameter": {
"chat": {
"domain": domain,
"temperature": 0.5,
"max_tokens": 2048
}
},
"payload": {"message": {"text": question}}
}
在生产环境中，严禁将 API Key 硬编码在代码中。建议采取以下措施：
.env 文件存储密钥，并通过 os.getenv() 读取。import os
# 推荐用法
xh_app_id = os.getenv("IFLYTEK_APP_ID")
xh_api_secret = os.getenv("IFLYTEK_API_SECRET")
xh_api_key = os.getenv("IFLYTEK_API_KEY")
if not all([xh_app_id, xh_api_secret, xh_api_key]):
raise ValueError("Missing environment variables")
modal = "generalv3"
llm = SparkLLM(appid=xh_app_id, api_secret=xh_api_secret, api_key=xh_api_key, model=modal)
response = llm.invoke("你好，请介绍一下 LangChain")
print(response)
本文详细介绍了如何使用 LangChain 框架自定义封装大语言模型接口。通过继承 LLM 基类并实现 _call 方法，开发者可以将讯飞星火等第三方模型无缝集成到 LangChain 生态中。这种方法不仅降低了模型切换的成本，还提升了系统的可维护性和安全性。在实际应用中，结合环境变量管理和完善的错误处理机制，可以构建出稳定可靠的大模型应用系统。
对于其他模型（如百度文心一言、阿里通义千问），只需调整鉴权逻辑和请求格式，复用相同的封装模式即可。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online