LangChain 学习(2) 消息类型与 @tool

LangChain 消息类型

  在 LangChain 中，所有的对话都通过消息(Message)对象传递。理解各类消息类型的用途是编写 Agent 的基础。

  LangChain 定义了四种核心消息类型，分别对应对话中的不同角色:

类型	角色	说明	典型内容
HumanMessage	用户	用户发送的消息	“今天天气怎么样”
AIMessage	AI 助手	模型的回复，可能包含 tool_calls	“今天杭州晴天，25°C”
SystemMessage	系统	系统指令，定义 AI 的角色和行为规则	你是一个专业的天气助手
ToolMessage	工具	工具执行后的返回结果	“晴，25°C，湿度 60%”

HumanMessage ———— 用户消息

  HumanMessage 代表用户发送给 AI 的消息，这是最常见的消息类型，也是对话的起点。示例代码如下:

from langchain.chat_models import init_chat_model
import os
from model_init import model_init
from langchain.messages import HumanMessage

message = HumanMessage(content="你好")

messages = [
    HumanMessage(content="你好"),
    HumanMessage(content="今天天气怎么样")
]

model = model_init()

response = model.invoke(messages)
print(response.content)

AIMessage ———— AI回复

  AIMessage 代表模型的回复。与普通文本不同，AIMessage 可能包含 tool_calls（工具调用请求）,示例代码如下：

from langchain.messages import AIMessage

ai_msg = AIMessage(content = "菜鸟教程是一个编程学习平台")

ai_msg_with_tools = AIMessage(
    content="",
    tool_calls=[
        {
            "name": "get_weather",
            "args": {"city": "杭州"},
            "id": "call_abc123",
            "type": "tool_call",
        }
    ]
)

print("=== 普通 AI 消息 ===")
print(f"content: {ai_msg.content}")
print(f"tool_calls: {ai_msg.tool_calls}")   # []
print("\n=== 含工具调用的 AI 消息 ===")
print(f"content: {ai_msg_with_tools.content}")
print(f"tool_calls: {ai_msg_with_tools.tool_calls}")

  一般来说，模型返回的 AIMessage 中还包含了其他信息，如输入 Tokens、输出 Tokens、总计 Tokens 等信息。调用示例代码如下：

from langchain.chat_models import init_chat_model
import os
from model_init import model_init
from langchain.messages import HumanMessage

message = HumanMessage(content="你好")

model = model_init()

response = model.invoke([message])
print(f"内容: {response.content}")
print(f"消息ID: {response.id}")
print(f"模型名: {response.response_metadata.get('model_name')}")
print(f"完成原因: {response.response_metadata.get('finish_reason')}")

# usage_metadata 包含 Token 用量信息
if response.usage_metadata:
    print(f"输入 tokens: {response.usage_metadata.get('input_tokens')}")
    print(f"输出 tokens: {response.usage_metadata.get('output_tokens')}")
    print(f"总计 tokens: {response.usage_metadata.get('total_tokens')}")

SystemMessage ———— 系统指令

  SystemMessage 用于设定 AI 的行为、角色和约束，它会被放在消息列表的最前面，指导模型如何回复，示例如下：

from langchain.chat_models import init_chat_model
import os
from model_init import model_init
from langchain.messages import HumanMessage, SystemMessage

model = model_init()

messages_with_system = [
    SystemMessage(content="你是一个小红书风格的博主，回复要活泼、使用 emoji、带话题标签"),
    HumanMessage(content="介绍菜鸟教程")
]
response = model.invoke(messages_with_system)
print(f"\n有系统指令: {response.content}")

ToolMessage ———— 工具返回结果

  ToolMessage 包含工具执行后的返回结果。它必须与对应的 tool_call 关联。示例如下：

from langchain.messages import HumanMessage, AIMessage, ToolMessage
from langchain.chat_models import init_chat_model

# 模拟一轮完整的工具调用对话
messages = [
    HumanMessage(content="杭州天气怎么样？"),

    # 模型请求调用工具
    AIMessage(
        content="",
        tool_calls=[
            {"name": "get_weather", "args": {"city": "杭州"},
             "id": "call_abc", "type": "tool_call"}
        ]
    ),

    # 工具返回结果（必须包含 tool_call_id 与上面的 id 对应）
    ToolMessage(
        content="晴，25°C，湿度 60%",
        tool_call_id="call_abc",   # 与 tool_call 的 id 对应
        name="get_weather",        # 工具名称
    ),
]

model = init_chat_model("deepseek:deepseek-v4-flash")
response = model.invoke(messages)
print(f"模型基于工具结果的回复: {response.content}")

  ToolMessage 的 tool_call_id 必须与 AIMessage 中 tool_call 的 id 精确匹配。如果不匹配，模型可能会忽略这个工具结果，或者产生混乱的行为。

AIMessageChunk ———— 流式输出的消息片段

  当使用 stream() 流式输出时，每个到达的片段是 AIMessageChunk，而非完整的 AIMessage：

from langchain.chat_models import init_chat_model

model = init_chat_model("deepseek:deepseek-v4-flash")

print("流式输出过程：")
# stream() 返回的是 AIMessageChunk 迭代器
for chunk in model.stream("用一句话介绍菜鸟教程 RUNOOB"):
    # 每个 chunk 是一小段文本
    print(chunk.content, end="", flush=True)
print()  # 换行

多模态消息

  如果模型支持多模态输入，可以让其分析图片内容，示例代码如下：

from langchain.chat_models import init_chat_model
import os
from model_init import model_init
from pathlib import Path
import base64
from langchain.messages import HumanMessage, SystemMessage

model = model_init()

def encode_img(image_path: str) -> str:
    """将图片转为 Base64 编码"""
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

image_data = encode_img("1.png")
messages = [
    HumanMessage(content=[
        {"type": "text", "text": "请描述这张截图"},
        {
            "type": "image_url",
            "image_url": {
                "url": f"data:image/png;base64,{image_data}",
                "detail": "auto"  # 可选：low, high, auto
            }
        }
    ])
]

response = model.invoke(messages)
print(f"图片分析结果: {response.content}")

裁剪消息历史

  当对话越来越长时，消息列表可能超出模型的上下文窗口。trim_messages() 函数可以帮助你智能地裁剪消息历史。

trimmed = trim_messages(
    messages,
    max_tokens=1000,           # 最多保留 1000 tokens
    strategy="last",           # 保留最后的系统消息 + 最近的对话
    token_counter=model,       # 使用模型的 token 计数方式
    include_system=True,       # 始终保留 SystemMessage
    start_on="human",          # 裁剪后以 human 消息开头
)

删除特定消息

  在某些场景中，我们可以使用 RemoveMessage() 从消息历史中删除特定消息，使用示例如下：

from langchain.messages import HumanMessage, AIMessage, RemoveMessage

# 假设有一段对话
messages = [
    HumanMessage(content="你好", id="msg_1"),
    AIMessage(content="你好！有什么可以帮你的？", id="msg_2"),
    HumanMessage(content="帮我查天气", id="msg_3"),
]

# 使用 RemoveMessage 删除特定消息（通过 ID）
# RemoveMessage 配合 add_messages reducer 使用
# 在更新 Agent 状态时，RemoveMessage 会从列表中移除对应 ID 的消息
removal = RemoveMessage(id="msg_3")

print(f"要删除的消息 ID: {removal.id}")
print(f"类型: {removal.type}")  # remove

LangChain @tool 装饰器

  工具(Tool)是 Agent 与外部世界交互的桥梁，通过 @tool 装饰器，可以将任何 Python 函数快速转换为 Agent 可以调用的工具。

@tool 基本语法

  @tool 是 LangChain 提供的装饰器，用法很简单，在函数上加上 @tool 装饰器，函数就变成了一个工具，示例如下：

from langchain.tools import tool
# 最简单的工具：一个普通函数 + @tool 装饰器
@tool
def hello_tool(name: str) -> str:
    """向指定的人打招呼。

    Args:
        name: 要打招呼的人的名字
    """
    return f"你好，{name}！欢迎来到菜鸟教程 RUNOOB。"

# 工具也是普通的 Python 函数，可以直接调用
result = hello_tool.invoke({"name": "小明"})
print(result)

# 工具包含自动生成的描述信息
print(f"\n工具名称: {hello_tool.name}")
print(f"工具描述: {hello_tool.description}")

  函数的文档字符串（docstring）会自动成为工具的描述。Agent 依赖这个描述来判断”这个工具能做什么”和”什么情况下应该调用它”。文档字符串写得越清晰，Agent 使用工具就越准确。在描述中说明参数含义、函数功能和使用场景。

  在创建 Agent 时将定义好的工具传给 create_agent() 的 tools 参数，Agent 就能使用它了：

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[search_courses, get_course_detail],
    system_prompt="你是菜鸟教程 RUNOOB 的学习顾问，帮助用户找到合适的课程。",
)

  一个 Agent 可以注册多个工具，模型会自动判断何时使用哪个工具。我们还可以为工具参数设置默认值，让 Agent 在调用工具时不用每次都指定所有参数。但注意：如果某个参数没有默认值且 Agent 没有提供，调用会失败。关键参数不要设默认值。

一些工具的高级特性

return_direct ———— 直接返回最终结果

  默认情况下，工具执行后结果会返回给模型，模型再基于工具结果生成最终回复。但有时工具结果本身就是你想要的最终答案。设置 return_direct=True 后，工具执行完就立即结束 Agent 循环，工具返回内容直接作为最终输出。

# return_direct 工具：结果直接作为最终输出
@tool(return_direct=True)
def search_direct(keyword: str) -> str:
    """搜索菜鸟教程 RUNOOB 的课程（直接返回模式）。

    当用户只需要搜索结果，不需要额外分析时使用此工具。
    """
    return f"搜索结果：Python3 基础教程、Python 数据分析、Python 爬虫入门"

  当你设置 return_direct=True 时，Agent 会跳过后续的模型思考步骤，直接返回工具结果。这在节省 Token 和降低时延方面非常有价值，但也意味着模型不会对工具结果做任何二次加工。如果一个 Agent 同时挂载了多个工具，其中既有 return_direct=True 的工具，也有普通工具，那么只要模型在这一轮调用中触发了任意一个 return_direct 工具，Agent 循环就会立即结束——即使同一轮还并行调用了其他普通工具，它们的结果也不会再被模型加工总结。设计包含多个工具的 Agent 时要留意这一点，避免”该总结的内容被跳过”。

InjectedToolCalled ———— 获取工具调用 ID

  有时工具需要知道”是谁调用了它”——InjectedToolCallId 可以在工具函数中注入当前的 tool_call_id，示例代码如下：

from typing import Annotated

from langchain.messages import ToolCall
from langchain.tools import tool, InjectedToolCallId


@tool
def log_user_action(
    action: str,
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> str:
    """记录用户操作到日志系统。

    Args:
        action: 用户操作描述
        tool_call_id: 系统自动注入的工具调用 ID
    """

    # 实际项目中这里会写入数据库或发送到日志服务

    return f"操作已记录 (调用ID: {tool_call_id}): {action}"


# 手动模拟模型产生的 ToolCall
tool_call: ToolCall = {
    "name": "log_user_action",
    "args": {
        "action": "用户查询了 Python 课程"
    },
    "id": "call_log_001",
    "type": "tool_call",
}


# InjectedToolCallId 会从 tool_call.id 中自动注入
result = log_user_action.invoke(tool_call)

print(result)

  带有 InjectedToolArg 标记的参数不需要由 Agent（模型）提供， 这些参数会由 LangChain 运行时在执行工具时自动注入。 由于这些参数不会出现在工具的 schema 中，模型无法看到它们， 因此不应该把它们作为用户需要填写的工具参数进行描述。
  在 Agent 工作流中， InjectedToolCallId 更常用于需要关联当前调用上下文的场景， 例如配合 Command 对象在工具内部更新 Agent 状态， 向 messages 列表追加关联当前调用的 ToolMessage， 或者实现工具调用追踪、审计、日志关联等功能。这也是它被设计为”注入参数”而不是普通参数的原因： 它代表的是 LangChain Runtime 当前正在执行的这一次 ToolCall 上下文， 而不是用户输入的一部分。

ToolException ———— 工具异常处理

  工具执行过程中可能会出错。使用 ToolException 抛出明确的工具异常，让 Agent 知道出了问题。示例如下：

from langchain.tools import tool, ToolException

@tool
def get_user_info(user_id: int) -> str:
    """根据用户 ID 查询用户信息。

    Args:
        user_id: 用户 ID，必须是正整数
    """
    # 数据校验
    if user_id <= 0:
        # 抛出 ToolException，而不是普通 Exception
        # ToolException 会被 Agent 捕获并告知模型
        raise ToolException(f"用户 ID 必须为正整数，收到了: {user_id}")

    # 模拟数据库查询
    users = {
        1: "张三（VIP 会员，注册于 2024-01-15）",
        2: "李四（普通用户，注册于 2024-03-20）",
    }

    if user_id not in users:
        raise ToolException(f"未找到 ID 为 {user_id} 的用户")

    return users[user_id]


# 正常调用
print(get_user_info.invoke({"user_id": 1}))

# 异常调用 1：无效 ID
try:
    get_user_info.invoke({"user_id": -1})
except ToolException as e:
    print(f"工具异常: {e}")

# 异常调用 2：用户不存在
try:
    get_user_info.invoke({"user_id": 999})
except ToolException as e:
    print(f"工具异常: {e}")

  这里之所以能在 .invoke() 外层用 try/except 捕获到 ToolException，是因为工具默认的 handle_tool_error 为 False——异常不会被工具自己吞掉，而是照常向上抛出。

handle_tool_error ———— 让工具自己处理错误

  当希望某个工具出错时不中断程序，而是把错误信息转成一段文本、当作正常返回值交给模型自己去理解和修正时，可以在定义工具时设置 handle_tool_error（注意是单数，没有 s）。这是 BaseTool 上的一个属性，最简单的设置方式是直接写在 @tool 装饰器里：

@tool(handle_tool_error=True)
def get_weather(city: str) -> str:
    """查询指定城市的天气。

    Args:
        city: 城市名称，必须是中文全称，如 "杭州"、"北京"
    """
    weather_data = {
        "杭州": "晴，25°C",
        "北京": "多云，18°C",
        "上海": "小雨，22°C",
    }
    if city not in weather_data:
        # 城市不在数据中时抛出 ToolException
        raise ToolException(
            f"未收录城市 '{city}'。"
            f"可使用城市：{', '.join(weather_data.keys())}。"
            f"请使用中文城市全称。"
        )
    return f"{city}天气：{weather_data[city]}"

  这样工具给出的报错就会直接返回给大模型，让模型生成处理。

LangChain 工具访问 ———— InjectedState 与 InjectedStore

  有时工具需要访问更多的上下文信息，比如当前对话的状态、用户的持久化数据等。LangChain 通过依赖注入机制，让工具函数能够自动获取这些信息。

InjectedState ———— 在工具中访问 Agent 状态

  默认情况下，工具只能通过参数接收模型传来的数据，但有时工具需要知道当前对话的上下文，比如之前对话的历史、用户已确认的信息等等，而 InjectedState 让工具可以直接读取 Agent 的完整状态，示例代码如下：

from typing import Annotated, Any, TypedDict
from model_init import model_init
from langchain.tools import tool, InjectedState
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langgraph.graph import MessagesState  # 内置消息状态


# 1. 定义状态，继承 MessagesState 并添加自定义字段
class AgentState(MessagesState):
    user_preferences: str  # 自定义字段


@tool
def remember_preference(
    preference: str,
    state: Annotated[dict, InjectedState],
) -> str:
    """记住用户的偏好设置。"""
    messages = state.get("messages", [])
    message_count = len(messages)
    previous_prefs = state.get("user_preferences", "无")

    return (
        f"已记住偏好: {preference}。"
        f"(当前对话共 {message_count} 条消息，"
        f"之前偏好: {previous_prefs})"
    )


# 2. 初始化模型
model = model_init()

# 3. 创建 Agent，使用自定义的状态类型
agent = create_agent(
    model=model,
    tools=[remember_preference],
    state_schema=AgentState,  
)

# 4. 初始化状态
initial_state = {
    "messages": [HumanMessage(content="我喜欢暗色主题")],
    "user_preferences": "浅色主题",  # 旧偏好
}

# 5. 调用 Agent
result = agent.invoke(initial_state)
print(result["messages"][-1].content)

InjectedStore ———— 在工具中访问永久化存储

  Agent 状态（state）是对话级别的，对话结束就没了。而 Store 是跨会话的持久化存储，可以用来保存用户偏好、学习进度等长期信息。示例代码如下：

from typing import Annotated
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, InjectedStore, InjectedState
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from model_init import model_init

# 创建 Store 并预置数据
store = InMemoryStore()
store.put(("runoob", "courses"), "catalog", {
    "data": {
        "Python3 基础教程": {"price": "免费", "duration": "20小时"},
        "Python 数据分析": {"price": "会员", "duration": "30小时"},
        "HTML 基础教程": {"price": "免费", "duration": "15小时"},
    }
})

@tool
def query_course_price(
    course_name: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """查询菜鸟教程 RUNOOB 中指定课程的价格信息。

    Args:
        course_name: 课程名称
    """
    item = store.get(("runoob", "courses"), "catalog")
    catalog = item.value["data"] if item else {}

    if course_name in catalog:
        info = catalog[course_name]
        return f"《{course_name}》- 价格：{info['price']}，学习时长：{info['duration']}"
    return f"未找到课程《{course_name}》"

model = model_init()
agent = create_agent(
    model=model,
    tools=[query_course_price],
    store=store,
    system_prompt="你是一位课程顾问,请不要使用 markdown 语法输出"
)

result = agent.invoke({
    "messages": [HumanMessage(content="Python3 基础教程和 Python 数据分析分别多少钱？")]
})
print(result["messages"][-1].content)

  在运行过程中，LangChain 会自动注入 store 给工具。写入如下工具，可以实现录入课程信息：

@tool
def add_course_price(    
    course_name: str,
    price: str,
    duration: str,
    store: Annotated[BaseStore, InjectedStore],
) -> str:
    """给 store 中加入课程、定价、时长信息"""
    item = store.get(("runoob", "courses"), "catalog")  # ← 统一 namespace
    catalog = item.value["data"] if item else {}
    catalog[course_name] = {"price": price, "duration": duration}
    store.put(("runoob", "courses"), "catalog", {"data": catalog})  # ← 写回完整数据
    return f"已成功添加课程《{course_name}》，价格：{price}，时长：{duration}"