跳到主要内容

快速开始指南

安装​

# 使用pip安装
pip install langgraph langchain

# 或使用uv
uv add langgraph langchain

第一个图：计算器​

让我们构建一个简单的计算器代理。

第1步：安装依赖​

pip install langchain langchain-anthropic
export ANTHROPIC_API_KEY="your-api-key"

第2步：定义状态​

from typing import TypedDict

class CalculatorState(TypedDict):
    expr: str              # 要计算的表达式
    result: int | None    # 结果
    iteration: int        # 迭代次数

第3步：定义节点​

def calculator_node(state: CalculatorState):
    """执行计算"""
    try:
        result = eval(state["expr"])
        return {
            "result": result,
            "iteration": state["iteration"] + 1
        }
    except Exception as e:
        return {
            "result": None,
            "iteration": state["iteration"] + 1
        }

def validator_node(state: CalculatorState):
    """验证表达式"""
    if not state["expr"]:
        raise ValueError("表达式不能为空")
    
    return state

第4步：构建图​

from langgraph.graph import StateGraph, START, END

# 创建图
graph = StateGraph(CalculatorState)

# 添加节点
graph.add_node("validate", validator_node)
graph.add_node("calculate", calculator_node)

# 添加边
graph.add_edge(START, "validate")
graph.add_edge("validate", "calculate")
graph.add_edge("calculate", END)

# 编译
app = graph.compile()

第5步：运行​

# 执行
result = app.invoke({
    "expr": "2 + 3 * 4",
    "result": None,
    "iteration": 0
})

print(f"结果: {result['result']}")      # 14
print(f"迭代: {result['iteration']}")   # 1

第二个图：条件分支​

演示如何使用条件边。

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

class DecisionState(TypedDict):
    score: int
    decision: str

def analyze_score(state: DecisionState):
    """分析分数"""
    return {"score": state["score"]}

def route_decision(state: DecisionState) -> Literal["high", "low"]:
    """根据分数路由"""
    if state["score"] >= 70:
        return "high"
    else:
        return "low"

def high_score_handler(state: DecisionState):
    """处理高分"""
    return {"decision": "优秀"}

def low_score_handler(state: DecisionState):
    """处理低分"""
    return {"decision": "需要改进"}

# 构建图
graph = StateGraph(DecisionState)

graph.add_node("analyze", analyze_score)
graph.add_node("high_handler", high_score_handler)
graph.add_node("low_handler", low_score_handler)

graph.add_edge(START, "analyze")

# 添加条件边
graph.add_conditional_edges(
    "analyze",
    route_decision,
    {
        "high": "high_handler",
        "low": "low_handler"
    }
)

graph.add_edge("high_handler", END)
graph.add_edge("low_handler", END)

app = graph.compile()

# 测试
result1 = app.invoke({"score": 85})
print(result1["decision"])  # 优秀

result2 = app.invoke({"score": 60})
print(result2["decision"])  # 需要改进

第三个图：循环​

实现可以重试的工作流。

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

class RetryState(TypedDict):
    task: str
    attempts: int
    success: bool

def execute_task(state: RetryState):
    """执行任务"""
    import random
    
    state["attempts"] += 1
    state["success"] = random.random() > 0.5
    
    return state

def should_retry(state: RetryState) -> Literal["execute", "end"]:
    """决定是否重试"""
    if state["success"] or state["attempts"] >= 3:
        return "end"
    else:
        return "execute"

# 构建图
graph = StateGraph(RetryState)

graph.add_node("execute", execute_task)

graph.add_edge(START, "execute")

graph.add_conditional_edges(
    "execute",
    should_retry,
    {
        "execute": "execute",  # 循环回自己
        "end": END
    }
)

app = graph.compile()

# 运行
result = app.invoke({"task": "test", "attempts": 0, "success": False})
print(f"任务成功: {result['success']}")
print(f"尝试次数: {result['attempts']}")

流式执行​

通过流式处理查看执行过程。

# 流式获取结果
for output in app.stream({"expr": "5 + 3", "result": None, "iteration": 0}):
    print(f"节点输出: {output}")
    # 输出:
    # {'validate': {...}}
    # {'calculate': {...}}

调试技巧​

1. 查看图结构​

# 获取图的可视化
try:
    image = app.get_graph().draw_mermaid_png()
    print("图已生成")
except Exception as e:
    print(f"无法生成图: {e}")

2. 添加日志​

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

def logged_node(state):
    logger.info(f"处理状态: {state}")
    # ... 处理逻辑
    logger.info(f"返回状态: {state}")
    return state

3. 逐步调试​

# 在每个节点后添加打印
result = app.invoke(initial_state)

for step in result.get("steps", []):
    print(f"步骤: {step['node']}")
    print(f"状态: {step['state']}")

常见问题​

Q: 图中的节点顺序是什么？ A: 由边的定义决定。从START开始，按照add_edge的指定顺序执行。

Q: 如何在节点间传递数据？ A: 通过状态字典。每个节点可以读取和修改状态。

Q: 可以有多个入口点吗？ A: 开始时只能从START开始，但可以通过invoke参数定制初始状态。

Q: 如何处理节点中的异常？ A: 在节点函数中使用try-except，或在图级别添加错误处理节点。

下一步​

现在你已经掌握了基础。接下来可以学习：