语言模型 API
BaseLanguageModel 与 LLM 统一接口
概述
LangChain 的语言模型模块提供了与各种 LLM 提供商交互的统一接口。所有语言模型都继承自 BaseLanguageModel 基类,提供一致的 API。
graph TD
A[BaseLanguageModel] --> B[LLM]
A --> C[BaseChatModel]
B --> D[OpenAI]
B --> E[Anthropic]
B --> F[Hugging Face]
C --> G[ChatOpenAI]
C --> H[ChatAnthropic]
C --> I[ChatGoogleGenerativeAI]
style A fill:#e1f5fe
style C fill:#c8e6c9
核心类
BaseLanguageModel
所有语言模型的抽象基类,定义了统一接口规范。
class BaseLanguageModel(Serializable, Runnable[LanguageModelInput, LanguageModelOutput], Generic[LanguageModelInput, LanguageModelOutput]):
"""语言模型基类"""
@property
@abstractmethod
def _llm_type(self) -> str:
"""返回模型类型标识"""
@abstractmethod
def _generate(
self,
prompts: List[str],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> LLMResult:
"""生成响应的核心方法"""
def generate_prompt(
self,
prompts: List[PromptValue],
stop: Optional[List[str]] = None,
callbacks: Callbacks = None,
**kwargs: Any,
) -> LLMResult:
"""基于 PromptValue 生成响应"""
async def agenerate_prompt(
self,
prompts: List[PromptValue],
stop: Optional[List[str]] = None,
callbacks: Callbacks = None,
**kwargs: Any,
) -> LLMResult:
"""异步生成响应"""
属性
| 属性 | 类型 | 说明 |
|---|---|---|
| _llm_type | str | 模型类型标识(如 "openai", "anthropic") |
相关链接
LLM
纯文本语言模型基类,处理字符串输入输出。
class LLM(BaseLanguageModel[str, str]):
"""纯文本语言模型"""
def __call__(self, text: str) -> str:
"""调用模型的便捷方法"""
def predict(self, text: str, **kwargs: Any) -> str:
"""预测文本(兼容方法)"""
def predict_messages(
self,
messages: List[BaseMessage],
**kwargs: Any
) -> BaseMessage:
"""预测消息响应"""
常用实现类
OpenAI- OpenAI GPT 模型Anthropic- Anthropic Claude 模型HuggingFacePipeline- Hugging Face 模型
核心方法
invoke
同步调用模型生成响应。
def invoke(
self,
input: LanguageModelInput,
config: Optional[RunnableConfig] = None,
stop: Optional[List[str]] = None,
**kwargs: Any,
) -> LanguageModelOutput:
"""
调用模型
Args:
input: 输入文本或消息列表
config: 运行配置(包含 callbacks, metadata 等)
stop: 停止词列表
**kwargs: 额外参数
Returns:
模型输出(字符串或 AIMessage)
"""
参数说明
| 参数 | 类型 | 说明 |
|---|---|---|
| input | str | List[str] | 输入提示文本 |
| config | RunnableConfig | None | 运行配置,可包含 callbacks、tags、metadata |
| stop | List[str] | None | 停止生成词列表 |
| **kwargs | Any | 模型特定参数(如 temperature, max_tokens) |
返回值
| 类型 | 说明 |
|---|---|
| str | 生成的文本响应 |
stream
流式调用模型,逐 token 返回结果。
def stream(
self,
input: LanguageModelInput,
config: Optional[RunnableConfig] = None,
stop: Optional[List[str]] = None,
**kwargs: Any,
) -> Iterator[LanguageModelOutput]:
"""
流式调用模型
Args:
input: 输入文本
config: 运行配置
stop: 停止词
**kwargs: 额外参数
Yields:
逐 token 返回的输出
"""
返回值
| 类型 | 说明 |
|---|---|
| Iterator[str] | 文本块的迭代器 |
batch
批量处理多个输入。
def batch(
self,
inputs: List[LanguageModelInput],
config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None,
stop: Optional[List[str]] = None,
**kwargs: Any,
) -> List[LanguageModelOutput]:
"""
批量调用模型
Args:
inputs: 输入列表
config: 配置(单个或多个)
stop: 停止词
**kwargs: 额外参数
Returns:
输出列表
"""
ainvoke / astream / abatch
异步版本的调用方法。
async def ainvoke(
self,
input: LanguageModelInput,
config: Optional[RunnableConfig] = None,
stop: Optional[List[str]] = None,
**kwargs: Any,
) -> LanguageModelOutput:
"""异步调用模型"""
async def astream(
self,
input: LanguageModelInput,
config: Optional[RunnableConfig] = None,
stop: Optional[List[str]] = None,
**kwargs: Any,
) -> AsyncIterator[LanguageModelOutput]:
"""异步流式调用"""
async def abatch(
self,
inputs: List[LanguageModelInput],
config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None,
**kwargs: Any,
) -> List[LanguageModelOutput]:
"""异步批量调用"""
bind
绑定参数到模型,创建新的可运行对象。
def bind(
self,
**kwargs: Any,
) -> Runnable[LanguageModelInput, LanguageModelOutput]:
"""
绑定参数
Args:
**kwargs: 要绑定的参数(如 temperature=0.7)
Returns:
绑定参数后的新 Runnable
"""
示例
python
from langchain_openai import OpenAI
llm = OpenAI(model="gpt-3.5-turbo-instruct")
cool_llm = llm.bind(temperature=0.3)
creative_llm = llm.bind(temperature=0.9)
with_types
配置输入输出类型。
def with_types(
self,
input_type: Optional[Type[LanguageModelInput]] = None,
output_type: Optional[Type[LanguageModelOutput]] = None,
) -> Runnable[LanguageModelInput, LanguageModelOutput]:
"""
配置类型
Args:
input_type: 输入类型
output_type: 输出类型
Returns:
配置类型后的 Runnable
"""
get_num_tokens
计算文本的 token 数量。
def get_num_tokens(self, text: str) -> int:
"""
计算 token 数量
Args:
text: 要计算的文本
Returns:
token 数量
"""
get_num_tokens_from_messages
计算消息列表的 token 数量。
def get_num_tokens_from_messages(
self,
messages: List[BaseMessage],
) -> int:
"""
计算消息的 token 数量
Args:
messages: 消息列表
Returns:
token 总数
"""
使用示例
python
from langchain_openai import OpenAI
# 初始化模型
llm = OpenAI(
model="gpt-3.5-turbo-instruct",
temperature=0.7,
max_tokens=1000,
api_key="your-api-key"
)
# 基础调用
response = llm.invoke("讲一个关于 Python 的故事")
print(response)
# 流式调用
for chunk in llm.stream("用三句话介绍人工智能"):
print(chunk, end="", flush=True)
# 批量调用
responses = llm.batch([
"什么是 Python?",
"什么是 JavaScript?",
"什么是 Rust?"
])
# 绑定参数
cool_llm = llm.bind(temperature=0.1)
response = cool_llm.invoke("写一首关于编程的诗")
# 带配置的调用
response = llm.invoke(
"写一首关于春天的诗",
config={
"tags": ["creative", "poetry"],
"metadata": {"user_id": "user_123"}
}
)
# 异步调用
import asyncio
async def async_example():
response = await llm.ainvoke("你好")
print(response)
async for chunk in llm.astream("介绍 LangChain"):
print(chunk, end="", flush=True)
asyncio.run(async_example())
相关 API
- Chat Models API - 聊天模型接口
- Prompts API - 提示模板
- Runnables API - Runnable 接口