搭建测试环境

LangSmith

langsmith是一个官方的测试环境，可以帮助我们测试智能体，在开发智能体时很有用:

# Python >= 3.11 is required.

 uv add langgraph-cli

然后你需要注册一个langsmith官方账号来获得密钥，这里我们可以只使用它的本地功能，而是从github上下载一个前端来部署(也就是不把项目交付给langGraph官方服务器跑测试，因为要挂梯子麻烦不安全)

先完成以下步骤:

创建 LangGraph 配置文件

在项目根目录下创建 langgraph.json 文件：

agent路径可自由更改，主要看你的智能体在哪里

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent.py:agent"
  },
  "env": ".env"
}

启动本地的 Agent 服务器：

langgraph dev

Ctrl + C关闭

LangGraph安装

uv add langgraph

IPython

状态State

状态展示了图的配置并随跟踪时间变化，作为所有节点和边的输入和输出

class State(TypedDict):
    """
    定义一个类型化的字典类，用于表示状态信息。
    这个类继承自TypedDict，提供了静态类型检查的支持，
    确保字典中必须包含指定类型的键值对。

    属性:
        state (str): 表示状态的字符串值
    """
    state: str

节点State

节点是一个python函数，函数的第一个参数是State(状态)

# 节点
def node_1(state):
    print("节点1执行中...")
    return {"state": state["state"] + "->节点1"}

def node_2(state):
    print("节点2执行中...")
    return {"state": state["state"] + "->节点2"}

def node_3(state):
    print("节点3执行中...")
    return {"state": state["state"] + "->节点3"}

def node_4(state):
    print("节点4执行中...")
    return {"state": state["state"] + "->节点4"}

边Edges

边代表两个节点的连接，定义它们之间的关系，有普通边和条件边

• 普通边:总是按照既定路径执行
• 条件边:节点之间的可选路径，基于函数逻辑返回下一个节点

# 边
def get_random_edge(state) -> Literal["node2", "node3"]:
    """
    获取一个随机的边。
    返回:
        str: 随机的边名称
    """
    current_state = state['state']
    if random.random() < 0.5:
        return "node2"
    else:
        return "node3"

构建图

import os
import tempfile

from langgraph.constants import START, END
from langgraph.graph import StateGraph
from IPython.display import Image, display
from src.node_and_edge import node_1, node_2, node_3, node_4, get_random_edge, State

# 生成图构建器
builder = StateGraph(State)
builder.add_node("node1", node_1)
builder.add_node("node2", node_2)
builder.add_node("node3", node_3)
builder.add_node("node4", node_4)

# 定义边
builder.add_edge(START, "node1")
builder.add_conditional_edges("node1", get_random_edge)
builder.add_edge("node2", "node4")
builder.add_edge("node3", "node4")
builder.add_edge("node4", END)

# 编译图
graph = builder.compile()

# 可视化
# 获取 PNG 数据
png_data = graph.get_graph().draw_mermaid_png()

# 保存到临时文件并打开
with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as tmp:
    tmp.write(png_data)
    tmp_path = tmp.name

# 用默认图片查看器打开（Windows）
os.startfile(tmp_path)
result = graph.invoke({"state": "你好，你能干什么"})

完整示例:

消息

消息是 LangChain 中模型上下文的基本单位。它们代表模型的输入和输出，携带内容和元数据，用于在与 LLM 交互时表示对话状态。

消息是包含以下内容的对象：

• 角色 - 标识消息类型（例如 system、user）
• 内容 - 表示消息的实际内容（例如文本、图像、音频、文档等）
• 元数据 - 可选字段，例如响应信息、消息 ID 和令牌使用情况

LangChain 提供了一种标准消息类型，可在所有模型提供商之间工作，确保无论调用哪个模型都能保持一致的行为。

消息的分类

• 系统消息 - 告诉模型如何行为并为交互提供上下文

• 人类消息 - 表示用户输入和与模型的交互，它们可以包含文本、图像、音频、文件以及任何其他多模态内容

• AI 消息 - 模型生成的响应，包括文本内容、工具调用和元数据

  • usage_metadata:该字段中保存令牌计数和其他使用元数据：

  • from langchain.chat_models import init_chat_model
    
    model = init_chat_model("openai:gpt-5-nano")
    
    response = model.invoke("Hello!")
    response.usage_metadata
    ---------------------------------------------------------------------
    {'input_tokens': 8,
     'output_tokens': 304,
     'total_tokens': 312,
     'input_token_details': {'audio': 0, 'cache_read': 0},
     'output_token_details': {'audio': 0, 'reasoning': 256}}
    

  • 流式传输和块:

    chunks = []
    full_message = None
    for chunk in model.stream("Hi"):
        chunks.append(chunk)
        print(chunk.text)
        full_message = chunk if full_message is None else full_message + chunk
    

• 工具消息 - 表示工具调用的输出

from langchain.messages import SystemMessage, HumanMessage, AIMessage
#一个消息对象列表
messages = [
    # 系统消息
    SystemMessage("You are a poetry expert"),
    # 人类消息
    HumanMessage("Write a haiku about spring"),
    # ai消息
    AIMessage("Cherry blossoms bloom...")
]
response = model.invoke(messages)

工具

智能体如何被大模型调用?

工具 是 代理（agents） 调用来执行操作的组件。它们通过允许模型通过定义明确的输入和输出与世界交互来扩展模型的功能。工具封装了一个可调用的函数及其输入架构（schema）。这些可以传递给兼容的 聊天模型（chat models），让模型决定是否以及使用什么参数来调用工具。在这些场景中，工具调用 使模型能够生成符合指定输入架构的请求。

服务器端工具使用 (Server-side tool use)

某些聊天模型（例如 OpenAI、Anthropic 和 Gemini）具有 内置工具，这些工具在服务器端执行，例如网络搜索和代码解释器。请参阅 提供商概览（provider overview） 了解如何使用您的特定聊天模型访问这些工具。

也就是说，langchain的工具就是一个被封装的函数，模型可以自主调用这些函数

创建工具

创建工具最简单的方法是使用 @tool 装饰器。默认情况下，函数的 文档字符串（docstring）会成为工具的描述，帮助模型理解何时使用它：

from langchain.tools import tool
@tool(parse_docstring=True)
def function_name(param1, param2):
    """Function summary line. （简单描述该函数或方法的功能）

    Args:
        param1 (int): The first parameter.
        param2 (str): The second parameter.

    Returns:
        bool: The return value. True for success, False otherwise.

    Raises: （这里提供了可能会碰到的报错，非必要可不写）
        ValueError: If `param1` is equal to `param2`.
    """

更多请查阅官方文档:
工具 | LangChain 中文文档

智能体如何知道使用哪个工具?

如果你使用了coogle风格注释，那么工具描述信息将成为智能体调用的依据

传递消息

使用llm时，我们需要在节点间传递消息，因此消息成为状态的一部分，我们来看官方库的实现:
from langgraph.graph import MessagesState:

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

以上就是一个经典的消息实体，使用Annotated和add_messages来自动追加消息

消息实体被用于图信息的传输，作为各个节点的参数被接收，比如以下就是一个小实例:

from IPython.display import Image, display
from langchain_core.messages import HumanMessage, ToolMessage
from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_core.tools import tool

from src.agent.llm import model
from src.utils.show_graph import show_graph


# 定义一个工具函数：乘法运算
@tool
def multiply_values(a, b):
    """
    将两个值相乘并返回结果。

    Args:
        a (float): 第一个值。
        b (float): 第二个值。

    Result:
        float: a和b的乘积。
    """
    return a * b

# 将工具绑定到LLM
llm_tools = model.bind_tools([multiply_values])

# 定义节点：带工具的LLM
def llm_with_tools(state: MessagesState):
    return {"messages": [llm_tools.invoke(state["messages"])]}


# 定义工具执行节点
def run_tool(state: MessagesState):
    # 获取最后一条消息（应为包含 tool_calls 的 AIMessage）
    last_msg = state["messages"][-1]

    # 遍历所有 tool_calls（这里只处理第一个，或全部）
    tool_messages = []
    for tool_call in last_msg.tool_calls:
        # 根据工具名称选择工具（这里只有 multiply_values）
        if tool_call["name"] == "multiply_values":
            # 执行工具
            result = multiply_values.invoke(tool_call["args"])
        else:
            result = f"Tool '{tool_call['name']}' not found."

        # 构造 ToolMessage（每个 tool_call 对应一个 ToolMessage）
        tool_messages.append(
            ToolMessage(
                content=str(result),  # 工具执行结果
                tool_call_id=tool_call["id"]  # 对应的调用 ID
            )
        )
    return {"messages": tool_messages}

# 构建图
builder = StateGraph(MessagesState)
builder.add_node("llm_with_tools", llm_with_tools)  # 添加节点
builder.add_node("run_tool", run_tool) # 添加调用工具节点
builder.add_edge(START, "llm_with_tools")  # 从START到llm_with_tools的边
builder.add_edge("llm_with_tools", "run_tool")
builder.add_edge("run_tool", END)
graph = builder.compile()  # 编译图

# 可视化图
show_graph(graph)
# 调用图并传入乘法问题
messages = graph.invoke({"messages": HumanMessage(content="200乘以30等于几")})
# 打印所有消息
for m in messages['messages']:
    m.pretty_print()

构建的图如下:

其中，MessagesState作为消息实体在边上进行传输。

================================ Human Message =================================

200乘以30等于几
================================== Ai Message ==================================
Tool Calls:
  multiply_values (call_af5391669c3d4a569c2987)
 Call ID: call_af5391669c3d4a569c2987
  Args:
    a: 200
    b: 30
================================= Tool Message =================================

6000

下一节:路由器及实战
end