如何评估图

langgraph

langgraph 是一个用于使用 LLM 构建有状态、多参与者应用的库，可用于创建代理和多代理工作流。评估 langgraph 图可能具有挑战性，因为单次调用可能涉及多次 LLM 调用，而具体发起哪些 LLM 调用可能取决于前序调用的输出。在本指南中，我们将重点介绍如何将图和图节点传递给 evaluate() / aevaluate()。如需了解构建代理时的评估技术和最佳实践，请参阅 langgraph 文档。

端到端评估

最常见的评估类型是端到端评估，即针对每个示例输入，评估图的最终输出。

定义图

首先，让我们构建一个简单的 ReACT 代理：

from typing import Annotated, Literal, TypedDict
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.prebuilt import ToolNode
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

class State(TypedDict):
    # 消息的类型是 "list"。注解中的 'add_messages' 函数
    # 定义了应如何更新此状态键
    # （在本例中，它会将消息追加到列表中，而不是覆盖消息）
    messages: Annotated[list, add_messages]

# 定义代理要使用的工具
@tool
def search(query: str) -> str:
    """调用网络搜索。"""
    # 这是一个占位实现，但不要告诉 LLM……
    if "sf" in query.lower() or "san francisco" in query.lower():
        return "现在是 60 度，有雾。"
    return "现在是 90 度，天气晴朗。"

tools = [search]
tool_node = ToolNode(tools)
model = init_chat_model("claude-sonnet-4-6").bind_tools(tools)

# 定义决定是否继续的函数
def should_continue(state: State) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]

    # 如果 LLM 发起了工具调用，就将流程路由到 "tools" 节点
    if last_message.tool_calls:
        return "tools"

    # 否则停止（回复用户）
    return END

# 定义调用模型的函数
def call_model(state: State):
    messages = state['messages']
    response = model.invoke(messages)

    # 我们返回一个列表，因为它会被添加到现有列表中
    return {"messages": [response]}

# 定义一个新图
workflow = StateGraph(State)

# 定义将在其中循环的两个节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# 将 'agent' 设置为入口点
# 这意味着该节点会被第一个调用
workflow.add_edge(START, "agent")

# 现在添加一条条件边
workflow.add_conditional_edges(
    # 首先定义起始节点。这里使用 'agent'。
    # 这意味着在调用 'agent' 节点后，将沿着这些边继续执行。
    "agent",
    # 接下来传入一个函数，由它决定下一步调用哪个节点。
    should_continue,
)

# 现在添加一条从 'tools' 到 'agent' 的普通边。
# 这意味着调用 'tools' 后，接下来会调用 'agent' 节点。
workflow.add_edge("tools", 'agent')

# 最后，编译它！
# 这会将其编译为 LangChain Runnable，
# 也就是说，你可以像使用其他 runnable 一样使用它。
# 注意，在编译图时，我们（可选地）传入了 memory
app = workflow.compile()

创建数据集

让我们创建一个包含问题和预期回答的简单数据集：

from langsmith import Client

questions = [
    "旧金山的天气怎么样",
    "旧金山的天气怎么样",
    "丹吉尔的天气怎么样"
]

answers = [
    "现在是 60 度，有雾。",
    "现在是 60 度，有雾。",
    "现在是 90 度，天气晴朗。",
]

ls_client = Client()
dataset = ls_client.create_dataset("天气代理")
ls_client.create_examples(
    inputs=[{"question": q} for q in questions],
    outputs=[{"answer": a} for a in answers],
    dataset_id=dataset.id,
)

创建评估器

再创建一个简单的评估器：需要 langsmith>=0.2.0

judge_llm = init_chat_model("gpt-5.5")

async def correct(outputs: dict, reference_outputs: dict) -> bool:
    instructions = (
        "给定实际回答和预期回答，请判断实际回答是否包含"
        "预期回答中的全部信息。如果实际回答确实包含预期回答中的"
        "全部信息，请回复“CORRECT”；否则回复“INCORRECT”。"
        "不要在回复中包含其他内容。"
    )
    # 我们的图输出的是一个 State 字典，这意味着在本例中
    # 会有一个 'messages' 键，而最后一条消息应该
    # 就是实际回答。
    actual_answer = outputs["messages"][-1].content
    expected_answer = reference_outputs["answer"]
    user_msg = (
        f"实际回答：{actual_answer}"
        f"\n\n预期回答：{expected_answer}"
    )
    response = await judge_llm.ainvoke(
        [
            {"role": "system", "content": instructions},
            {"role": "user", "content": user_msg}
        ]
    )
    return response.content.upper() == "CORRECT"

运行评估

现在我们可以运行评估并查看结果。我们只需要包装图函数，使其能够接收与示例中存储格式相同的输入：

如果所有图节点都定义为同步函数，则可以使用 evaluate 或 aevaluate。如果有任何节点定义为异步函数，则需要使用 aevaluate

需要 langsmith>=0.2.0

import asyncio
from langsmith import aevaluate

def example_to_state(inputs: dict) -> dict:
  return {"messages": [{"role": "user", "content": inputs['question']}]}

# 这里使用 LCEL 声明式语法。
# 记住，langgraph 图也是 langchain runnable。
target = example_to_state | app

async def main():
    experiment_results = await aevaluate(
        target,
        data="天气代理",
        evaluators=[correct],
        max_concurrency=4,  # 可选
        experiment_prefix="claude-sonnet-4-6-baseline",  # 可选
        metadata={  # 可选，用于填充 UI 中的模型/提示词/工具列
            "models": "google_genai:gemini-3.6-flash",
            "tools": [{"name": "search", "description": "调用网络搜索。"}],
        },
    )
    print(experiment_results)

asyncio.run(main())

评估中间步骤

很多时候，除了评估代理的最终输出之外，评估它执行过的中间步骤也很有价值。langgraph 的一个优点是，图的输出是一个状态对象，其中通常已经包含所执行中间步骤的信息。通常，只需查看状态中的消息，就可以评估我们感兴趣的内容。例如，我们可以查看消息，断言模型是否在第一步调用了 search 工具。需要 langsmith>=0.2.0

def right_tool(outputs: dict) -> bool:
    tool_calls = outputs["messages"][1].tool_calls
    return bool(tool_calls and tool_calls[0]["name"] == "search")

async def main():
    experiment_results = await aevaluate(
        target,
        data="天气代理",
        evaluators=[correct, right_tool],
        max_concurrency=4,  # 可选
        experiment_prefix="claude-sonnet-4-6-baseline",  # 可选
        metadata={  # 可选，用于填充 UI 中的模型/提示词/工具列
            "models": "google_genai:gemini-3.6-flash",
            "tools": [{"name": "search", "description": "调用网络搜索。"}],
        },
    )
    print(experiment_results)

如果需要访问中间步骤中未包含在状态里的信息，可以查看 Run 对象。该对象包含所有节点输入和输出的完整追踪信息：

如需进一步了解可以传递给自定义评估器的参数，请参阅这篇操作指南。

from langsmith.schemas import Run, Example

def right_tool_from_run(run: Run, example: Example) -> dict:
    # 获取文档和回答
    first_model_run = next(run for run in root_run.child_runs if run.name == "agent")
    tool_calls = first_model_run.outputs["messages"][-1].tool_calls
    right_tool = bool(tool_calls and tool_calls[0]["name"] == "search")
    return {"key": "right_tool", "value": right_tool}

async def main():
    experiment_results = await aevaluate(
        target,
        data="天气代理",
        evaluators=[correct, right_tool_from_run],
        max_concurrency=4,  # 可选
        experiment_prefix="claude-sonnet-4-6-baseline",  # 可选
        metadata={  # 可选，用于填充 UI 中的模型/提示词/工具列
            "models": "google_genai:gemini-3.6-flash",
            "tools": [{"name": "search", "description": "调用网络搜索。"}],
        },
    )
    print(experiment_results)

运行并评估单个节点

有时，你可能希望直接评估单个节点，以节省时间和成本。langgraph 可以让这一过程变得非常简单。在这种情况下，我们甚至可以继续使用之前使用的评估器。

node_target = example_to_state | app.nodes["agent"]

async def main():
    node_experiment_results = await aevaluate(
        node_target,
        data="天气代理",
        evaluators=[right_tool_from_run],
        max_concurrency=4,  # 可选
        experiment_prefix="claude-sonnet-4-6-model-node",  # 可选
        metadata={  # 可选，用于填充 UI 中的模型/提示词/工具列
            "models": "google_genai:gemini-3.6-flash",
            "tools": [{"name": "search", "description": "调用网络搜索。"}],
        },
    )
    print(node_experiment_results)

参考代码

Click to see a consolidated code snippet

import asyncio
from typing import Annotated, Literal, TypedDict
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.prebuilt import ToolNode
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langsmith import Client, aevaluate

# 定义一个图
class State(TypedDict):
    # 消息的类型是 "list"。注解中的 'add_messages' 函数
    # 定义了应如何更新此状态键
    # （在本例中，它会将消息追加到列表中，而不是覆盖消息）
    messages: Annotated[list, add_messages]

# 定义代理要使用的工具
@tool
def search(query: str) -> str:
    """调用网络搜索。"""
    # 这是一个占位实现，但不要告诉 LLM……
    if "sf" in query.lower() or "san francisco" in query.lower():
        return "现在是 60 度，有雾。"
    return "现在是 90 度，天气晴朗。"

tools = [search]
tool_node = ToolNode(tools)
model = init_chat_model("claude-sonnet-4-6").bind_tools(tools)

# 定义决定是否继续的函数
def should_continue(state: State) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]

    # 如果 LLM 发起了工具调用，就将流程路由到 "tools" 节点
    if last_message.tool_calls:
        return "tools"

    # 否则停止（回复用户）
    return END

# 定义调用模型的函数
def call_model(state: State):
    messages = state['messages']
    response = model.invoke(messages)
    # 我们返回一个列表，因为它会被添加到现有列表中
    return {"messages": [response]}

# 定义一个新图
workflow = StateGraph(State)

# 定义将在其中循环的两个节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# 将 'agent' 设置为入口点
# 这意味着该节点会被第一个调用
workflow.add_edge(START, "agent")

# 现在添加一条条件边
workflow.add_conditional_edges(
    # 首先定义起始节点。这里使用 'agent'。
    # 这意味着在调用 'agent' 节点后，将沿着这些边继续执行。
    "agent",
    # 接下来传入一个函数，由它决定下一步调用哪个节点。
    should_continue,
)

# 现在添加一条从 'tools' 到 'agent' 的普通边。
# 这意味着调用 'tools' 后，接下来会调用 'agent' 节点。
workflow.add_edge("tools", 'agent')

# 最后，编译它！
# 这会将其编译为 LangChain Runnable，
# 也就是说，你可以像使用其他 runnable 一样使用它。
# 注意，在编译图时，我们（可选地）传入了 memory
app = workflow.compile()

questions = [
    "旧金山的天气怎么样",
    "旧金山的天气怎么样",
    "丹吉尔的天气怎么样"
]

answers = [
    "现在是 60 度，有雾。",
    "现在是 60 度，有雾。",
    "现在是 90 度，天气晴朗。",
]

# 创建数据集
ls_client = Client()
dataset = ls_client.create_dataset("天气代理")
ls_client.create_examples(
    inputs=[{"question": q} for q in questions],
    outputs=[{"answer": a} for a in answers],
    dataset_id=dataset.id,
)

# 定义评估器

judge_llm = init_chat_model("gpt-5.5")

async def correct(outputs: dict, reference_outputs: dict) -> bool:
    instructions = (
        "给定实际回答和预期回答，请判断实际回答是否包含"
        "预期回答中的全部信息。如果实际回答确实包含预期回答中的"
        "全部信息，请回复“CORRECT”；否则回复“INCORRECT”。"
        "不要在回复中包含其他内容。"
    )
    # 我们的图输出的是一个 State 字典，这意味着在本例中
    # 会有一个 'messages' 键，而最后一条消息应该
    # 就是实际回答。
    actual_answer = outputs["messages"][-1].content
    expected_answer = reference_outputs["answer"]
    user_msg = (
        f"实际回答：{actual_answer}"
        f"\n\n预期回答：{expected_answer}"
    )
    response = await judge_llm.ainvoke(
        [
            {"role": "system", "content": instructions},
            {"role": "user", "content": user_msg}
        ]
    )
    return response.content.upper() == "CORRECT"

def right_tool(outputs: dict) -> bool:
    tool_calls = outputs["messages"][1].tool_calls
    return bool(tool_calls and tool_calls[0]["name"] == "search")

def example_to_state(inputs: dict) -> dict:
  return {"messages": [{"role": "user", "content": inputs['question']}]}

# 这里使用 LCEL 声明式语法。
# 记住，langgraph 图也是 langchain runnable。
target = example_to_state | app

# 运行评估
async def main():
    experiment_results = await aevaluate(
        target,
        data="天气代理",
        evaluators=[correct, right_tool],
        max_concurrency=4,  # 可选
        experiment_prefix="claude-sonnet-4-6-baseline",  # 可选
        metadata={  # 可选，用于填充 UI 中的模型/提示词/工具列
            "models": "google_genai:gemini-3.6-flash",
            "tools": [{"name": "search", "description": "调用网络搜索。"}],
        },
    )
    print(experiment_results)

asyncio.run(main())

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Datasets

Run an evaluation

Evaluation techniques

Analyze experiment results

Tutorials

Common data types

端到端评估

定义图

创建数据集

创建评估器

运行评估

评估中间步骤

运行并评估单个节点

相关内容

参考代码

​端到端评估

​定义图

​创建数据集

​创建评估器

​运行评估

​评估中间步骤

​运行并评估单个节点

​相关内容

​参考代码

端到端评估

定义图

创建数据集

创建评估器

运行评估

评估中间步骤

运行并评估单个节点

相关内容

参考代码