一.定义聊天模型

1.ChatOpenAI

class langchain_openai.chat_models.base.ChatOpenAI 是 LangChain 为 OpenAI 的聊天模型（如 gpt-5 , gpt-5-mini ）提供的具体实现类。

其继承了 class langchain_openai.chat_models.base.BaseChatOpenAI ，且BaseChatOpenAI 实现了标准的 Runnable 接口。

ChatOpenAI 常用初始化参数：

model	要使用的 OpenAI 模型的名称
temperature	采样温度，温度值越高，AI 回答越天马行空；温度越低，回答越保守靠谱
max_tokens	要生成的最大令牌数
timeout	请求超时时间
max_retries	最大重试次数
openai_api_key / api_key	OpenAI API 密钥。如果未传入，将从环境变量中读取OPENAI_API_KEY
base_url	API 请求的基本 URL
organization	OpenAI 组织 ID。如果未传入，将从 env var OPENAI_ORG_ID 中读取

举例介绍一下base_url的使用，DeepSeek API 使用与 OpenAI 兼容的 API 格式：

因此，我们可以使用OpenAI的聊天组件来调用DeepSeek：

model = ChatOpenAI(
    model="deepseek-chat",
    api_key="你的API KEY",      # 指定 api key
    base_url="https://api.deepseek.com/v1",     # 指定前置请求路径
)

2.init_chat_model

init_chat_model() 是一个工厂函数，它可以初始化多种支持的聊天模型（如 OpenAI、Anthropic、FireworksAI 等），不仅仅是OpenAI 的聊天模型。

init_chat_model() 常用参数：

model	要使用的模型的名称
model_provider	模型提供方。支持的 model_provider 值和相应的集成包： • openai -> langchain-openai • deepseek -> langchain-deepseek
configurable_fields	设置哪些模型参数是可配置的。若配置为： • None: 没有可配置的字段。 • 'any' ：所有字段都是可配置的，类似 api_key 、 base_url 等可以在运行时更改。 • Union[List[str], Tuple[str, …]]：指定的字段是可配置的。
config_prefix	• 配置为非空字符串，则模型将在运行时通过查找config["configurable"]["{config_prefix}_{param}"] 字段设置配置项 。 • 配置为空字符串，那么模型将可以通过 config["configurable"]["{param}"] 字段设置配置项。
temperature	采样温度，温度值越高，AI 回答越天马行空；温度越低，回答越保守靠谱
max_tokens	要生成的最大令牌数
timeout	请求超时时间
max_retries	最大重试次数
openai_api_key / api_key	OpenAI API 密钥。如果未传入，将从环境变量中读取OPENAI_API_KEY 。
base_url	API 请求的基本 URL

1）基本用法

# 返回 langchain_deepseek.ChatDeepSeek 实例
deepseek_model = init_chat_model("deepseek-chat",
                                 model_provider="deepseek",
                                 temperature=0)

# 由于所有模型集成都实现了ChatModel接⼝，因此可以以相同的⽅式使⽤它们。
print("deepseek-chat: " + deepseek_model.invoke("what's your name").content)

2）创建可配置模型

class langchain_core.runnables.config.RunnableConfig 常用参数：

configurable	通过 .configurable_fields（）在此 Runnable 或子 Runnable 上配置的属性的运行时值
run_id	针对此调用运行的跟踪器的唯一标识符。如果未提供，将生成新的 UUID
run_name	此调用的跟踪器运行的名称。默认为类的名称
metadata	此次调用和任何子调用的元数据。键是字符串，值是 JSON 类型：dict[str， Any]

# 可配置的模型（模型模拟器）
config_model = init_chat_model(temperature=0.3,
                               api_key="sk-c3e81377fd23430bbb1b3e4bd11ac3b6")
messages = [
    SystemMessage(content="请补全一段故事，100个字以内："),
    HumanMessage(content="一只猫正在__？")
]
# .invoke() 的config参数才真正意义上定义了模型
print(f"config_model:{config_model.invoke(input=messages, 
                                          config={"configurable" : {"model" : "deepseek-chat"}}).content}")

3）有默认值的可配置模型

主要靠 configurable_fields 和 config_prefix 这两个参数实现。configurable_fields 指定哪些模型参数可以在运行时被覆盖。config_prefix 为配置项创建命名空间，避免与其他配置冲突

# 可配置的模型（默认参数）
model = init_chat_model(
    model="gpt-4o-mini",
    model_provider="openai",
    temperature=0.3,
    max_tokens=1024,
    configurable_fields=("max_tokens", "model", "model_provider",),
    config_prefix="first",
)
messages = [
    SystemMessage(content="请补全一段故事，100个字以内："),
    HumanMessage(content="一只猫正在__？")
]
result = model.invoke(
    input = messages,
    config={
        "configurable" : {
            "first_max_tokens" : 10,
            "first_model": "deepseek-chat",
            "first_model_provider": "deepseek",
        }
    }
)

3.本地部署的 LLM 定义聊天模型

使用本地大模型，首先要先下载 Ollama ，然后自己部署一个大模型。

然后再下载 Ollama 的包：

pip install -U langchain_ollama

有了 Ollama 的包就可以使用ChatOllama。

class langchain_ollama.chat_models.base.ChatOllama 是 LangChain 为通过Ollama 部署的聊天模型提供的具体实现类。 ChatOllama 同样也实现了标准的 Runnable 接口。

ChatOllama 常用初始化参数：

model	要使用的 Ollama 模型的名称
temperature	采样温度，温度值越高，AI 回答越天马行空；温度越低，回答越保守靠谱
timeout	请求超时时间
base_url	API 请求的基本 URL
num_ctx	设置用于生成下一个令牌的上下文窗口的大小。（默认值：2048）
num_gpu	要使用的 GPU 数量。在 macOS 上，默认为 1 表示启用金属支持，默认为 0 表示禁用

代码示例：

ollama_model = ChatOllama(model="deepseek-r1:1.5b",
                          base_url="http://127.0.0.1:11434")
print(ollama_model.invoke("你是谁？").content)

二.调用工具

LLM 本身是一个封闭的知识系统，其能力来源于训练其的数据，训练完后不更新，数据又滞后性。工具调用根本作用就是让大语言模型（LLM）具备与外部世界交互的能力。

在 LangChain 中，聊天模型提供了额外的功能：工具调用。它能使 LLM 与外部服务、API 和数据库进行交互。工具调用还可用于从非结构化数据中提取结构化信息并执行各种其他任务。

1.创建工具

1）@tool 装饰器创建工具

LangChain 中实现了一个 @tool 装饰器来创建工具， @tool 装饰器是自定义工具的最简单方法。

# 定义工具
@tool
def add(a: int, b: int) -> int:
    """两数相加

    Args:
        a: 第一个整数
        b: 第二个整数
    """
    return a + b

print(add.invoke({"a": 2, "b": 3}))

这里要注意，方法中的注释必须加，因为函数名、类型提示和文档字符串都是传递给工具 Schema 的一部分，不可缺失。

除了这种方式，还有其他方式可以让工具 schema，获取相关工具声明需要的内容。下面再展示其他常用的工具定义模式：

模式一：依赖 Pydantic 类

#  Pydantic 类
class AddInput(BaseModel):
    """Add two integers."""

    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")

@tool(args_schema=AddInput)
def add(a: int, b: int) -> int:
    return a + b

print(add.invoke({"a": 2, "b": 3}))

模式二：依赖 Annotated

@tool
def add(
        a : Annotated[int, ..., "第一个整数"],
        b : Annotated[int, ..., "第二个整数"],
) -> int:
    """两数相加"""
    return a + b

print(add.invoke({"a": 2, "b": 3}))

2）StructuredTool 类提供的函数创建工具

class langchain_core.tools.structured.StructuredTool 类用来初始化工具，其中 from_function 类方法通过给定的函数来创建并返回一个工具。

StructuredTool  关键参数：

func	要设置的工具函数
coroutine	协程函数，要设置的异步工具函数
name	工具名称。默认为函数名称
description	工具描述。默认为函数文档字符串
args_schema	工具输入参数的schema。默认为 None
response_format	工具响应格式。默认为“content”

示例一：常规写法

def add(a: int, b: int) -> int:
    """两数相加"""
    return a + b

add_tool = StructuredTool.from_function(func=add)

print(add_tool.invoke({"a": 2, "b": 5}))

示例二：依赖 Pydantic 类

class AddInput(BaseModel):
    a:int = Field(description="第一个整数")
    b:int = Field(description="第二个整数")


def add(a: int, b: int) -> int:
    return a + b

add_tool = StructuredTool.from_function(
    func=add,
    name="ADD",             # 工具名
    description="两数相加",   # 工具描述
    args_schema=AddInput,   # 工具参数
)

print(add_tool.invoke({"a": 2, "b": 5}))

示例三：加入 response_format 配置

class AddInput(BaseModel):
    a:int = Field(description="第一个整数")
    b:int = Field(description="第二个整数")

def add(a: int, b: int) -> Tuple[str, List[int]]:
    nums = [a, b]
    content = f"{nums}相加的结果是{a + b}"
    return content, nums

add_tool = StructuredTool.from_function(
    func=add,
    name="ADD",             # 工具名
    description="两数相加",   # 工具描述
    args_schema=AddInput,   # 工具参数
    response_format="content_and_artifact"
)

# 模拟大模型调用
print(add_tool.invoke(
    {
        "name": "ADD",
        "args": {"a": 3, "b": 4},
        "type": "tool_call",  # 必填
        "id": "111",  # 必填, 用来将工具调用请求和结果关联起来。
    }
))

2.绑定工具

为了实际将这些工具绑定到聊天模型，可以使用聊天模型的 .bind_tools() 方法。

@tool
def add(
        a : Annotated[int, ..., "第一个整数"],
        b : Annotated[int, ..., "第二个整数"],
) -> int:
    """两数相加"""
    return a + b

@tool
def multiply(
        a : Annotated[int, ..., "第一个整数"],
        b : Annotated[int, ..., "第二个整数"],
) -> int:
    """两数相乘"""
    return a * b

model = ChatOpenAI(
    model="deepseek-chat",
    base_url="https://api.deepseek.com/v1", 
)

# 绑定⼯具，返回⼀个 Runnable 实例
tools = [add, multiply]
model_with_tools = model.bind_tools(tools)

tools	绑定到此聊天模型的工具定义列表。支持的类型为：字典、pydantic.BaseModel 类、Python 函数和 BaseTool（如 @tool 装饰器创建的类）
tool_choice	默认空，要求模型调用哪个工具
parallel_tool_calls	默认为 None ，允许并行工具使用。设置为 False 以禁用并行工具
kwargs(Any)	任何附加参数都直接传递给 bind()

3.工具调用

通过 .bind_tools() 方法我们可知，它返回了一个 Runnable 实例，因此我们可以使用该 Runnable 实例，调用 .invoke() 方法，完成工具调用。

# 调⽤⼯具
result = model_with_tools.invoke("9乘6等于多少？")
print(result)

输出结果：

content = '我来帮您计算9乘6等于多少。'
additional_kwargs = {'tool_calls': [
    {'id': 'call_00_BGnjEj4q06YzoKHL95eQs4Jw', 'function': {'arguments': '{"a": 9, "b": 6}', 'name': 'multiply'},
     'type': 'function', 'index': 0}], 'refusal': None}
response_metadata = {'token_usage': {'completion_tokens': 29, 'prompt_tokens': 238, 'total_tokens': 267,
                                     'completion_tokens_details': None,
                                     'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0},
                                     'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 238},
                     'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_ffc7281d48_prod0820_fp8_kvcache',
                     'id': '7fe62ab3-3734-45c6-b5a7-0bfaf968fa59', 'service_tier': None, 'finish_reason': 'tool_calls',
                     'logprobs': None}
id = 'lc_run--bf17fb35-3eeb-49dc-9cc4-437d776ffdca-0'
tool_calls = [
    {'name': 'multiply', 'args': {'a': 9, 'b': 6}, 'id': 'call_00_BGnjEj4q06YzoKHL95eQs4Jw', 'type': 'tool_call'}]
usage_metadata = {'input_tokens': 238, 'output_tokens': 29, 'total_tokens': 267,
                  'input_token_details': {'cache_read': 0}, 'output_token_details': {}}

content	消息的内容
additional_kwargs	与消息关联的其他有效负载数据。对于来自 AI 的消息，可能包括模型提供程序编码的工具调用
response_metadata	响应元数据。例如：响应标头、logprobs、令牌计数、模型名称
id	本次调用的唯一标识符
tool_calls	模型识别出的工具调用请求列表
usage_metadata	本次调用的使用量统计信息

工具调用的一个关键原则是，模型根据输入的相关性决定何时使用工具。模型并不总是需要调用工具。例如，给定一个不相关的输入，模型不会调用该工具：

result = model_with_tools.invoke("你是谁？")
print(result)

4.强制调用工具

如果想要强制调用工具，那么就要在绑定工具时，设置 tool_choice="any" 

# 绑定⼯具，返回⼀个 Runnable 实例
tools = [add, multiply]
model_with_tools = model.bind_tools(tools,tool_choice="any")

5.将工具输出传递给聊天模型

到此，我们只是调用了工具，但不要忘了，我们调用工具的目的就是为了得到答案，因此我们要将工具输出传递给聊天模型，包括 HumanMessage 、 AIMessage （工具调用）、ToolMessage。聊天模型根据以上的消息输入，将最终结果 AIMessage 返回。

# 绑定工具
tools = [add, multiply]
model_with_tools = model.bind_tools(tools=tools)

# 定义消息列表，添加要传递给聊天模型的消息
message = [
    HumanMessage("2乘3等于多少？6加6等于多少？")
]

ai_msg = model_with_tools.invoke(message)
message.append(ai_msg)

# 构造ToolMessage, 并添加到消息列表中去
for tool_call in ai_msg.tool_calls:
    selected_tool = {"add": add, "multiply": multiply}[tool_call["name"].lower()]
    tool_msg = selected_tool.invoke(tool_call)
    message.append(tool_msg)

print(message)
print(model_with_tools.invoke(message).content)

输出结果：

三.结构化输出

在 LangChain 中，聊天模型提供了额外的功能：结构化输出。一种使聊天模型以结构化格式（例如JSON）进行响应的技术。

聊天模型的 with_structured_output 方法则允许我们预先定义一个期望的数据结构，并要求大模型必须按照这个结构返回信息。

1.with_structured_output

要想使用结构化输出能力，LangChain 提供了一种方法 .with_structured_output() ，该方法需要先定义输出结构，然后执行通过 .with_structured_output() 得到的 Runnable 实例。

此方法将 输出结构 作为参数输入，返回一个类似 model 的 Runnable。不同之处在于执行 Runnable 后的输出结果，输出的不是字符串或消息 ，而是输出与给定输出结构相对应的对象。

该输出结构可以指定为 TypedDict 类、JSON Schema 或 Pydantic 类。如果使用 TypedDict 或JSON Schema，则 Runnable 将返回一个字典，如果使用 Pydantic 类，则将返回一个 Pydantic 对象。

注意，DeepSeek不支持 with_structured_output。

1）返回 Pydantic 对象

# 定义⼤模型
model = ChatOpenAI(model="gpt-4o-mini")
# 定义输出结构：Pydantic 类
class Joke(BaseModel):
    """给用户讲的一个笑话"""

    setup: str = Field(description="这个笑话的开头")
    punchline: str = Field(description="这个笑话的妙语")
    rating: Optional[int] = Field(default=None, description="从1-10分，给这个笑话评分")
# 嵌套输出
class Data(BaseModel):
    """获取关于笑话的数据列表"""

    jokes: List[Joke]
# 使用结构化输出
structured_model = model.with_structured_output(Joke)
result = structured_model.invoke("给我讲⼀个关于唱歌的笑话")
print(result)

2）返回 TypedDict

# TypedDict
class Joke(TypedDict):
    """给用户讲的一个笑话"""

    setup: Annotated[str, ..., "这个笑话的开头"]
    punchline: Annotated[str, ..., "这个笑话的妙语"]
    rating: Annotated[Optional[int], None, "从1-10分，给这个笑话评分"]


# model_with_structured = model.with_structured_output(Joke)
# print(model_with_structured.invoke("讲一个关于跳舞的笑话"))
model_with_structured = model.with_structured_output(Joke, include_raw=True)
print(model_with_structured.invoke("讲一个关于跳舞的笑话"))

include_raw=True 参数的作用是在返回结构化数据的同时，也包含原始的模型响应。

3）返回 JSON

# JSON Schema
json_schema = {
    "title": "joke",
    "description": "给用户讲一个笑话。",
    "type": "object",
    "properties": {
        "setup": {
            "type": "string",
            "description": "这个笑话的开头",
        },
        "punchline": {
            "type": "string",
            "description": "这个笑话的妙语",
        },
        "rating": {
            "type": "integer",
            "description": "从1到10分，给这个笑话评分",
            "default": None,
        },
    },
    "required": ["setup", "punchline"],
}

model_with_structured = model.with_structured_output(json_schema)
print(model_with_structured.invoke("讲一个关于跳舞的笑话"))

四.流式传输

首先说一下什么是流式传输，在我们使用DeepSeek或豆包的时候，输出答案是一点一点的输出，不是等半天一下子输出。这种方法的输出优化了我们用户的体验，我们可以先看已经输出的部分。

但是我们在IDE里面得到的输出都是直接一下子输出，如果想要实现流式输出，可以使用LangChain给我们提供的方法。

1.stream() 同步传输

在 LangChain 聊天模型中，可以使用其 .stream() 方法，来同步生成流式响应的效果。

聊天模型的 .stream() 方法返回一个迭代器，该迭代器在生成输出时同步产生输出消息块 。可以使用 for 循环实时处理每个块。

model = ChatDeepSeek(model="deepseek-chat")

# 返回一个迭代器，产生的消息块
chunks = []
for chunk in model.stream("写一段关于春天的作文，100字"):
    chunks.append(chunk)
    # chunk: AIMessageChunk
    print(chunk.content, end="|", flush=True)

print(model.invoke("写一段关于春天的作文，100字").content)

2.astream() 异步传输

可以使用 .astream() 方法，来异步生成流式响应的效果，这专为非阻塞工作流程而设计。可以在异步代码中使用它来实现相同的实时流式处理行为。

model = ChatDeepSeek(model="deepseek-chat")

# 异步流式输出
async def async_stream():
    print("===异步调用===")
    async for chunk in model.astream("写一段关于春天的作文，100字"):
        print(chunk.content, end="|", flush=True)

asyncio.run(async_stream())

3.StrOutputParser 解析模型的输出

model = ChatDeepSeek(model="deepseek-chat")

# 定义输出解析器
parser = StrOutputParser()

# 定义链
chain = model | parser

for chunk in chain.stream("写⼀段赞美程序员的歌词，需要5句话"):
    print(chunk, end="|", flush=True)

五.LangSmith 跟踪 LLM 应用

长话短说，LangSmith 可以监控我们的应用，下面直接演示。

LangSmith 官网：LangSmith

1.去官网设置一个API KEY

2.设置环境变量

LANGSMITH_TRACING="true"
LANGSMITH_API_KEY="你的 LangSmith API Key"

3.调用一下代码，查看官网