輸入消息 -> ChatOpenAI -> 模型回復(fù)

# 它負(fù)責(zé)把 LangChain 的統(tǒng)一模型接口，轉(zhuǎn)換成“OpenAI 接口格式”的請求。
# 名字里有 openai，但不代表只能調(diào)用 OpenAI 官方模型。
# 只要第三方服務(wù)也支持 OpenAI 那套請求格式，比如 DeepSeek，也可以用它接入。
from langchain_openai import ChatOpenAI


# ChatOpenAI：代碼里真正創(chuàng)建和調(diào)用的聊天模型客戶端。
# 它不是模型本身，只負(fù)責(zé)接收配置、消息和參數(shù)，然后去請求遠(yuǎn)程模型服務(wù)。
model = ChatOpenAI(
    model="deepseek-v4-pro",
    api_key="your_deepseek_api_key_here",
    base_url="https://api.deepseek.com",
)


# OpenAI 官方接口 / 兼容 OpenAI 格式的第三方接口：真正提供模型能力的遠(yuǎn)程接口。
# 這里 base_url 指向 DeepSeek，所以請求會發(fā)給 DeepSeek。
# 如果 base_url 指向 OpenAI，請求就會發(fā)給 OpenAI。
response = model.invoke("你好，請介紹一下 LangChain")

uv add langchain-openai python-dotenv

pip install langchain-openai python-dotenv

LLM_MODEL=deepseek-v4-pro
LLM_API_KEY=your_deepseek_api_key_here
LLM_BASE_URL=https://api.deepseek.com

import os

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI


load_dotenv()


# 讀取必填環(huán)境變量。
# 如果變量不存在，直接拋錯，讓問題在啟動階段暴露出來，而不是等到請求模型時才失敗。
def required_env(name: str) -> str:
    value = os.getenv(name)

    if not value:
        raise RuntimeError(f"Missing environment variable: {name}")

    return value


# 創(chuàng)建聊天模型客戶端。
# 后續(xù)所有模型調(diào)用都從這里拿實(shí)例，避免每個示例重復(fù)配置 model、api_key、base_url。
def create_model() -> ChatOpenAI:
    base_url = os.getenv("LLM_BASE_URL")

    # 模型初始化集中在這里，后續(xù) Prompt、Tool、Runnable 都復(fù)用同一個入口。
    model_kwargs = {
        "model": required_env("LLM_MODEL"),
        "api_key": required_env("LLM_API_KEY"),
        "temperature": 0.2,
        "timeout": 30,
        "max_retries": 2,
    }

    # 不配置 LLM_BASE_URL 時走默認(rèn) OpenAI 地址；配置后可以接 DeepSeek 這類服務(wù)。
    if base_url:
        model_kwargs["base_url"] = base_url

    return ChatOpenAI(**model_kwargs)

def invoke_demo() -> None:
    model = create_model()

    # invoke 會等模型完整生成后再返回，適合一次性拿完整結(jié)果。
    response = model.invoke(
        [
            {
                "role": "user",
                "content": "用三句話解釋 LangChain 是什么。",
            }
        ]
    )

    print(response.content)

import sys


def stream_demo() -> None:
    model = create_model()

    # stream 會返回一段段 AIMessageChunk，適合做實(shí)時輸出。
    stream = model.stream(
        [
            {
                "role": "user",
                "content": "介紹一下重慶火鍋文化。",
            }
        ]
    )

    for chunk in stream:
        # 這里只處理文本片段；后面接觸多模態(tài)或工具調(diào)用時，content 可能會更復(fù)雜。
        if isinstance(chunk.content, str):
            sys.stdout.write(chunk.content)
            sys.stdout.flush()

    sys.stdout.write("\n")

ChatOpenAI(
    api_key="...",
    base_url="...",
    model="deepseek-v4-pro",
)

ChatOpenAI(
    temperature=0.2,
    max_tokens=1000,
    top_p=1,
)

ChatOpenAI(
    timeout=30,
    max_retries=2,
)