#聊聊Agent开发#

1. 状态（State）

代表应用程序当前快照的共享数据结构

可以使用 TypedDict 或 Pydantic 模型定义

包含作为所有节点和边输入模式的模式（schema）

通过减速器函数（reducer functions）进行更新，指定如何应用更改

1. TypedDict 和 Pydantic 是什么？

它们都是定义状态结构的方式：、

简单说：都是用来定义你的数据结构，就像数据库的字段定义。

#pydantic方式
from pydantic import BaseModel

class State(BaseModel):
    messages: list
    llm_calls: int

#typedDict方式
from typing_extensions import TypedDict, Annotated
import operator

class State(TypedDict):
    messages: list
    llm_calls: int

2. 减速器函数（Reducer Function）是什么？

减速器函数定义了如何合并状态更新。

比如，节点可能返回 {"messages": [new_msg]}，但你想追加到消息列表，而不是覆盖它。

from typing_extensions import TypedDict, Annotated
import operator

# ❌ 没有减速器 - 会被覆盖
class State1(TypedDict):
    messages: list

# ✅ 有减速器 - 会追加
class State2(TypedDict):
    messages: Annotated[list, operator.add]  # 使用 add 函数追加

2. 节点（Nodes）

编码代理逻辑的函数

接收当前状态作为输入

执行计算或副作用

返回更新后的状态

代表图中执行的实际工作

3. 边（Edges）

基于当前状态确定接下来执行哪个节点的函数

可以是条件分支或固定转换

告诉图接下来要做什么

启用图中的路由和控制流

4. 状态图（StateGraph）

构建LangGraph应用程序的主要抽象

允许你定义节点和边的图

编译时自动创建Pregel应用程序

使得组合复杂的循环工作流变得简单

from langgraph.graph import StateGraph, START
from typing_extensions import TypedDict, Annotated
import operator

# 定义状态
class State(TypedDict):
    messages: Annotated[list, operator.add]
    topic: str
    essay: str

# 定义节点
def write_essay(state: State):
    """写文章"""
    essay_content = f"关于 {state['topic']} 的文章内容..."
    print(f"[write_essay 节点] 写了文章: {essay_content}")
    return {"essay": essay_content}

def review_essay(state: State):
    """审阅文章"""
    review = f"审阅意见：文章讲的是 {state['topic']}"
    print(f"[review_essay 节点] 添加审阅: {review}")
    return {"messages": [review]}

# 构建图
builder = StateGraph(State)
builder.add_node("write", write_essay)
builder.add_node("review", review_essay)
builder.add_edge(START, "write")
builder.add_edge("write", "review")

# 编译
graph = builder.compile()

# 运行
print("=== 开始执行图 ===\n")
result = graph.invoke({
    "messages": ["开始"],
    "topic": "AI",
    "essay": ""
})

print("\n=== 最终输出 ===")
print(f"messages: {result['messages']}")
print(f"topic: {result['topic']}")
print(f"essay: {result['essay']}")

实际输出会是这样：=== 开始执行图 === [write_essay 节点] 写了文章: 关于 AI 的文章内容... [review_essay 节点] 添加审阅: 审阅意见：文章讲的是 AI === 最终输出 === messages: ['开始', '审阅意见：文章讲的是 AI'] topic: AI essay: 关于 AI 的文章内容... 

关键点解释：

1.messages 追加了：从 ['开始'] 变成 ['开始', '审阅意见：文章讲的是 AI']因为我们用了 Annotated[list, operator.add] 减速器

2.essay 被设置：从空字符串变成 "关于 AI 的文章内容..."因为 essay 没有减速器，所以直接被覆盖

3.topic 保持不变：因为没有节点修改它，所以还是 "AI"

如果没有减速器会怎样？

class State(TypedDict):
    messages: list  # 没有 operator.add
# 输出会是:
messages: ['审阅意见：文章讲的是 AI']  # 原来的 '开始' 被覆盖了！

关键原则

节点做工作，边决定做什么。 这种设计使控制流显式且可追踪——你总是可以通过查看当前节点和状态来理解代理接下来会做什么。

通过组合节点和边，你可以创建复杂的代理工作流，使状态随时间演变。节点和边仅仅是函数，所以它们可以包含LLM或任何其他代码逻辑。

参考资料：

https://docs.langchain.com/oss/python/langgraph/use-graph-api

https://docs.langchain.com/oss/python/langgraph/graph-api