【LangChain】langchain_openai.ChatOpenAI 类：基于 OpenAI 的聊天完成 API（Chat Completions API）

langchain_openai.ChatOpenAI 是 LangChain 中用于调用 OpenAI 对话模型（如 gpt-3.5-turbo、gpt-4）的类，基于 OpenAI 的聊天完成 API（Chat Completions API）。它继承自 BaseChatModel，支持结构化的消息输入（如 SystemMessage、HumanMessage），适合对话、问答和复杂任务。初始

正义的彬彬侠

2496人浏览 · 2025-05-08 11:48:49

正义的彬彬侠 · 2025-05-08 11:48:49 发布

langchain_openai.ChatOpenAI 类是 LangChain 中用于与 OpenAI 对话模型（chat models）交互的类，继承自 langchain_core.language_models.chat_models.BaseChatModel，而非直接继承 BaseLLM。

本文基于 LangChain 0.3.x ，详细介绍 ChatOpenAI 的定义、初始化、常用方法和属性，并提供一个独立示例，展示如何使用 ChatOpenAI 调用 gpt-3.5-turbo 模型处理人工智能主题的对话，示例结合 ChatPromptTemplate 和 invoke 方法。

`langchain_openai.ChatOpenAI` 简介

langchain_openai.ChatOpenAI 是 LangChain 中用于调用 OpenAI 对话模型（如 gpt-3.5-turbo、gpt-4）的类，基于 OpenAI 的聊天完成 API（Chat Completions API）。它继承自 BaseChatModel，支持结构化的消息输入（如 SystemMessage、HumanMessage），适合对话、问答和复杂任务。

核心功能：

调用 OpenAI 聊天完成 API，生成对话响应。
支持同步（invoke、generate）、异步（ainvoke、agenerate）和流式（stream、astream）生成。
处理结构化消息，支持系统提示、用户输入和对话历史。
配置 OpenAI 特定的参数（如 temperature、max_tokens）。
集成到 LCEL 链，适合 RAG、对话系统或多轮交互。

适用场景：

构建基于 OpenAI 的对话或问答系统。
处理多轮对话，维护上下文。
集成到生产级应用，利用 OpenAI 的高性能对话模型。
开发需要结构化输入的 LLM 应用（如客服、助手）。

与 langchain_openai.OpenAI 对比：

OpenAI：用于完成模型（如 gpt-3.5-turbo-instruct），处理原始文本提示，返回文本。
ChatOpenAI：用于对话模型（如 gpt-3.5-turbo、gpt-4），支持结构化消息（SystemMessage、HumanMessage），更适合对话场景。

类定义和初始化

以下是 langchain_openai.ChatOpenAI 的定义，基于 LangChain 源码（langchain_openai/chat_models/base.py）和官方文档（ChatOpenAI)。

类签名

class ChatOpenAI(BaseChatModel):
    def __init__(
        self,
        *,
        model: str = "gpt-3.5-turbo",
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
        timeout: Optional[float] = None,
        max_retries: int = 2,
        api_key: Optional[str] = None,
        organization: Optional[str] = None,
        base_url: Optional[str] = None,
        default_headers: Optional[Dict[str, str]] = None,
        default_query: Optional[Dict[str, Any]] = None,
        http_client: Optional[Any] = None,
        model_kwargs: Optional[Dict[str, Any]] = None,
        n: int = 1,
        stop: Optional[List[str]] = None,
        presence_penalty: Optional[float] = None,
        frequency_penalty: Optional[float] = None,
        top_p: Optional[float] = None,
        logprobs: Optional[bool] = None,
        top_logprobs: Optional[int] = None,
        **kwargs: Any
    ) -> None

关键参数：
- model（str，默认 "gpt-3.5-turbo"）：OpenAI 对话模型名称。
- temperature（float，默认 0.7）：控制生成随机性（0~2）。
- max_tokens（Optional[int]，默认 None）：最大输出 token 数。
- timeout（Optional[float]，默认 None）：请求超时（秒）。
- max_retries（int，默认 2）：请求失败重试次数。
- api_key（Optional[str]，默认 None）：OpenAI API 密钥（可通过环境变量 OPENAI_API_KEY 设置）。
- organization（Optional[str]，默认 None）：OpenAI 组织 ID。
- base_url（Optional[str]，默认 None）：自定义 API 地址（默认 https://api.openai.com/v1）。
- n（int，默认 1）：每提示生成的结果数。
- stop（Optional[List[str]]，默认 None）：停止词列表。
- presence_penalty（Optional[float]，默认 None）：鼓励新词出现。
- frequency_penalty（Optional[float]，默认 None）：降低重复词频率。
- top_p（Optional[float]，默认 None）：核采样概率。
- logprobs（Optional[bool]，默认 None）：是否返回对数概率。
- top_logprobs（Optional[int]，默认 None）：返回前 N 个 token 的对数概率。
- model_kwargs（Optional[Dict[str, Any]]，默认 None）：其他 OpenAI API 参数。

初始化示例

from langchain_openai import ChatOpenAI
import os
os.environ["OPENAI_API_KEY"] = "your-api-key"
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.7,
    max_tokens=512,
    api_key=os.environ["OPENAI_API_KEY"]
)

支持的模型（部分，需确认 OpenAI API 可用性）：

gpt-3.5-turbo
gpt-3.5-turbo-0125
gpt-4
gpt-4-turbo
gpt-4o
gpt-4o-mini
更多见 OpenAI API 文档.

常用方法

ChatOpenAI 继承自 BaseChatModel，支持对话模型的特定接口。以下是常用的方法，基于使用频率和 ChatOpenAI 特性。

1. `invoke`

def invoke(self, input: LanguageModelInput, config: Optional[RunnableConfig] = None, *, stop: Optional[List[str]] = None, **kwargs: Any) -> AIMessage

功能：同步调用，处理消息列表，返回 AIMessage。
输入：LanguageModelInput（通常是 List[BaseMessage]，如 [SystemMessage, HumanMessage]）。
使用场景：单次对话、问答。

示例：

from langchain_core.messages import HumanMessage
response = llm.invoke([HumanMessage(content="什么是人工智能？")])
print(response.content)

2. `generate`

def generate(self, messages: List[List[BaseMessage]], stop: Optional[List[str]] = None, callbacks: Optional[Callbacks] = None, **kwargs: Any) -> LLMResult

功能：同步批量生成，返回 LLMResult（包含 generations 和 llm_output）。
输入：List[List[BaseMessage]]，每个子列表是一组对话消息。
使用场景：批量对话、需要 token 使用量。

示例：

messages = [
    [HumanMessage(content="什么是人工智能？")],
    [HumanMessage(content="AI 有哪些应用？")]
]
result = llm.generate(messages)
for generations in result.generations:
    print(generations[0].text)
print(result.llm_output)  # {'token_usage': ...}

3. `stream`

def stream(self, input: LanguageModelInput, config: Optional[RunnableConfig] = None, *, stop: Optional[List[str]] = None, **kwargs: Any) -> Iterator[AIMessageChunk]

功能：同步流式生成，逐块返回 AIMessageChunk。
使用场景：实时聊天界面。

示例：

for chunk in llm.stream([HumanMessage(content="人工智能有哪些应用？")]):
    print(chunk.content, end="", flush=True)

4. `ainvoke`

async def ainvoke(self, input: LanguageModelInput, config: Optional[RunnableConfig] = None, *, stop: Optional[List[str]] = None, **kwargs: Any) -> AIMessage

功能：异步单次生成。
使用场景：高并发 Web 服务。

示例：

import asyncio
async def run():
    response = await llm.ainvoke([HumanMessage(content="什么是人工智能？")])
    print(response.content)
asyncio.run(run())

5. `agenerate`

async def agenerate(self, messages: List[List[BaseMessage]], stop: Optional[List[str]] = None, callbacks: Optional[Callbacks] = None, **kwargs: Any) -> LLMResult

功能：异步批量生成。
使用场景：异步批量对话。

示例：

async def run():
    result = await llm.agenerate(messages)
    for generations in result.generations:
        print(generations[0].text)
asyncio.run(run())

6. `astream`

async def astream(self, input: LanguageModelInput, config: Optional[RunnableConfig] = None, *, stop: Optional[List[str]] = None, **kwargs: Any) -> AsyncIterator[AIMessageChunk]

功能：异步流式生成。
使用场景：WebSocket 聊天。

示例：

async def run():
    async for chunk in llm.astream([HumanMessage(content="人工智能有哪些应用？")]):
        print(chunk.content, end="", flush=True)
asyncio.run(run())

ChatOpenAI 特有特性：

支持结构化消息输入：

from langchain_core.messages import SystemMessage, HumanMessage
messages = [
    SystemMessage(content="你是 AI 专家"),
    HumanMessage(content="什么是人工智能？")
]

返回 AIMessage 或 AIMessageChunk，包含 content 和元数据：

print(response.additional_kwargs)  # {'finish_reason': 'stop', ...}

支持 token 使用量统计：

print(result.llm_output["token_usage"])

常用属性

以下是 ChatOpenAI 类的常用属性，基于源码和文档。

1. `model`

类型：str
功能：指定 OpenAI 对话模型名称。
示例：
```
print(llm.model)  # gpt-3.5-turbo
```

2. `temperature`

类型：float
功能：控制生成随机性（0~2）。
示例：
```
llm.temperature = 0.5
```

3. `max_tokens`

类型：Optional[int]
功能：最大输出 token 数。
示例：
```
llm.max_tokens = 1024
```

4. `api_key`

类型：Optional[str]
功能：OpenAI API 密钥。
示例：
```
llm.api_key = "new-api-key"
```

5. `cache`

类型：Optional[BaseCache]
功能：缓存生成结果。

示例：

from langchain_core.caches import InMemoryCache
llm.cache = InMemoryCache()

6. `callbacks`

类型：Optional[Callbacks]
功能：默认回调函数。

示例：

from langchain_core.callbacks import StdOutCallbackHandler
llm.callbacks = [StdOutCallbackHandler()]

使用 `langchain_openai.ChatOpenAI` 的对话示例

以下是一个独立示例，展示如何使用 langchain_openai.ChatOpenAI 调用 gpt-3.5-turbo 模型处理人工智能主题的对话，结合 ChatPromptTemplate 和 invoke 方法，启用缓存和回调，支持系统消息和用户问题。

准备环境：

获取 OpenAI API 密钥：OpenAI Platform。
设置环境变量：
```
export OPENAI_API_KEY="your-api-key"
```

安装依赖：

pip install --upgrade langchain langchain-openai

代码：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_core.caches import InMemoryCache
from langchain_core.callbacks import StdOutCallbackHandler
import os

# 设置 API 密钥
os.environ["OPENAI_API_KEY"] = "your-api-key"

# 初始化 ChatOpenAI
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.7,
    max_tokens=512,
    cache=InMemoryCache(),  # 启用缓存
    callbacks=[StdOutCallbackHandler()]  # 默认回调
)

# 定义 ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个人工智能专家，回答以下问题。"),
    ("human", "{question}")
])

# 创建问答链
qa_chain = (
    {"question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 使用 invoke 方法
print("invoke 方法：")
question = "什么是人工智能？"
try:
    response = llm.invoke(
        prompt.format_prompt(question=question).to_messages(),
        config={
            "max_tokens": 512,
            "tags": ["ai_question"]
        }
    )
    print(f"问题: {question}")
    print(f"回答: {response.content}")
except Exception as e:
    print(f"错误: {e}")

# 使用链调用 invoke
print("\n链式 invoke：")
response = qa_chain.invoke(question)
print(f"问题: {question}")
print(f"回答: {response}")

# 查看属性
print("\n模型属性：")
print(f"模型: {llm.model}")
print(f"温度: {llm.temperature}")
print(f"缓存状态: {llm.cache is not None}")

输出示例（实际输出取决于模型和 API 响应）：

invoke 方法：
[StdOutCallbackHandler] 开始生成...
[StdOutCallbackHandler] 生成结束
问题: 什么是人工智能？
回答: 人工智能（AI）是计算机科学的一个分支，旨在创建能够执行需要人类智能的任务的系统，例如学习、推理、问题解决和决策。

链式 invoke：
问题: 什么是人工智能？
回答: 人工智能（AI）是计算机科学的一个分支，旨在创建能够执行需要人类智能的任务的系统，例如学习、推理、问题解决和决策。

模型属性：
模型: gpt-3.5-turbo
温度: 0.7
缓存状态: True

代码说明

LLM 初始化：
- 使用 ChatOpenAI 调用 gpt-3.5-turbo，配置 temperature=0.7、max_tokens=512。
- 启用 InMemoryCache 和 StdOutCallbackHandler。
提示模板：
- ChatPromptTemplate 定义系统消息和用户问题。
问答链：
- LCEL 链组合 prompt、llm 和 StrOutputParser，简化调用。
invoke 使用：
- 直接调用 llm.invoke，使用格式化的消息列表。
- 通过 qa_chain.invoke 调用链，展示集成方式。
属性展示：
- 打印 model、temperature 和 cache 状态。
错误处理：
- 使用 try-except 捕获 API 错误。

运行要求：

有效的 OpenAI API 密钥：
```
export OPENAI_API_KEY="your-api-key"
```

安装依赖：

pip install --upgrade langchain langchain-openai

网络连接：访问 https://api.openai.com。

注意事项

API 密钥：
- 确保 OPENAI_API_KEY 已设置：
```
echo $OPENAI_API_KEY
```
- 或在代码中显式设置：
```
llm = ChatOpenAI(api_key="your-api-key")
```
模型可用性：
- 确认模型支持聊天完成 API（gpt-3.5-turbo 而非 gpt-3.5-turbo-instruct）：
```
print(llm.model)
```
- 检查 OpenAI 文档：Models.

消息格式：

使用 SystemMessage 和 HumanMessage：

from langchain_core.messages import SystemMessage, HumanMessage
messages = [SystemMessage(content="系统提示"), HumanMessage(content="用户问题")]

或使用 ChatPromptTemplate 简化：

prompt = ChatPromptTemplate.from_messages([("system", "提示"), ("human", "{question}")])

参数配置：

stop：控制输出终止：

response = llm.invoke(messages, stop=["\n"])

n：生成多条结果：

result = llm.generate([messages], n=3)
print([g.text for g in result.generations[0]])

回调：使用 StdOutCallbackHandler 记录过程：

response = llm.invoke(messages, config={"callbacks": [StdOutCallbackHandler()]})

性能优化：
- 缓存：启用 InMemoryCache 优化重复查询：
```
llm.cache = InMemoryCache()
```
- 批量处理：使用 generate 处理多组消息：
```
result = llm.generate([messages1, messages2])
```
- 超时和重试：调整 timeout 和 max_retries：
```
llm = ChatOpenAI(timeout=10, max_retries=3)
```
错误调试：
- API 密钥无效：
  - 检查密钥：
```
print(os.environ.get("OPENAI_API_KEY"))
```
  - 重新生成密钥：OpenAI Platform.
- 模型不可用：
  - 确认模型名称：
```
llm = ChatOpenAI(model="gpt-3.5-turbo")
```
- 网络错误：
  - 检查连接：
```
curl https://api.openai.com
```
  - 增加超时：
```
llm = ChatOpenAI(timeout=30)
```
- 输出为空：
  - 检查消息：
```
print(messages)
```
  - 增加 max_tokens：
```
response = llm.invoke(messages, max_tokens=1024)
```

常见问题

Q1：如何选择 ChatOpenAI 和 OpenAI？
A：

ChatOpenAI：适合对话模型（如 gpt-3.5-turbo），支持结构化消息。
OpenAI：适合完成模型（如 gpt-3.5-turbo-instruct），处理原始文本。

from langchain_openai import OpenAI
llm = OpenAI(model="gpt-3.5-turbo-instruct")

Q2：如何启用流式生成？
A：使用 stream 或 astream：

for chunk in llm.stream(messages):
    print(chunk.content, end="")
# 异步
async for chunk in llm.astream(messages):
    print(chunk.content, end="")

Q3：如何查看 token 使用量？
A：使用 generate 获取 llm_output：

result = llm.generate([messages])
print(result.llm_output["token_usage"])

Q4：如何在 FastAPI 中使用？
A：异步调用 ainvoke 或 astream：

from fastapi import FastAPI
from langchain_core.messages import HumanMessage
app = FastAPI()
@app.post("/chat")
async def chat(prompt: str):
    response = await llm.ainvoke([HumanMessage(content=prompt)])
    return {"result": response.content}

总结

langchain_openai.ChatOpenAI 是 LangChain 中调用 OpenAI 对话模型的核心类，核心功能包括：

定义：继承 BaseChatModel，支持 OpenAI 聊天完成 API。
初始化：配置 model、api_key、temperature 等。
常用方法：invoke（单次）、generate（批量）、stream（流式）等。
常用属性：model、temperature、cache 等。

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

【免费下载】 index-tts-vllm：高性能文本转语音解决方案

`index-tts-vllm` 是一款开源的文本转语音（TTS）项目，基于 `index-tts` 进行了优化，通过引入 `vllm` 库重新实现了 GPT 模型的推理过程，大幅提升了推理速度和并发处理能力。项目旨在为开发者提供一个高效、稳定、易于集成的 TTS 解决方案。## 项目技术分析`index-tts-vllm` 采用了以下核心技术：- **vllm 库**：用于加速 GP

AI Agent技术社区

2025年8月SCI-炭黑四角蜂优化算法 Tetragonula carbonaria Optimization Algorithm-附Matlab免费代码

AI Agent技术社区

AgentScope示例应用：多智能体对话

在人工智能快速发展的今天，单一智能体已经难以满足复杂任务的需求。想象一下这样的场景：一个项目团队需要讨论技术方案，不同专业背景的成员各抒己见，最终达成共识。这正是多智能体对话要解决的问题——让多个AI智能体像真实团队成员一样进行有意义的交流。AgentScope作为面向开发者的多智能体编程框架，提供了强大的多智能体对话能力。本文将深入探讨如何使用AgentScope构建高效的多智能体对话系统。..