Python通过LangChain调用LLM模型的完整过程

输入消息 -> ChatOpenAI -> 模型回复

# 它负责把 LangChain 的统一模型接口，转换成“OpenAI 接口格式”的请求。
# 名字里有 openai，但不代表只能调用 OpenAI 官方模型。
# 只要第三方服务也支持 OpenAI 那套请求格式，比如 DeepSeek，也可以用它接入。
from langchain_openai import ChatOpenAI


# ChatOpenAI：代码里真正创建和调用的聊天模型客户端。
# 它不是模型本身，只负责接收配置、消息和参数，然后去请求远程模型服务。
model = ChatOpenAI(
    model="deepseek-v4-pro",
    api_key="your_deepseek_api_key_here",
    base_url="https://api.deepseek.com",
)


# OpenAI 官方接口 / 兼容 OpenAI 格式的第三方接口：真正提供模型能力的远程接口。
# 这里 base_url 指向 DeepSeek，所以请求会发给 DeepSeek。
# 如果 base_url 指向 OpenAI，请求就会发给 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()


# 读取必填环境变量。
# 如果变量不存在，直接抛错，让问题在启动阶段暴露出来，而不是等到请求模型时才失败。
def required_env(name: str) -> str:
    value = os.getenv(name)

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

    return value


# 创建聊天模型客户端。
# 后续所有模型调用都从这里拿实例，避免每个示例重复配置 model、api_key、base_url。
def create_model() -> ChatOpenAI:
    base_url = os.getenv("LLM_BASE_URL")

    # 模型初始化集中在这里，后续 Prompt、Tool、Runnable 都复用同一个入口。
    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 时走默认 OpenAI 地址；配置后可以接 DeepSeek 这类服务。
    if base_url:
        model_kwargs["base_url"] = base_url

    return ChatOpenAI(**model_kwargs)

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

    # invoke 会等模型完整生成后再返回，适合一次性拿完整结果。
    response = model.invoke(
        [
            {
                "role": "user",
                "content": "用三句话解释 LangChain 是什么。",
            }
        ]
    )

    print(response.content)

import sys


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

    # stream 会返回一段段 AIMessageChunk，适合做实时输出。
    stream = model.stream(
        [
            {
                "role": "user",
                "content": "介绍一下重庆火锅文化。",
            }
        ]
    )

    for chunk in stream:
        # 这里只处理文本片段；后面接触多模态或工具调用时，content 可能会更复杂。
        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,
)