聊天模型 API
BaseChatModel 与聊天模型实现
概述
聊天模型（Chat Models）是 LangChain 中最常用的模型类型，它们使用消息列表作为输入和输出，支持多轮对话。
graph TD
A[BaseChatModel] --> B[ChatOpenAI]
A --> C[ChatAnthropic]
A --> D[ChatGoogleGenerativeAI]
A --> E[AzureChatOpenAI]
A --> F[ChatBedrock]
B --> G[模型提供商]
C --> G
D --> G
E --> G
F --> G
style A fill:#e1f5fe
style G fill:#c8e6c9
核心类
BaseChatModel
所有聊天模型的抽象基类。
class BaseChatModel(BaseLanguageModel[List[BaseMessage], BaseMessage]):
"""聊天模型基类"""
@property
@abstractmethod
def _llm_type(self) -> str:
"""返回模型类型"""
@abstractmethod
def _generate(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> ChatResult:
"""生成响应的核心方法"""
def bind_tools(
self,
tools: Sequence[Union[StructuredTool, dict]],
**kwargs: Any,
) -> Runnable[LanguageModelInput, BaseMessage]:
"""绑定工具到模型"""
def with_structured_output(
self,
schema: Union[Dict, Type[BaseModel]],
*,
method: Literal["tool_calling", "function_calling"] = "tool_calling",
**kwargs: Any,
) -> Runnable[LanguageModelInput, Union[Dict, BaseModel]]:
"""配置结构化输出"""核心属性
| 属性 | 类型 | 说明 |
|---|---|---|
| bind_tools | method | 绑定工具到模型 |
| with_structured_output | method | 配置结构化输出 |
| with_retry | method | 添加重试逻辑 |
核心方法
invoke
调用聊天模型。
def invoke(
self,
input: List[BaseMessage],
config: Optional[RunnableConfig] = None,
stop: Optional[List[str]] = None,
**kwargs: Any,
) -> BaseMessage:
"""
调用聊天模型
Args:
input: 消息列表
config: 运行配置
stop: 停止词
**kwargs: 额外参数
Returns:
AIMessage 响应
"""参数说明
| 参数 | 类型 | 说明 |
|---|---|---|
| input | List[BaseMessage] | 消息列表（HumanMessage, SystemMessage, AIMessage） |
| config | RunnableConfig | 运行配置（callbacks, metadata, tags） |
| stop | List[str] | 停止词列表 |
| **kwargs | Any | 模型特定参数（temperature, max_tokens 等） |
stream
流式调用聊天模型。
def stream(
self,
input: List[BaseMessage],
config: Optional[RunnableConfig] = None,
stop: Optional[List[str]] = None,
**kwargs: Any,
) -> Iterator[BaseMessageChunk]:
"""
流式调用
Yields:
AIMessageChunk 迭代器
"""bind_tools
绑定工具到模型，使模型能够调用工具。
def bind_tools(
self,
tools: Sequence[Union[StructuredTool, dict]],
*,
tool_choice: Optional[Union[dict, str, Literal["auto", "none", "required"]]] = None,
**kwargs: Any,
) -> Runnable[List[BaseMessage], BaseMessage]:
"""
绑定工具
Args:
tools: 工具列表
tool_choice: 工具选择策略
- "auto": 模型决定是否调用工具
- "none": 不调用工具
- "required": 必须调用工具
- dict/str: 指定特定工具
**kwargs: 额外参数
Returns:
绑定工具后的 Runnable
"""with_structured_output
配置模型输出结构化数据。
def with_structured_output(
self,
schema: Union[Dict, Type[BaseModel]],
*,
method: Literal["tool_calling", "function_calling"] = "tool_calling",
**kwargs: Any,
) -> Runnable[List[BaseMessage], Union[Dict, BaseModel]]:
"""
配置结构化输出
Args:
schema: 输出结构（Pydantic 模型或字典）
method: 方法类型
- "tool_calling": 使用工具调用
- "function_calling": 使用函数调用
**kwargs: 额外参数
Returns:
返回结构化数据的 Runnable
"""示例
python
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
class Response(BaseModel):
answer: str
confidence: float
llm = ChatOpenAI(model="gpt-4o")
structured_llm = llm.with_structured_output(Response)
result = structured_llm.invoke([
("system", "你是一个助手"),
("user", "巴黎是法国的首都吗？")
])
# result: Response(answer="是的，巴黎是法国的首都", confidence=0.95)with_retry
添加自动重试逻辑。
def with_retry(
self,
*,
retry_if_exception_type: Tuple[Type[BaseException], ...] = (Exception,),
max_retries: int = 3,
wait_exponential_multiplier: float = 1.0,
wait_exponential_max: float = 10.0,
) -> Runnable[LanguageModelInput, LanguageModelOutput]:
"""
添加重试
Args:
retry_if_exception_type: 需要重试的异常类型
max_retries: 最大重试次数
wait_exponential_multiplier: 指数退避乘数
wait_exponential_max: 最大等待时间
Returns:
带重试的 Runnable
"""实现类
ChatOpenAI
OpenAI 聊天模型实现。
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model: str = "gpt-4o",
temperature: float = 0.7,
max_tokens: Optional[int] = None,
api_key: Optional[str] = None,
base_url: Optional[str] = None,
organization: Optional[str] = None,
timeout: Optional[float] = None,
max_retries: int = 2,
streaming: bool = False,
n: int = 1,
top_p: float = 1.0,
presence_penalty: float = 0.0,
frequency_penalty: float = 0.0,
)参数说明
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| model | str | gpt-4o | 模型名称 |
| temperature | float | 0.7 | 采样温度 (0-2) |
| max_tokens | int | None | None | 最大生成 token 数 |
| api_key | str | None | None | API 密钥 |
| base_url | str | None | None | API 基础 URL |
| timeout | float | None | None | 请求超时（秒） |
| max_retries | int | 2 | 最大重试次数 |
| streaming | bool | False | 是否启用流式 |
| n | int | 1 | 生成候选数 |
| top_p | float | 1.0 | 核采样参数 |
| presence_penalty | float | 0.0 | 存在惩罚 (-2 到 2) |
| frequency_penalty | float | 0.0 | 频率惩罚 (-2 到 2) |
ChatAnthropic
Anthropic Claude 聊天模型实现。
from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic(
model: str = "claude-3-5-sonnet-20241022",
temperature: float = 0.7,
max_tokens: int = 1024,
api_key: Optional[str] = None,
timeout: Optional[float] = None,
max_retries: int = 2,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
streaming: bool = False,
)Claude 模型
- claude-3-5-sonnet-20241022 - 最新 Sonnet
- claude-3-5-sonnet-20240620 - Sonnet 3.5
- claude-3-opus-20240229 - Opus 3
- claude-3-sonnet-20240229 - Sonnet 3
- claude-3-haiku-20240307 - Haiku 3
ChatGoogleGenerativeAI
Google Gemini 聊天模型实现。
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(
model: str = "gemini-1.5-pro",
temperature: float = 0.7,
max_tokens: Optional[int] = None,
api_key: Optional[str] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
)Gemini 模型
- gemini-1.5-pro - Gemini 1.5 Pro
- gemini-1.5-flash - Gemini 1.5 Flash
- gemini-pro - Gemini Pro
AzureChatOpenAI
Azure OpenAI 聊天模型实现。
from langchain_openai import AzureChatOpenAI
llm = AzureChatOpenAI(
azure_deployment: str,
openai_api_version: str,
azure_endpoint: str,
api_key: Optional[str] = None,
azure_ad_token: Optional[str] = None,
temperature: float = 0.7,
max_tokens: Optional[int] = None,
)ChatBedrock
AWS Bedrock 聊天模型实现。
from langchain_aws import ChatBedrock
llm = ChatBedrock(
model_id: str = "anthropic.claude-3-5-sonnet-20241022-v2:0",
region_name: str = "us-east-1",
credentials_profile_name: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 1024,
)使用示例
python
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
# 初始化
llm = ChatOpenAI(model="gpt-4o")
# 基础调用
response = llm.invoke([
SystemMessage(content="你是一个有帮助的助手"),
HumanMessage(content="你好！")
])
print(response.content)
# 流式调用
for chunk in llm.stream([HumanMessage(content="讲一个故事")]):
print(chunk.content, end="", flush=True)
# 绑定工具
from langchain_core.tools import tool
@tool
def get_weather(location: str) -> str:
"""获取指定地点的天气"""
return f"{location} 今天是晴天，温度 25°C"
llm_with_tools = llm.bind_tools([get_weather])
response = llm_with_tools.invoke([HumanMessage(content="北京的天气怎么样？")])
# 结构化输出
from pydantic import BaseModel
class Answer(BaseModel):
summary: str
confidence: float
structured_llm = llm.with_structured_output(Answer)
result = structured_llm.invoke([HumanMessage(content="总结 LangChain")])
# 批量处理
messages = [
[HumanMessage(content="什么是 Python?")],
[HumanMessage(content="什么是 JavaScript?")],
[HumanMessage(content="什么是 Rust?")]
]
responses = llm.batch(messages)
# 异步调用
import asyncio
async def async_chat():
response = await llm.ainvoke([
HumanMessage(content="你好")
])
print(response.content)
async for chunk in llm.astream([HumanMessage(content="介绍自己")]):
print(chunk.content, end="", flush=True)
asyncio.run(async_chat())