前三周的 ops-agent 已经可以通过 LangChain create_agent 调用只读工具：查询日志、指标、服务状态和 Runbook。它适合快速验证 Tool Calling，但当流程开始出现固定步骤、条件分支、并行采集和路径测试时，仅依靠 Agent 循环就不够清晰了。

LangGraph 的价值不是让模型变得更聪明，而是把执行流程显式表达出来：状态由谁更新，下一步执行什么，哪些节点可以并行，以及失败后最终处于什么状态。

1. 先理解最小模型

LangGraph 可以先记成三个概念：

State：流程当前拥有的数据
Node：读取 State，并返回 State 更新的函数
Edge：决定下一个 Node 的连接

一个 Graph 的基本生命周期是：

定义 State
   ↓
添加 Node
   ↓
添加 Edge
   ↓
compile()
   ↓
invoke(input)

安装本周依赖：

cd ops-agent
python3 -m pip install -e ".[llm]"

1.1 第一个 Graph：一个 Node

from typing import TypedDict
from langgraph.graph import END, START, StateGraph


class State(TypedDict, total=False):
    message: str


def hello(state: State) -> dict[str, str]:
    return {"message": f"Hello, {state.get('message', 'LangGraph')}"}


builder = StateGraph(State)
builder.add_node("hello", hello)
builder.add_edge(START, "hello")
builder.add_edge("hello", END)
graph = builder.compile()

print(graph.invoke({"message": "LangGraph"}))
# {'message': 'Hello, LangGraph'}

这里没有 LLM，也没有 Tool。START 和 END 是图的入口和出口，hello 是普通 Python 函数。

2. LangGraph 中的 Node 是什么

在 LangGraph 中，Node 通常不是一个需要继承的类，而是一个可调用对象，最常见的形式就是 Python 函数。它有三个关键特征：

输入：当前 State
处理：执行一段业务逻辑
输出：部分 State 更新，或 Command

下面的 hello 就是一个 Node：

def hello(state: State) -> dict[str, str]:
    return {"message": f"Hello, {state.get('message', 'LangGraph')}"}

它的输入是当前 State，输出是一个字典。这个字典不是新的完整 State，而是要写回 Graph 的局部更新。LangGraph 会根据 State 定义，把返回值应用到当前状态中。

builder.add_node("hello", hello)

这行代码做了两件事：

  1. 用字符串 "hello" 作为 Graph 中的节点名称；
  2. 把 hello 这个可调用对象注册为该节点的执行逻辑。

注册 Node 不会立即执行函数。真正的执行顺序由 Edge 决定，调用 compile() 得到可运行的 Graph，调用 invoke() 时才会执行：

add_node()  → 注册 Node
add_edge()  → 定义连接
compile()   → 构建可执行 Graph
invoke()    → 按路径执行 Node

Node 可以是同步函数，也可以是异步函数。基础阶段先使用同步函数即可；后续接入网络 Tool 或异步模型时，再考虑 async def。节点本身不应该偷偷改变外部全局变量，应该通过返回 State 更新把结果交给 Graph 管理。

2.1 START 和 END：两个特殊节点

START 和 END 也经常被称为特殊节点，但它们和普通 Node 有一个重要区别：它们是 LangGraph 提供的虚拟入口和出口，不需要我们编写函数，也不会执行具体业务逻辑。

名称 作用 是否需要实现函数
START 表示 Graph 的执行入口，把初始输入交给第一个业务 Node 不需要
END 表示 Graph 的执行出口，流程到达这里后结束 不需要

因此，最小 Graph 中的两条边分别表示：

builder.add_edge(START, "hello")  # 从入口进入 hello
builder.add_edge("hello", END)    # hello 完成后结束

它们和普通 Node 的区别可以这样理解：

START ──▶ 业务 Node ──▶ 业务 Node ──▶ END
虚拟入口       会执行函数                 虚拟出口

START 和 END 不能作为普通业务函数使用，也不需要通过 add_node() 注册。只要使用它们连接 Graph，就能清楚表达流程从哪里开始、在哪里结束。

3. State 如何更新

运行 examples/02_state_update.py，先看完整的 State 定义：

import operator
from typing import Annotated, TypedDict
from langgraph.graph import END, START, StateGraph


class State(TypedDict, total=False):
    count: int
    history: Annotated[list[str], operator.add]

State 有两个作用：

  1. 声明 Graph 中允许流转的数据字段和类型；
  2. 为每个字段声明更新方式。

其中 count 没有 Reducer，节点返回新值时会覆盖旧值；history 使用 operator.add，多个节点返回的列表会被追加，而不是互相覆盖。

接着定义两个节点：

def increment(state: State) -> dict[str, object]:
    count = state.get("count", 0) + 1
    return {"count": count, "history": [f"count={count}"]}


def finish(state: State) -> dict[str, object]:
    return {"history": [f"finished at {state['count']}"]}

increment 读取当前 State，并且只返回本次需要更新的字段。Graph 会按照 Edge 顺序把更新交给下一个节点：

builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_node("finish", finish)
builder.add_edge(START, "increment")
builder.add_edge("increment", "finish")
builder.add_edge("finish", END)
graph = builder.compile()

result = graph.invoke({"count": 0, "history": []})
print(result)
# {'count': 1, 'history': ['count=1', 'finished at 1']}

这里的生效过程是：

初始 State：count=0, history=[]
        ↓ increment
更新：count=1, history=["count=1"]
        ↓ finish
更新：history=["finished at 1"]
        ↓ Reducer(operator.add)
最终：count=1, history=["count=1", "finished at 1"]

也就是说，State 不是只用于类型提示；它直接决定了节点更新如何应用到 Graph 的共享状态中。

python3 examples/01_hello_graph.py
python3 examples/02_state_update.py

4. Conditional Edge 如何分支

examples/03_conditional_edge.py 根据 value 的奇偶性选择不同 Node。Conditional Edge 的核心接口可以先记成：

builder.add_conditional_edges(source, path, path_map)

三个参数的作用是：

参数 含义
source 从哪个 Node 执行完之后开始判断，例如 "classify"
path 路由函数，接收最新 State，返回路由值或目标 Node 名称
path_map 可选映射，把路由函数返回值映射成真正的目标 Node

下面的例子中，classify 先根据输入写入 route：

class State(TypedDict, total=False):
    value: int
    route: str
    result: str


def classify(state: State) -> dict[str, str]:
    return {"route": "even" if state.get("value", 0) % 2 == 0 else "odd"}


def route(state: State) -> Literal["even", "odd"]:
    return state["route"]

再把路由函数注册为 classify 执行后的条件边：

builder.add_conditional_edges(
    "classify",
    route,
    {"even": "even", "odd": "odd"},
)

它的执行顺序不是在构建 Graph 时判断，而是在运行时按下面的过程执行：

1. 执行 classify(state)
2. 应用 classify 返回的 State 更新：route="even"
3. 调用 route(最新 State)
4. route 返回 "even"
5. 用 path_map 找到目标 Node："even"
6. 执行 even Node

path_map 的作用是把“业务路由值”和“Graph 节点名称”解耦。例如：

def route(state: State) -> Literal["needs_review", "normal"]:
    return "needs_review" if state["value"] > 10 else "normal"


builder.add_conditional_edges(
    "classify",
    route,
    {
        "needs_review": "manual_review_node",
        "normal": "answer_node",
    },
)

如果不传 path_map，路由函数返回值就必须直接是目标 Node 名称：

builder.add_conditional_edges("classify", route)

路由函数只负责决定下一步，不负责生成业务结果；业务结果由 even 或 odd Node 生成。一个 source Node 应选择一种路由方式，不要同时为它配置普通静态边和 Conditional Edge，否则两条路径都可能被执行。运行：

python3 examples/03_conditional_edge.py

5. Parallel 与 Reducer

examples/04_parallel_reducer.py 从 START 同时进入三个读取节点：

START
  ├── read_logs
  ├── read_metrics
  └── read_runbook
             ↓
           merge

三个节点都写入同一个 evidence 字段，因此 State 必须声明 Reducer：

class State(TypedDict, total=False):
    evidence: Annotated[list[str], operator.add]
    summary: list[str]

这里的 Reducer 不是 Python 的特殊类型，而是 LangGraph 对“State 更新如何合并”的称呼。Python 只提供了两部分：list[str] 表示字段类型，operator.add 是一个普通的 Python 函数；Annotated 用来把这个函数作为元数据附加到字段上。LangGraph 读取这段声明后，才会把它当作该字段的 Reducer。

默认情况下，节点返回的字段会覆盖旧值。例如：

summary: list[str]

# 节点 A 返回
{"summary": ["logs"]}
# 节点 B 再返回
{"summary": ["metrics"]}
# 默认结果：
["metrics"]

声明 Reducer 后，LangGraph 会在应用节点更新时调用合并函数：

evidence: Annotated[list[str], operator.add]

# 近似理解为：
operator.add([], ["logs"])
operator.add(["logs"], ["metrics"])
operator.add(["logs", "metrics"], ["runbook"])

所以 Reducer 是“框架如何处理 Python 返回值”的合并规则，不是 Python 本身额外增加的语法，也不是线程锁。

在这个示例中，Parallel 由三条从 START 出发的边体现：

builder.add_edge(START, "read_logs")
builder.add_edge(START, "read_metrics")
builder.add_edge(START, "read_runbook")

这三条边的目标节点属于同一个执行步骤，Graph 会在这一轮调度三个节点。三个节点分别只返回自己的证据：

def read_logs(state: State) -> dict[str, list[str]]:
    return {"evidence": ["logs"]}


def read_metrics(state: State) -> dict[str, list[str]]:
    return {"evidence": ["metrics"]}


def read_runbook(state: State) -> dict[str, list[str]]:
    return {"evidence": ["runbook"]}

Reducer 由这一行声明：

evidence: Annotated[list[str], operator.add]

它表示：当多个节点同时更新 evidence 时，不要让后写入的值覆盖前面的值，而是使用 operator.add 合并列表。因此本次执行相当于：

初始：[]
read_logs：    [] + ["logs"]
read_metrics：    + ["metrics"]
read_runbook：    + ["runbook"]
最终 evidence：包含 "logs"、"metrics"、"runbook"（顺序不应作为业务保证）

三条边汇聚到 merge，形成 Fan-in：

builder.add_edge("read_logs", "merge")
builder.add_edge("read_metrics", "merge")
builder.add_edge("read_runbook", "merge")

这里的三条入边不会让 merge 被调用三次。对于一次 graph.invoke()：

read_logs、read_metrics、read_runbook 各执行一次
        ↓
LangGraph 等待三个上游节点完成，并合并它们的 State 更新
        ↓
merge 只执行一次，读取合并后的完整 State
        ↓
merge → END

因此，merge 是 Fan-in 的汇聚节点，不是每收到一条入边就执行一次。示例中的 trace 字段专门用来验证这一点：最终结果中 merge 的出现次数应该是 1。

merge 读取已经合并好的 evidence，只写入 summary：

def merge(state: State) -> dict[str, list[str]]:
    return {"summary": sorted(state.get("evidence", []))}

summary 只有 merge 一个节点写入，所以不需要 Reducer。若删除 evidence 上的 operator.add，三个并行节点同时写同一字段时，Graph 就不知道应该覆盖还是合并，通常会报并发 State 更新错误。

运行：

python3 examples/04_parallel_reducer.py

输出中的 evidence 是 Reducer 合并结果，summary 是 merge 节点对合并结果进行排序后的结果：

{
    "evidence": ["logs", "metrics", "runbook"],  # 具体顺序不应依赖
    "summary": ["logs", "metrics", "runbook"],
    "trace": ["read_logs", "read_metrics", "read_runbook", "merge"],
}

到这里，已经覆盖了 LangGraph 基础的最小知识闭环：顺序执行、State 更新、条件分支、并行和状态合并。

6. 从 Agent 循环到 Workflow

一个简单 Agent 大致是：

用户问题
   ↓
模型决定是否调用 Tool
   ↓
执行 Tool
   ↓
把结果交回模型
   ↓
模型继续决定下一步

这套方式的优点是灵活，但它也把流程控制权交给了模型。对于运维故障诊断，我希望下面这些步骤是确定的：

接收事件 → 校验 → 分类 → 收集证据 → 分析 → 验证

因此本周把流程改造成：

START
  ↓
receive_incident
  ↓
validate_incident ── invalid ──→ validation_failed → END
  ↓ valid
classify_incident
  ↓
select_service_flow
  ├── redis_flow
  ├── mysql_flow
  └── generic_flow
  ↓
query_logs + query_metrics + search_runbook
  ↓
merge_evidence
  ↓
analyze_incident
  ↓
verify_diagnosis → END

7. 把基础概念组合到 ops-agent

前面几个章节使用最小示例解释了 LangGraph 的通用机制。下面回到实际项目，看看这些机制如何组合成企业运维故障诊断流程。

7.1 IncidentState

在 ops-agent/app/graph/state.py 中，State 使用 TypedDict 描述：

class IncidentState(TypedDict, total=False):
    incident_id: str
    raw_input: str
    service: str
    tool_service: str
    severity: str
    service_flow: str
    validation_errors: list[str]
    raw_evidence: Annotated[list[Evidence], operator.add]
    evidence: list[Evidence]
    diagnosis: Diagnosis
    diagnosis_error: str
    status: str
    trace: Annotated[list[str], operator.add]

在 ops-agent 中，State 是整个故障诊断流程的共享快照：

  • service 表示分类结果；
  • raw_evidence 表示并行节点产生的原始证据；
  • diagnosis 表示结构化分析结果；
  • trace 用来证明实际经过了哪些节点；
  • status 表示流程最后是成功、工具部分失败还是模型失败。

项目节点仍然只返回本次需要更新的字段，例如分类节点只负责写入服务和严重等级：

def classify_incident(state: IncidentState) -> dict[str, Any]:
    return {
        "service": "redis",
        "severity": "medium",
        "trace": ["classify_incident"],
    }

7.2 项目中的 Conditional Edge

分类节点只负责写入 service，路由函数负责把状态映射到节点：

builder.add_conditional_edges(
    "select_service_flow",
    _service_route,
    {
        "redis_flow": "redis_flow",
        "mysql_flow": "mysql_flow",
        "generic_flow": "generic_flow",
    },
)

对于 Redis 事件，测试可以明确断言：

self.assertIn("redis_flow", result["trace"])
self.assertNotIn("mysql_flow", result["trace"])

这和“模型大概率会选择正确工具”不是一回事。前者是可测试的业务约束，后者只是一次生成结果。

7.3 项目中的 Parallel 与 Reducer

日志、指标和 Runbook 之间没有前后依赖，因此可以并行：

prepare_context
      ├── query_logs
      ├── query_metrics
      └── search_runbook
                    ↓
              merge_evidence

如果多个节点在同一个 super-step 中更新同一个 State 字段，就必须定义合并规则。项目使用：

raw_evidence: Annotated[list[Evidence], operator.add]

但没有让 merge_evidence 再次追加到这个字段，否则会造成重复。并行节点先写 raw_evidence，合并节点再覆盖生成排序稳定的 evidence：

def merge_evidence(state: IncidentState) -> dict[str, Any]:
    evidence = sorted(
        state.get("raw_evidence", []),
        key=lambda item: (item.get("kind", ""), item.get("source", "")),
    )
    return {"evidence": evidence, "trace": ["merge_evidence"]}

这里有一个容易忽略的工程问题：并行执行顺序不是业务顺序。测试应该比较证据集合或合并后的稳定顺序，而不是断言三个 Tool 哪一个先完成。

8. LLM 只是一个 Node

analyze_incident 使用前三周已经学习过的 Structured Output：

class Diagnosis(BaseModel):
    summary: str
    possible_causes: list[str]
    confidence: float = Field(ge=0, le=1)
    recommended_actions: list[str]

Graph 控制：

证据收集完成 → 调用分析模型 → 校验诊断 → 结束

模型不负责跳过校验，也不负责执行高风险动作。测试中注入 Fake Model，真实模型只是可选的手工实验：

result = build_incident_graph(model=FakeStructuredModel()).invoke(
    {"incident_id": "inc-001", "raw_input": "Redis timeout"}
)

如果模型抛出异常，Graph 最终状态是 failed，不会把空结果当成成功诊断。

9. Agent Node 与固定 Workflow

固定流程和动态研究并不冲突。项目支持把现有只读 Agent 放入 Graph：

merge_evidence
      ↓
research_agent
      ↓
analyze_incident

这个 Agent 只能使用：

query_logs
query_metrics
search_runbook

因此职责边界变成：

Graph：决定阶段和业务边界
Agent：在研究阶段动态选择只读工具
Tool Policy：决定调用是否允许

并不是每个节点都需要 LLM。校验、路由、合并和最终状态检查都应该保持普通 Python 节点。

10. Command 什么时候使用

本周还实现了独立的 Command 示例：

return Command(
    update={"value": "...|recorded"},
    goto="finish",
)

它适合“更新 State 的同时决定下一节点”的场景。对于普通固定流程，Conditional Edge 更直观；不要在同一节点上同时配置静态出边和 Command(goto=...)，否则可能产生重复路径。

11. 为什么 Graph 更容易测试

本周新增的测试不只检查最终答案，还检查执行路径：

test_graph_basics.py       Hello、State、Conditional、Reducer 基础示例
test_graph_compile.py      Graph 能否编译和运行
test_graph_validation.py   非法输入是否提前结束
test_graph_routing.py      Redis / MySQL / Generic 路由
test_graph_parallel.py     三个证据节点和 Reducer
test_graph_llm.py          结构化输出、模型失败和 Agent Node

运行：

cd ops-agent
python3 -m unittest discover -s tests -v

当前离线测试覆盖 54 个用例，真实 LLM 不属于测试前提。这样可以把“模型输出不稳定”和“Graph 控制流错误”分开诊断。

12. 当前边界

本周的 OpsGraph V0.1 仍然没有：

Checkpoint
Persistence
Resume
Interrupt
HITL

这些能力会在下一周加入。当前版本的目标只是先把 Workflow 的结构、状态传递、路由、并行和测试方法建立起来。

结论

create_agent 适合快速构建“模型决定下一步”的 Agent；LangGraph 适合把复杂业务拆成可观察、可控制、可测试的 Workflow。

两者不是互相替代：在 ops-agent 中，LangGraph 控制整体流程，Agent 只负责受限的动态研究，Tool 层继续负责 Schema、权限、超时、重试、幂等和审计。

参考资料：