# 导入 LangChain 专门用来创建「结构化工具」的类
from langchain_core.tools import StructuredTool

# 定义一个普通的 Python 函数：两数相乘
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""  # 文档字符串，大模型会看懂这个功能
    return a * b

# 重点：把普通函数 → 变成 LangChain 能用的「工具」
calculator_tool = StructuredTool.from_function(func=multiply)

# 调用这个工具（传入字典格式的参数）
print(calculator_tool.invoke({"a": 2, b": 3}))  # 输出 6

1. 使用 args_schema 参数，依赖 Pydantic 类定义并提供工具输入参数的 schema 属性。

2. 使用 description 参数，替代文档字符串中对于工具描述的 schema 属性。

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field


class CaCalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")

def multiply(a: int, b: int) -> int:
    return a * b

calculator_tool = StructuredTool.from_function(
    func = multiply,
    name = "Calculator",
    description="两数相乘",
    args_schema=CaCalculatorInput,
)
print(calculator_tool.invoke({"a": 3, "b": 2}))
print(calculator_tool.name)
print(calculator_tool.description)

---

四、绑定工具

定义工具后，要把工具列表绑到聊天模型上。可以使用聊天模型的 bind_tools() 方法，绑定之后，模型在一次推理中如果认为需要，就会在返回的 AIMessage 里附带 tool_calls。

例如：

from langchain_deepseek import ChatDeepSeek

# 定义大模型
model = ChatDeepSeek(model="deepseek-v4-flash")

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

bind_tools() 会做两件事：

1. 将工具的名称、描述和参数 schema 注入到模型的系统指令中（或者通过特定 API 参数）。

2. 返回一个新的可运行对象，之后的调用都将携带这个工具列表信息。

现在 chat_with_tools 就是一个知道“我身边有两个工具可用”的聊天模型了。

---

五、调用工具

 #调用工具
response = model_with_tools.invoke("3 + 5 等于多少？")
print(response)

模型通过response的内容来决定调用哪个工具，比如 "3 + 5 等于多少？"的话模型就会调用 add 工具。

模型并没有回答等于 8 ，而是表达了一个调用请求。它说：“我需要调用 add，参数是  3 和 5”。

---

六、工具属性

在 LangChain 中，每个工具都是 BaseTool 的实例，具有以下重要属性：

• name：唯一标识。模型会在 tool_calls 中引用这个名字。

• description：功能说明。务必写得清晰、具体。坏的描述会让模型在错误的时机调用。

• args_schema：一个 Pydantic 模型，定义了参数的名称、类型、是否必填、默认值等。模型会严格按照它来生成 JSON 参数。

• return_direct：如果设为 True，工具执行后直接返回结果给用户，不再经过模型润色（在 Agent 中常用）。

例如，我们可以用更精细的方式定义工具：

from pydantic import BaseModel, Field
from langchain_core.tools import StructuredTool

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称，例如'北京'")

def get_weather_func(city: str) -> str:
    weather_db = {"北京": "霾，5°C"}
    return weather_db.get(city, "未知")

get_weather_tool = StructuredTool.from_function(
    func=get_weather_func,
    name="get_weather",
    description="查询指定城市的实时天气",
    args_schema=WeatherInput
)

当模型看到 args_schema 定义的字段和描述后，就能准确地填入 {"city": "北京"}。

---

七、将工具输出传递给聊天模型

到这里可以发现，我们仅仅只是成功调用了工具，但是聊天模型并没有给我们返回我们真正需要的答案。此时就需要：

为什么要发 ToolMessage 呢？

之前我们讲过，聊天模型通常不是接受单个字符串作为输入，而是接受聊天消息（XxxMessage）列表，因此在这里我们需要将工具的返回，构造成 ToolMessage，再传输给聊天模型！！！方便的是，如果我们使用 @tool 装饰器创建的工具，使用 tool.invoke(tool_calls)，将自动返回一个 ToolMessage。完整示例如下：

from langchain_deepseek import ChatDeepSeek
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from typing_extensions import Annotated

# 定义大模型
model = ChatDeepSeek(model="deepseek-chat")

# 定义工具
@tool
def add(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Add two integers."""
    return a + b

@tool
def multiply(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Multiply two integers."""
    return a * b

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

# 添加AIMessage到消息中
messages = [
    HumanMessage("9乘6等于多少？5加3等于多少? ")
]
ai_msg = model_with_tools.invoke(messages)
messages.append(ai_msg)

for tool_call in ai_msg.tool_calls:
    # 根据工具名选择对应工具函数（不区分大小写）
    selected_tool = {"add": add, "multiply": multiply}[tool_call["name"].lower()]
    # 执行工具调用，返回 ToolMessage
    tool_msg = selected_tool.invoke(tool_call)
    # 将 ToolMessage 加入消息
    messages.append(tool_msg)

print(messages)

result = model.invoke(messages)
print(result)

---

八、结构化输出

很多时候，我们不希望模型的最终回答是一段纯文本，而是希望得到 JSON、对象等结构化数据，方便程序处理。这就需要输出解析器。

8.1 StrOutputParser —— 提取纯文本字符串

StrOutputParser 是 LangChain 框架内置的基础输出解析器，实现了 BaseOutputParser 标准接口，专用于将大模型返回的 AIMessage 消息对象，解析为纯字符串（str）类型。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_deepseek import ChatDeepSeek

# 定义模型
model = ChatDeepSeek(model="deepseek-chat")

# 提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个翻译助手"),
    ("human", "将'{text}'翻译成英文")
])
chain = prompt | model | StrOutputParser()
text = chain.invoke({"text": "你好世界"})
print(text)  # "Hello World"

• prompt：生成标准化提示词
• chat：大模型执行推理，返回 AIMessage 对象
• StrOutputParser：将 AIMessage 解析为纯字符串，完成最终输出

输入参数

        ↓

prompt 生成对话消息

        ↓

模型返回 AIMessage（复杂对象）

        ↓

StrOutputParser 提取 .content  →  转为 str

        ↓

得到纯文本字符串

8.2 PydanticOutputParser —— 最强大的结构化输出

利用 Pydantic 定义想要的输出格式，解析器会自动生成格式指令并校验输出。

from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from pydantic import BaseModel, Field

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

class Person(BaseModel):
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    email: str = Field(description="邮箱")

parser = PydanticOutputParser(pydantic_object=Person)

prompt = ChatPromptTemplate.from_messages([
    ("system", "提取用户信息，按指定格式输出。{format_instructions}"),
    ("human", "我叫李明，28岁，邮箱是liming@example.com")
])
prompt = prompt.partial(format_instructions=parser.get_format_instructions())

chain = prompt | model| parser
person_obj = chain.invoke({})
print(person_obj.name)  # 李明
print(person_obj.age)   # 28

parser.get_format_instructions() 会生成类似这样的指令：

The output should be formatted as a JSON instance that conforms to the JSON schema below...
{"properties": {"name": {"description": "姓名", "type": "string"}, ...}}

模型读取这段指令后，会直接输出符合该 JSON Schema 的字符串，解析器再将其转化为 Pydantic 对象。如果格式不符，解析器会抛出异常，可以在链中配置重试。

注意： 在支持原生 function calling 的模型上，也可以结合工具调用获取结构化输出，但 PydanticOutputParser 依然是适用范围最广的通用方案。

---

九、流式传输

聊天类应用通常需要逐字显示回复，提升用户体验。LangChain 提供了 stream 方法。

messages = [
    ("system", "你是一个诗人，作答要简洁。"),
    ("human", "写一首关于海的三行诗")
]

for chunk in chat.stream(messages):
    # chunk 是一个 AIMessageChunk 对象，content 包含增量文本
    print(chunk.content, end="", flush=True)

输出结果：

波涛轻吟着古老的歌，
月光洒下银色的故事，
远方是梦的彼岸。

stream 返回的是一个生成器，每次 yield 一个 token 块。也可以结合异步环境使用 astream：

async for chunk in chat.astream(messages):
    print(chunk.content, end="", flush=True)

如果工具调用过程中也需要流式输出，通常我们在工具调用循环的最终回答生成阶段使用 stream，而不是在第一次（可能产生 tool_calls 的）调用中流式，因为 tool_calls 需要完整返回才能解析。

---

十、总结

把以上所有流程串联，写一个终端运行的、带工具调用和流式回答的聊天机器人。

import os
from langchain_deepseek import ChatDeepSeek
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool

# ---------- 1. API定义模型 ----------
chat = ChatDeepSeek(model="deepseek-chat", temperature=0)


# ---------- 2. 定义工具 ----------
@tool
def get_weather(city: str) -> str:
    """查询指定城市天气"""
    db = {"北京": "霾 5°C", "上海": "小雨 12°C"}
    return db.get(city, "无数据")


@tool
def calculator(exp: str) -> str:
    """计算数学表达式"""
    return str(eval(exp))


# ---------- 3. 绑定工具 ----------
chat_with_tools = chat.bind_tools([get_weather, calculator])

# ---------- 4. 交互循环 ----------
messages = []
print("🤖 聊天机器人启动（输入 quit 退出）")
while True:
    user_input = input("你：")
    if user_input.lower() == "quit":
        break
    messages.append(HumanMessage(content=user_input))

    # 首次调用（可能产生工具调用）
    response = chat_with_tools.invoke(messages)

    # 处理工具调用
    while response.tool_calls:
        messages.append(response)  # 将AIMessage加入历史
        for tc in response.tool_calls:
            if tc["name"] == "get_weather":
                result = get_weather.invoke(tc["args"])
            elif tc["name"] == "calculator":
                result = calculator.invoke(tc["args"])
            else:
                result = "未知工具"
            messages.append(ToolMessage(content=result, tool_call_id=tc["id"]))
        # 再次调用模型
        response = chat_with_tools.invoke(messages)

    # 模型给出最终文本回答，并进行流式输出
    print("AI：", end="")
    # messages 中已经包含了本次所有上下文，但我们也可以用 stream 重新生成最终回答
    # 简便起见直接打印 content，流式演示我们另外模拟：
    full = chat.invoke(messages)
    print(full.content)
    messages.append(full)  # 将最终回答也加入历史

代码核心模块拆解：

1. 模型初始化 ：导入依赖，连接 DeepSeek 大模型，设置低随机性，保证回答稳定。

2. 自定义工具 ：

   • 天气工具：内置固定城市天气数据，用于查询天气；

   • 计算工具：借助eval执行数学表达式运算；

   • 通过@tool装饰器标记，让模型识别为可调用函数。

3. 工具绑定 ：将两个自定义工具绑定给大模型，赋予模型主动调用工具的能力。

4. 循环对话逻辑 ：

   • 开启无限聊天循环，收集用户输入，输入quit即可退出；

   • 用列表存储完整对话记录，实现上下文连贯；

5. 自动工具调用核心 ：

   • 模型首次接收问题后，判断是否需要调用工具；

   • 若需要，代码自动执行对应工具、获取结果并回填对话记录；

   • 再次请求大模型，让模型结合工具返回的数据，生成最终回答。

6. 结果输出 ：打印 AI 最终回复，并将 AI 回答存入上下文，维持多轮对话连贯