构建你的第一个LangGraph智能体——初学者实用手册

本指南将逐步演示如何从原始材料构建出可运行的系统，内容来自《构建你的第一个LangGraph智能体：实用初学者指南》。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库的代码，无需猜测其用途。 在概览阶段，应在修改代码之前明确输入参数、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需推测隐藏状态。 除了功能结果外，还需记录执行时间以及token或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。

为什么智能体如此重要

在处理“为什么智能体很重要”这一阶段时，首先写下相关契约：所需的输入参数、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新执行后续节点时，恢复流程不应再次计费相同的大型语言模型调用。

User question → LLM → Answer

User question → Retrieve context → LLM → Answer

User question → Agent decides what to do → Calls tools → Updates state → Produces answer

什么是LangGraph？

在处理“什么是LangGraph”这一阶段时，首先需写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及死信处理都属于产品功能的一部分，而非后续需要补充的内容。 在成本较高的操作之后设置检查点。当操作员重新尝试某个节点时，恢复流程不应再次调用相同的LLM接口。

项目设置

在项目设置阶段，首先需明确相关规范：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改始终符合预期。 建议采用小型、可测试的单元而非庞大的脚本。当某个步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新执行后续节点时，恢复流程不应再次调用相同的大型语言模型。 在项目设置阶段，首先需明确相关规范：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改始终符合预期。 除了功能结果外，还需记录执行时间以及token或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外收费。

pip install langgraph langchain langchain-openai langchain-community python-dotenv pydantic

pip install jupyter notebook ipython

OPENAI_API_KEY=your_api_key_here

import os
import json
from typing import Optional, List

from dotenv import load_dotenv
from pydantic import BaseModel

from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langchain.tools import tool

load_dotenv()

llm = init_chat_model("openai:gpt-4o-mini")

第一步：定义智能体状态

将“定义阶段”视为可测量的结构来处理最为有效。在扩大范围之前，先记录一份最佳案例、一个失败案例以及回滚说明。 配置应置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放，以便操作人员无需查看整个结构即可进行审计。 保持结构状态的扁平化与类型化。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会导致在中断后无法继续处理。

class PlayerState(BaseModel):
    question: str
    selected_tools: Optional[List[str]] = None

    name: Optional[str] = None
    club: Optional[str] = None
    country: Optional[str] = None
    number: Optional[int] = None
    rating: Optional[int] = None

    goals: Optional[List[int]] = None
    minutes_played: Optional[List[int]] = None

    summary: Optional[str] = None

第二步：创建工具

将“第二步：创建工具”阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份理想的操作流程、一个失败案例以及回滚说明。同时文档化正常流程与恢复流程。重试机制、人工审核环节以及错误处理都属于产品功能的一部分，而非后续需要补充的内容。应提供具有明确结构定义和清晰副作用标注的工具，这样主机在自动批准之前就能知道哪些调用会改变状态。

工具1：获取玩家信息

将工具1的“获取玩家信息”阶段视为可度量的对象来使用效果最佳。在扩大范围之前，先记录一个理想案例、一个失败案例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某一步骤失败时，故障应指向单一责任点，而非复杂的流程链。 工具的设计应具备明确的架构规范和清晰的副作用标识。主机需要在自动批准之前知道哪些调用会修改状态。 将工具1的“获取玩家信息”阶段视为可度量的对象来使用效果最佳。在扩大范围之前，先记录一个理想案例、一个失败案例以及回滚说明。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。

@tool
def fetch_player_information_tool(name: str):
    """
    Returns the football club and country of a player.
    """
    data = {
        "Haaland": {
            "club": "Manchester City",
            "country": "Norway"
        },
        "Kane": {
            "club": "Bayern Munich",
            "country": "England"
        },
        "Lautaro": {
            "club": "Inter Milan",
            "country": "Argentina"
        },
        "Ronaldo": {
            "club": "Al-Nassr",
            "country": "Portugal"
        }
    }

    return data.get(name, {
        "club": "unknown",
        "country": "unknown"
    })

def fetch_player_information(state: PlayerState):
    return fetch_player_information_tool.invoke({
        "name": state.name
    })

工具2：获取球衣号码

在工具2的获取球衣信息阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审核。 在网关处进行身份验证，在数据层再次授权。仅凭承载令牌并不足以界定租户边界。

@tool
def fetch_player_jersey_number_tool(name: str):
    """
    Returns the jersey number of a football player.
    """
    data = {
        "Haaland": 9,
        "Kane": 9,
        "Lautaro": 10,
        "Ronaldo": 7
    }

    return {
        "number": data.get(name, 0)
    }

def fetch_player_jersey_number(state: PlayerState):
    return fetch_player_jersey_number_tool.invoke({
        "name": state.name
    })

工具3：获取球员评分

在工具3的“获取玩家数据”阶段，修改代码之前需先明确输入参数、该步骤的负责人以及结束条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品功能的一部分，而非后续需要补充的内容。 在网关处进行身份验证，在数据层再次授权。仅凭承载令牌并不能作为租户边界。

@tool
def fetch_player_rating_tool(name: str):
    """
    Returns the player's football video game rating.
    """
    data = {
        "Haaland": 92,
        "Kane": 89,
        "Lautaro": 88,
        "Ronaldo": 90
    }

    return {
        "rating": data.get(name, 0)
    }

def fetch_player_rating(state: PlayerState):
    return fetch_player_rating_tool.invoke({
        "name": state.name
    })

步骤3：添加数据检索节点

在第三步“添加数据”阶段，修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 相较于冗长的脚本，更应采用小型且可测试的单元。当某一步骤失败时，故障原因应能明确指向某个具体责任方，而非整个复杂的流程。 需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分是虚假信息还是索引缺失导致的错误。 在第三步“添加数据”阶段，修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。

检索目标

在处理“检索目标”阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，这样操作人员无需查看整个系统结构即可进行审计。 在调整提示词之前，先使用固定的问题集来衡量检索的召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。

def retrieve_goals(state: PlayerState):
    data = {
        "Haaland": [25, 40, 28, 33, 36],
        "Kane": [33, 37, 41, 38, 29],
        "Lautaro": [19, 25, 27, 24, 25],
        "Ronaldo": [27, 32, 28, 30, 36]
    }

    return {
        "goals": data.get(state.name, [0])
    }

检索比赛时长

在处理“检索播放时长”这一阶段时，首先需记录下相关契约：所需输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合要求。 同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。 在调整提示词之前，先使用固定的问题集来衡量检索准确率。仅仅更换提示词往往无法解决检索效果不佳的问题。

def retrieve_minutes_played(state: PlayerState):
    data = {
        "Haaland": [2108, 3102, 3156, 2617, 2758],
        "Kane": [2924, 2850, 3133, 2784, 2680],
        "Lautaro": [2445, 2498, 2519, 2773],
        "Ronaldo": [3001, 2560, 2804, 2487, 2771]
    }

    return {
        "minutes_played": data.get(state.name, [0])
    }

第4步：提取玩家姓名

在执行第4步“提取阶段”时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 建议采用小型、可测试的单元而非庞大的脚本。当某一步骤失败时，故障应指向单一责任点，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的LLM服务。 在执行第4步“提取阶段”时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及Token或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外收费。

def extract_name(state: PlayerState):
    prompt = f"""
You are a football name extraction assistant.

Extract only the surname of the football player mentioned in the question.

Question:
{state.question}

Return only one word.
"""

    response = llm.invoke([
        HumanMessage(content=prompt)
    ]).content.strip()

    return {
        "name": response
    }

Will Erling Haaland win the World Cup with Norway?

{"name": "Haaland"}

第5步：构建规划节点

将第5步的构建过程视为可度量的结构会更为有效。在扩大范围之前，先记录一份最佳实践案例、一个失败案例以及回滚说明。 配置信息应置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放，以便操作人员无需查看整个结构就能进行审计。 保持图结构的扁平化与类型化。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致无法继续处理。

def planner(state: PlayerState):
    prompt = f"""
You are a football assistant planner.

Available tools:
[
  "fetch_player_jersey_number",
  "fetch_player_information",
  "fetch_player_rating"
]

User question:
{state.question}

Choose which tools are needed to answer the question.

Return only a valid JSON array of tool names.
Example:
["fetch_player_information", "fetch_player_rating"]
"""

    response = llm.invoke([
        HumanMessage(content=prompt)
    ]).content.strip()

    try:
        selected_tools = json.loads(response)
    except Exception:
        selected_tools = []

    return {
        "selected_tools": selected_tools
    }

第6步：撰写最终总结

将第6步“编写代码”视为可度量的工作面时效果最佳。在扩大范围之前，先记录一个成功的用例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。保持图结构简洁且类型明确，嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致流程无法继续。

def write_summary(state: PlayerState):
    player_data = {
        "name": state.name,
        "club": state.club,
        "country": state.country,
        "number": state.number,
        "rating": state.rating,
        "goals": state.goals,
        "minutes_played": state.minutes_played
    }

    prompt = f"""
You are a football journalist.

Using the data below, write a concise markdown summary about the player.

Player data:
{json.dumps(player_data, indent=4)}

The summary should include:
- Full player name if it can be inferred
- Country with a flag emoji if available
- Club if available
- Jersey number if available
- Rating if available
- Total goals in the last three seasons
- Average minutes per goal over the available data
- A direct answer to the user's question

User question:
{state.question}
"""

    response = llm.invoke([
        HumanMessage(content=prompt)
    ]).content

    return {
        "summary": response
    }

第7步：构建LangGraph工作流

将“第7步：构建阶段”视为可度量的对象来处理效果最佳。在扩大范围之前，需记录一份理想的执行日志、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某一步骤出现故障时，故障应指向单一责任模块，而非复杂的流程链。 保持图结构的状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在流程中断后导致无法继续执行。 将“第7步：构建阶段”视为可度量的对象来处理效果最佳。在扩大范围之前，需记录一份理想的执行日志、一个故障案例以及回滚说明。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境转向共享环境时出现意外费用。

graph_builder = StateGraph(PlayerState)

graph_builder.add_node("extract_name", extract_name)
graph_builder.add_node("planner", planner)

graph_builder.add_node("fetch_player_jersey_number", fetch_player_jersey_number)
graph_builder.add_node("fetch_player_information", fetch_player_information)
graph_builder.add_node("fetch_player_rating", fetch_player_rating)

graph_builder.add_node("retrieve_goals", retrieve_goals)
graph_builder.add_node("retrieve_minutes_played", retrieve_minutes_played)
graph_builder.add_node("write_summary", write_summary)

第8步：添加边

在“第8步：添加边”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的边，必须经过人工审批。编译时的连接方式并不等同于业务流程的完整性。

graph_builder.add_edge(START, "extract_name")
graph_builder.add_edge("extract_name", "planner")

def route_tools(state: PlayerState):
    if state.selected_tools:
        return state.selected_tools

    return ["retrieve_goals"]

graph_builder.add_conditional_edges(
    "planner",
    route_tools,
    {
        "fetch_player_jersey_number": "fetch_player_jersey_number",
        "fetch_player_information": "fetch_player_information",
        "fetch_player_rating": "fetch_player_rating",
        "retrieve_goals": "retrieve_goals"
    }
)

graph_builder.add_edge("fetch_player_jersey_number", "retrieve_goals")
graph_builder.add_edge("fetch_player_information", "retrieve_goals")
graph_builder.add_edge("fetch_player_rating", "retrieve_goals")

graph_builder.add_edge("retrieve_goals", "retrieve_minutes_played")
graph_builder.add_edge("retrieve_minutes_played", "write_summary")
graph_builder.add_edge("write_summary", END)

graph = graph_builder.compile()

第9步：运行代理

在第九步“运行阶段”中，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的功能。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。

result = graph.invoke({
    "question": "Will Haaland be able to win the FIFA World Cup for Norway based on his recent performance?"
})

print(result["summary"])

## Erling Haaland Performance Summary

- **Player:** Erling Haaland
- **Country:** Norway 🇳🇴
- **Club:** Manchester City
- **FIFA-style rating:** 92
- **Recent goals:** 28, 33, and 36 in the last three seasons
- **Total goals in last three seasons:** 97
- **Average minutes per goal:** Around 88 minutes per goal

Haaland’s individual numbers are outstanding. His scoring rate, consistency, and elite rating show that he is one of the most dangerous forwards in world football.

However, winning the FIFA World Cup depends on much more than one player. Norway would need strong squad depth, tactical discipline, defensive quality, and consistent performances throughout the tournament.

So, based on recent performance, Haaland can significantly improve Norway’s chances, but a World Cup title would still be difficult to predict from individual statistics alone.

result

为何这个示例很重要

在“为何此示例重要”这一阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。 在“为何此示例重要”这一阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果之外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。

1. 状态定义

在处理“1状态”阶段时，首先需明确合同规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中管理，以便操作员无需查看整个流程即可进行审计。 在耗时较高的步骤之后设置检查点。当操作员重新执行后续节点时，恢复流程不应再次调用相同的大型语言模型。

2. 工具使用

在处理“工具使用”阶段时，首先写下相关契约：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。 需为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试代理将陷入无止境的循环，耗费大量时间。

3. 规划

在完成三个规划阶段时，首先需写明合同条款：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持透明。 建议使用小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。 在完成三个规划阶段时，首先需写明合同条款：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持透明。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境转向共享环境时出现意外收费。

4. 条件路由

将“4种条件路由”阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，先记录一份理想的运行案例、一个故障案例以及回滚说明。 应将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，这样操作人员无需查看整个架构即可进行审计。

5. 最终分析

可能的改进措施

最终结论

非常感谢您阅读此内容。

如果您希望保持联系或了解更多我的工作，可以在此处找到我：

操作检查清单