过去两年里，LangGraph 几乎成了我在 AI 领域搭建一切的核心。

不论是 chatbots、MCP assistants、voicebots、内部自动化 agents——只要涉及到 reasoning、tools 或 multi-step workflows，我十有八九会用 LangGraph 构建。

它是那个在客户项目、个人实验，甚至每天跑在生产里的系统中反复出现的技术栈。

1. 什么是 LangGraph？

LangGraph 是构建在 LangChain 之上的库，重点在于为基于 LLM 的应用提供 explicit control flow（显式控制流）。

与其把应用组织成线性 pipeline，LangGraph 会把它建模为一个 图（graph），执行可以分支、循环，并重新访问更早的步骤。

传统 LangChain 工作流通常是 Directed Acyclic Graphs (DAGs)：它们只会向前推进并终止。

这对很多场景足够了，但一旦你想要系统可以迭代、重新评估中间结果，或动态决定下一步做什么，这就会受限。

很多真实世界的 agents 需要循环、重试，或者根据结果调整行为，而不是照着固定序列走。

LangGraph 通过允许图中出现 cycles（循环）来解决这个问题。

Nodes 可以被重复访问，decisions 可以反复做，control flow 是显式处理的，而不是埋在 prompts 里。

因此你可以构建会“plan → act → observe → continue”的 agents，直到达到 stopping condition。

LangGraph: nodes, states, and edges.

在概念层面，LangGraph 围绕一组很小的核心抽象构建。一个 agent 被定义为一个 graph，其中所有步骤共享并更新一个共同的 state。

理解 state、nodes 和 edges，基本就理解了 LangGraph 的工作方式。

1.1. State：agent 的记忆

State 是在图中流动的共享数据结构。每个 node 接收当前 state，并可以返回对其的更新。

你可以把 state 看作 agent 的累积上下文：messages、decisions、tool outputs、中间结果，以及任何需要跨步骤保留的信息。

因为 state 是显式传递的，每个 node 都能访问到到目前为止发生的一切，而不必依赖隐式的 prompt 历史。

1.2. Nodes：聚焦的行为单元

Nodes 是图的可执行单元。每个 node 都代表一个单一且定义清晰的操作，比如：

• 调用一个 LLM
• 调用某个 tool 或 API
• 转换或校验数据
• 进行路由决策

在实践中，LangGraph 在 nodes 足够小且聚焦时效果最佳。只做一件事的 nodes 更容易测试、推理与独立修改，随着 agent 的演进也更易维护。

1.3. Edges：控制流

Edges 定义了执行如何从一个 node 移动到下一个。它们编码了 agent 的 control flow，可以是：

• 顺序的（总是移动到下一个 node）
• 条件的（根据 state 选择下一个 node）
• 循环的（返回到之前的 node）

通过显式定义 edges，LangGraph 让控制流以 graph 的形式可见，而不是藏在 prompt 逻辑里。

这样就更容易理解 agent 何时循环、何时分支、何时终止。

弄清这些核心概念后，我们从理论转向一个具体的示例。

接下来我们将一步步构建一个基于 Strava 的训练 agent，用 state、nodes 和 edges 建模一个会根据真实训练数据自动适应的 workflow。

2. Step-by-step Guide

2.1. 从你要自动化的流程开始

想象你要构建一个作为你个人 Strava-powered training operator 的 AI agent。

你的目标清晰（例如即将到来的比赛），训练约束明确（例如每周 3 次），并希望计划会根据你实际完成的训练自动调整。

这个 agent 应该能够：

• 连接你的 Strava 账号并读取近期活动（跑步、越野跑，…）
• 理解你的目标（比赛距离/日期、目标配速/完赛目标、海拔爬升）
• 生成简洁的一周训练计划，遵守进度提升上限与 deloads
• 对比“计划 vs 实际”以识别缺训/额外/过硬的训练，以及过载风险
• 当出现风险信号时调整下一周计划，并解释变化了什么以及为什么
• 通过 SMTP（已配置时）发送邮件交付计划；若未配置，则在日志/终端中预览

在 LangGraph 中实现一个 agent，一般会遵循相同的五个步骤。

3. Step 1：将工作流拆解为离散步骤

先把要自动化的流程拆解成清晰的离散步骤。

每个步骤成为一个 node：一个职责单一的小函数。

识别出这些步骤后，我们再把它们连起来，描述 agent 的整体流向。

下面的图示展示了这些 nodes 如何在 Strava 训练 agent 的工作流中协同工作。

Mermaid Graph

4. Step 2：明确每个步骤要做什么

有了工作流，我们来确定 graph 中每个 node 的职责，以及它需要什么才能把工作做好。

对于每个步骤，我们会决定：

• 它是什么类型的操作
• 它需要什么上下文（静态 vs 动态）
• 它应该输出什么

因为这个 agent 完全自动运行，并且只发送每周邮件，所以我们不包含任何用户输入或 human-in-the-loop 步骤。

Types of nodes (LangGraph Tutorial)

4.1. Data steps

当我们需要从外部系统检索信息时使用 data steps。

Sync Strava Activities

从拉取原始训练数据开始。

• 这个步骤做什么：获取最近约 90 天的 Strava 活动。
• 需要的上下文：通过 access token 和 client id/secret 进行 Strava 认证。
• 期望结果：包含后续所需指标的近期活动列表（日期、时长、距离、配速、海拔、活动类型）。

4.2. LLM steps

当 agent 需要分析数据、权衡取舍或生成可读输出时，使用 LLM steps。

Summarize Recent Training

接下来把原始活动日志转化为有意义的信号。

• 这个步骤做什么：把活动聚合为更高层的训练指标，比如每周训练量、hard sessions 数量、long run 时长以及一致性。
• 需要的上下文：来自 Strava 的近期活动、基本定义（什么算 hard session、long run、周窗口的界定）
• 期望结果：可用于推理与决策的结构化近期训练摘要。

Evaluate Progress vs Goal

现在评估整体进展。

• 这个步骤做什么：把近期训练信号与用户的目标和约束相比较，以判断整体状态。
• 需要的上下文：训练摘要、目标定义（比赛日期、距离、海拔爬升、目标努力程度）、约束（例如每周 3 次训练）、风险启发式（负荷激增、强度堆叠）。
• 期望结果：对训练状态的结构化评估（on track / behind / overloaded）、置信度、风险标记、推荐策略（keep、adjust、deload）。

Generate Next Week Plan

现在生成实际的计划。

• 这个步骤做什么：基于你的目标与近期训练，在尊重 sessions 数和进度上限的前提下，生成一周训练安排（以公里或时长为单位）。
• 需要的上下文：目标（比赛/日期/距离/海拔）、sessions_per_week，以及近期训练摘要（训练量、强度信号）。
• 期望结果：下一周的结构化计划（每次训练的日期、描述、时长、强度），并保留原始生成结果以便后续比较/调整。

Adjust Plan + Add Warnings

若检测到风险，需要清晰呈现并相应调整训练方案。

• 这个步骤做什么：在必要时修改计划，并在出现过载或不一致时生成简明的警示或注意事项。
• 需要的上下文：风险标记、草拟训练计划。
• 期望结果：最终计划，以及用户需要注意的简短警示列表。

Compose Weekly Email

最后把一切转化为可读内容。

• 这个步骤做什么：生成清晰的周报邮件，含训练计划、简要原理说明，以及任何警示或风险。
• 需要的上下文：最终训练计划、关键摘要信号、警示（若有）
• 期望结果：可直接发送的周报邮件（subject + body）。

4.3. Action steps

当 agent 需要与外部世界交互时使用 action steps。

Send Email

通过交付结果来闭环。

• 这个步骤做什么：把每周训练邮件发送给用户。
• 需要的上下文：收件人邮箱、邮件主题与正文
• 期望结果：邮件成功发送并记录日志。

现在我们已经把步骤分成了 LLM、data 和 action 三类，清楚了 graph 中每个 node 的工作类型。

下一个同样重要的问题是：这些 nodes 如何共享信息？

为此，我们需要设计 agent 的 state——这份共享内存让每个步骤可以在前一步的成果上继续构建。

5. Step 3：设计你的 state

既然知道了每个步骤做什么，接下来要决定这些步骤如何共享信息。

State 是 agent 的共享记忆：nodes 会把它们检索、计算或决定的内容写在那里，以便下游步骤使用。

决定什么应该放进 state，有两个简单规则：如果某个信息需要跨多个步骤持久存在，就把它放进 state；如果同样的信息可以从现有数据再次推导出来，我们就避免存它，而是在需要时再计算。

对于我们的 Strava Training Agent，需要跟踪：

• Raw Strava activities：无法在之后重建，且会被多个步骤复用。
• Training summary：如每周训练量、hard sessions 数量、long run 时长等聚合信号。
• Goal and constraints：静态输入，如比赛日期、距离、每周 sessions 数。
• Evaluation results：对进展、置信度与风险标记的评估，用来驱动下游决策。
• Generated training plan：将要发送给用户的下一周计划。
• Execution metadata：用于调试与恢复的时间戳与标识。

定义我们的 state：

from typing import TypedDict, Literal, List, Dict

class TrainingEvaluation(TypedDict):
    status: Literal["on_track", "behind", "overloaded"]
    confidence: float
    risk_flags: List[str]
    recommendation: Literal["keep", "adjust", "deload"]

class TrainingSession(TypedDict):
    day: str
    description: str
    duration_min: int
    intensity: Literal["easy", "moderate", "hard"]

class StravaTrainingAgentState(TypedDict):
    # Raw Strava data
    activities: List[Dict] | None
    # Aggregated training signals
    training_summary: Dict | None
    # Goal and constraints
    goal: Dict
    sessions_per_week: int
    # Evaluation output
    evaluation: TrainingEvaluation | None
    # Generated plan
    next_week_plan: List[TrainingSession] | None
    # Generated communication
    weekly_email: str | None
    # Execution metadata
    run_id: str
    last_sync_timestamp: str

注意，state 只包含原始或结构化数据。

每个 node 都从共享 state 里读取，做好自己的那一件事，再以结构化形式把结果写回。

6. Step 4：构建你的 nodes

有了工作流和 state，我们就可以把每个步骤落成代码了。

在 LangGraph 中，一个 node 其实就是一个 Python 函数：它把当前 state 作为输入，并返回它想更新的字段。

这种简单是有意为之——把注意力放在系统中的数据如何流经各处，而不是框架细节。

所有 nodes 的完整实现见 GitHub 仓库。在本文中，我们重点看最关键的一个：Generate Next Week Plan。

为了让示例更清晰，我们用一个极简的 ChatOpenAI 封装。这个封装只是把 LLM 调用集中起来，保持 node 逻辑清爽。

from dataclasses import dataclass
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

@dataclass
class LLMService:
    model: str = "gpt-4o-mini"
    temperature: float = 0.1

    def __post_init__(self) -> None:
        self.client = ChatOpenAI(model=self.model, temperature=self.temperature)

    def structured_completion(self, system: str, user: str) -> str:
        return self.client.invoke(
            [SystemMessage(content=system), HumanMessage(content=user)]
        ).content

generate_next_week_plan 这个 node 是 agent “智能”真正落地的地方。

它之前的一切——同步 Strava、汇总训练、评估进展——都是在准备输入。它之后的一切——细化或传达结果——则是在消费输出。

这个 node 的职责很直接：给定当前 state，产出一个尊重既定策略与约束的一周训练计划。

这个 node 完全是 state-driven 的。

它读取运动员的目标、近期训练的简要摘要、当前推荐（keep、adjust 或 deload），以及每周 sessions 数。不需要额外上下文。

System prompt 把模型定位为一名跑步教练，并设定护栏：只返回结构化 JSON、严格遵守 sessions 数、非 deload 场景下限制进度提升上限、描述简洁。

User prompt 则注入来自 state 的实际上下文。

def generate_next_week_plan(state: StravaTrainingAgentState) -> dict:
    """LangGraph node: generate next week's training plan."""
    llm: LLMService = _get_service("llm")
    goal = state.get("goal", {})
    summary = state.get("training_summary", {})
    sessions = state.get("sessions_per_week", 3)
    recommendation = (state.get("evaluation") or {}).get("recommendation", "adjust")

    system_prompt = (
        "You are a running coach. Return ONLY valid JSON: "
        "a list of sessions with fields: day (Mon-Sun), description, "
        "duration_min, intensity (easy|moderate|hard). "
        "Respect sessions_per_week exactly. Cap progression at +10% unless deloading (20–30% down)."
    )
    user_prompt = (
        f"Goal: {json.dumps(goal)}\n"
        f"Recent summary: {json.dumps(summary)}\n"
        f"Recommendation: {recommendation}\n"
        f"Sessions per week: {sessions}"
    )
    plan = llm.structured_completion(system_prompt, user_prompt)
    return {"next_week_plan": plan}

7. Step 5：把它们串起来

实现好所有 nodes 后，最后一步就是把它们连接成可运行的 LangGraph 工作流。

这时 agent 的 control flow 会变得清晰可见：我们定义哪个 node 之后运行哪个，哪里分支，何时结束。

先用我们的 state schema 创建一个 StateGraph，并注册每个 node：

# Initialize graph
graph = StateGraph(StravaTrainingAgentState)

# Add nodes
graph.add_node("sync_strava_activities", sync_strava_activities)
graph.add_node("summarize_recent_training", summarize_recent_training)
graph.add_node("evaluate_progress_vs_goal", evaluate_progress_vs_goal)
graph.add_node("generate_next_week_plan", generate_next_week_plan)
graph.add_node("adjust_plan_add_warnings", adjust_plan_add_warnings)
graph.add_node("compose_weekly_email", compose_weekly_email)
graph.add_node("send_email", send_email)

接着添加定义执行流向的 edges。

大多数 edges 很简单，是顺序执行：一个 node 总是接着另一个。

关键点是 generate_next_week_plan 之后的 conditional edge。

此时 agent 已经产出了计划，但我们不总是以相同方式处理它。

如果评估结果表明一切正常（keep），我们可以直接发布。如果推荐是 adjust 或 deload，我们会路由到额外的 node 去适当缓和计划并附上警示。

在 LangGraph 中，add_conditional_edges 需要：

• 分支的起点 node，
• 返回一个 key 的路由函数，
• 从 key 到下一个 node 的映射。

这里的路由函数只是从 state 里读出 recommendation：

# Add edges
graph.add_edge(START, "sync_strava_activities")
graph.add_edge("sync_strava_activities", "summarize_recent_training")
graph.add_edge("summarize_recent_training", "evaluate_progress_vs_goal")
graph.add_edge("evaluate_progress_vs_goal", "generate_next_week_plan")

graph.add_conditional_edges(
    "generate_next_week_plan",
    lambda state: state.get("evaluation", {}).get("recommendation", "adjust"),
    {
        "keep": "compose_weekly_email",
        "adjust": "adjust_plan_add_warnings",
        "deload": "adjust_plan_add_warnings",
    },
)

graph.add_edge("adjust_plan_add_warnings", "compose_weekly_email")
graph.add_edge("compose_weekly_email", "send_email")
graph.add_edge("send_email", END)

# Compile graph
app = graph.compile()

最后，我们把剩余的 nodes 连接起来并编译这个 graph。

8. Step 6：测试这个 agent

要在本地测试整个 agent，先克隆 GitHub 仓库并按 README 的设置步骤进行。

创建（或激活）虚拟环境、安装依赖，并在 .env 中配置所需的 Strava 和 OpenAI 凭据后，只需一个命令即可运行 agent：

python strava_training_agent.py

每次运行时，agent 会拉取你近期的 Strava 活动、汇总你的训练、生成下一周计划，并撰写每周邮件。

SMTP 配置是可选项。如果未设置邮件相关环境变量，agent 会打印并记录邮件预览而不是发送邮件。

这样你就可以不先搭建邮件基础设施，也能端到端地测试完整工作流。

8.1. 自动化运行

当 agent 在本地运行良好后，下一个问题是如何自动运行它。

由于这个工作流只需要每周执行一次，并不需要常驻基础设施。

对于个人项目和教程，GitHub Actions 往往是最简单的选择。

仓库包含一个 weekly-agent.yml workflow，用 cron 定时触发并运行与你本地相同的脚本。

Strava 和 OpenAI（以及可选的 SMTP）凭据存放在加密的 GitHub Secrets 中，由 GitHub 负责执行与日志。这个方案投入小、可靠，也便于他人复现。

9. 最后的想法

本教程的目标不是做出最复杂的训练系统，而是展示 LangGraph 如何以显式、可测试、易于推理的方式来组织 agentic workflows。

把问题建模为一个 graph——共享的 state、小而专注的 nodes，以及清晰定义的 control flow——你就能得到一个行为可见且可预测的 agent，即便它会根据输入变化进行自适应。

同样的模式远不止适用于 Strava 这个例子：客服 agents、数据分析 assistants、内部工具，或任何需要 reasoning、action 与 loop 的系统都适用。

当然，agent 本身也有不少明显的改进方向。

在本教程中，我们有意保持 prompting 简单，以便把注意力聚焦在架构上。

若在 prompt design 上投入更多——增加更丰富的教练语境、更清晰的进度规则、恢复启发式，或特定运动的知识——不改 graph 的情况下，就能显著提升训练计划质量。工作流搭好后，prompting 往往是最具杠杆的改进点。

在此基础上，你可以延展用户反馈回路、跨训练周期的长期记忆、加入更多数据源（如睡眠或 HRV），或更严格的安全检查以在风险信号累积时暂停进度。

关键在于，这些改进都是增量式的。

只要你能把问题拆成 nodes、谨慎定义 state，并有意识地“布线”graph，你就已经有了用 LangGraph 构建稳健、可适应 agents 的坚实基础。