3.1 提示词模板

3.1 StateGraph 状态图基础 — LangChain框架精通 LangGraph 工作流编排

本节导读：深入学习 LangGraph 的核心概念 StateGraph，掌握状态图的构建、节点定义、边连接和状态管理，为构建复杂的 Agent 工作流奠定基础。

学习目标

• 理解 LangGraph 的核心概念和架构设计
• 掌握 StateGraph 的基本原理和组件结构
• 学习节点（Node）和边（Edge）的定义和使用方法
• 理解状态（State）管理和数据流转机制
• 实现基础的 Agent 工作流和状态转换
• 掌握 LangGraph 的调试和监控技巧

核心概念

LangGraph 架构原理

LangGraph 是 LangChain 提供的工作流编排框架，它基于有向图（Directed Graph）的概念来构建复杂的 Agent 系统。与传统线性链式处理不同，LangGraph 支持循环、分支、条件判断等复杂控制逻辑。

StateGraph 核心组件

1. 状态（State）：定义 Agent 工作流中的数据结构，包含所有共享信息
2. 节点（Node）：执行特定任务的函数或处理单元
3. 边（Edge）：连接节点的路径，定义状态流转的方向
4. 图（Graph）：由节点和边组成的工作流拓扑结构
5. 条件边（Conditional Edge）：根据状态决定下一步的边

StateGraph vs 传统链式架构

环境准备 / 前置知识

基础依赖安装

# LangGraph 核心包
pip install langgraph langchain langchain-openai langchain-community
pip install langchain-tools langchain-memory
​

核心导入模块

import os
from typing import Dict, List, Any, Optional, TypedDict
from langgraph.graph import StateGraph, END, START
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
​

分步实战

步骤 1：基础 StateGraph 构建

首先创建一个简单的 StateGraph 来理解基本概念：

import os
from typing import Dict, List, Any, TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage

# 设置 API 密钥
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# 定义状态类型
class AgentState(TypedDict):
    """Agent 状态定义"""
    messages: List[HumanMessage | AIMessage]
    current_step: str
    is_complete: bool

# 创建简单的节点函数
def start_node(state: AgentState) -> AgentState:
    """开始节点"""
    print("=== 开始处理 ===")
    return {
        **state,
        "current_step": "started",
        "is_complete": False
    }

def process_node(state: AgentState) -> AgentState:
    """处理节点"""
    print(f"=== 处理消息: {state['messages'][-1].content} ===")
    
    # 创建聊天模型
    llm = ChatOpenAI(
        model="gpt-4-turbo",
        temperature=0.7,
        max_tokens=1000
    )
    
    # 调用 LLM
    response = llm.invoke(state["messages"])
    
    return {
        **state,
        "messages": state["messages"] + [response],
        "current_step": "processed"
    }

def end_node(state: AgentState) -> AgentState:
    """结束节点"""
    print("=== 处理完成 ===")
    return {
        **state,
        "is_complete": True,
        "current_step": "completed"
    }

# 创建状态图
def create_basic_graph():
    """创建基础状态图"""
    
    # 定义初始状态
    initial_state: AgentState = {
        "messages": [HumanMessage(content="你好，请介绍一下自己")],
        "current_step": "initial",
        "is_complete": False
    }
    
    # 创建图
    workflow = StateGraph(AgentState)
    
    # 添加节点
    workflow.add_node("start", start_node)
    workflow.add_node("process", process_node)
    workflow.add_node("end", end_node)
    
    # 添加边
    workflow.add_edge(START, "start")
    workflow.add_edge("start", "process")
    workflow.add_edge("process", "end")
    workflow.add_edge("end", END)
    
    # 编译图
    app = workflow.compile()
    
    return app, initial_state

# 使用基础图
print("=== 基础 StateGraph 测试 ===")
app, initial_state = create_basic_graph()

# 运行图
result = app.invoke(initial_state)

print("\n最终状态：")
print(f"当前步骤：{result['current_step']}")
print(f"是否完成：{result['is_complete']}")
print(f"消息数量：{len(result['messages'])}")
print(f"最后回复：{result['messages'][-1].content}")
​

步骤 2：条件边的使用

实现基于条件的节点选择和状态流转：

from typing import Literal

# 定义更多状态类型
class EnhancedAgentState(TypedDict):
    """增强的 Agent 状态"""
    messages: List[HumanMessage | AIMessage]
    current_step: str
    is_complete: bool
    task_type: str
    processing_result: Optional[str]

# 条件判断函数
def determine_next_step(state: EnhancedAgentState) -> Literal["analysis", "writing", "coding", "end"]:
    """根据任务类型决定下一步"""
    task_type = state.get("task_type", "")
    
    if "分析" in task_type or "分析" in state["messages"][-1].content:
        return "analysis"
    elif "写作" in task_type or "文章" in state["messages"][-1].content:
        return "writing"
    elif "编程" in task_type or "代码" in state["messages"][-1].content:
        return "coding"
    else:
        return "end"

# 不同任务的节点函数
def analysis_node(state: EnhancedAgentState) -> EnhancedAgentState:
    """分析任务节点"""
    print("=== 执行分析任务 ===")
    
    llm = ChatOpenAI(
        model="gpt-4-turbo",
        temperature=0.3,
        max_tokens=1500
    )
    
    prompt = f"""
你是一个分析专家，请对以下内容进行详细分析：

用户需求：{state['messages'][-1].content}

要求：
1. 识别核心需求和目标
2. 分析问题的关键点
3. 提供可行的解决方案
4. 总结分析结果

请用专业、清晰的语言回答。
"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    return {
        **state,
        "processing_result": response.content,
        "current_step": "analysis_completed"
    }

def writing_node(state: EnhancedAgentState) -> EnhancedAgentState:
    """写作任务节点"""
    print("=== 执行写作任务 ===")
    
    llm = ChatOpenAI(
        model="gpt-4-turbo",
        temperature=0.8,
        max_tokens=2000
    )
    
    prompt = f"""
你是一个专业写作专家，请根据以下用户需求创作内容：

用户需求：{state['messages'][-1].content}

要求：
1. 内容要有逻辑性和连贯性
2. 语言要生动有趣
3. 结构要清晰完整
4. 符合用户期望的写作风格

请创作高质量的内容。
"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    return {
        **state,
        "processing_result": response.content,
        "current_step": "writing_completed"
    }

def coding_node(state: EnhancedAgentState) -> EnhancedAgentState:
    """编程任务节点"""
    print("=== 执行编程任务 ===")
    
    llm = ChatOpenAI(
        model="gpt-4-turbo",
        temperature=0.2,
        max_tokens=2000
    )
    
    prompt = f"""
你是一个编程专家，请根据以下用户需求编写代码：

用户需求：{state['messages'][-1].content}

要求：
1. 代码要简洁高效
2. 要有完整的注释说明
3. 包含必要的错误处理
4. 提供使用示例

请编写高质量的代码。
"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    return {
        **state,
        "processing_result": response.content,
        "current_step": "coding_completed"
    }

def end_node(state: EnhancedAgentState) -> EnhancedAgentState:
    """结束节点"""
    print("=== 任务完成 ===")
    return {
        **state,
        "is_complete": True,
        "current_step": "final"
    }

# 创建条件图
def create_conditional_graph():
    """创建条件状态图"""
    
    # 定义初始状态
    initial_state: EnhancedAgentState = {
        "messages": [HumanMessage(content="我需要分析一个数据分析项目")],
        "current_step": "initial",
        "is_complete": False,
        "task_type": "数据分析",
        "processing_result": None
    }
    
    # 创建图
    workflow = StateGraph(EnhancedAgentState)
    
    # 添加节点
    workflow.add_node("analysis", analysis_node)
    workflow.add_node("writing", writing_node)
    workflow.add_node("coding", coding_node)
    workflow.add_node("end", end_node)
    
    # 添加边
    workflow.add_edge(START, "analysis")
    workflow.add_conditional_edges(
        "analysis",
        determine_next_step,
        {
            "analysis": "analysis",  # 继续分析
            "writing": "writing",    # 转到写作
            "coding": "coding",      # 转到编程
            "end": "end"            # 结束
        }
    )
    workflow.add_conditional_edges(
        "writing",
        determine_next_step,
        {
            "analysis": "analysis",
            "writing": "writing",
            "coding": "coding",
            "end": "end"
        }
    )
    workflow.add_conditional_edges(
        "coding",
        determine_next_step,
        {
            "analysis": "analysis",
            "writing": "writing",
            "coding": "coding",
            "end": "end"
        }
    )
    workflow.add_edge("end", END)
    
    # 编译图
    app = workflow.compile()
    
    return app, initial_state

# 使用条件图
print("=== 条件 StateGraph 测试 ===")
app, initial_state = create_conditional_graph()

# 运行图
result = app.invoke(initial_state)

print("\n最终结果：")
print(f"任务类型：{result['task_type']}")
print(f"处理结果：{result['processing_result'][:200]}...")
print(f"完成状态：{result['is_complete']}")
​

步骤 3：带工具的状态图

整合工具到状态图中，实现智能的任务处理：

from langchain.tools import tool

# 定义工具
@tool
def calculator(expression: str) -> str:
    """计算器工具"""
    try:
        import re
        # 安全计算
        numbers = re.findall(r'\d+', expression)
        operators = re.findall(r'[+\-*/]', expression)
        
        if len(numbers) >= 2 and len(operators) >= 1:
            num1, num2 = map(int, numbers[:2])
            op = operators[0]
            
            if op == '+':
                result = num1 + num2
            elif op == '-':
                result = num1 - num2
            elif op == '*':
                result = num1 * num2
            elif op == '/':
                if num2 != 0:
                    result = num1 / num2
                else:
                    return "错误：除数不能为零"
            else:
                return "不支持的操作符"
            
            return f"计算结果: {result}"
        else:
            return "请提供有效的计算表达式"
    except Exception as e:
        return f"计算错误: {str(e)}"

# 定义带工具的状态类型
class ToolAgentState(TypedDict):
    """带工具的 Agent 状态"""
    messages: List[HumanMessage | AIMessage]
    current_step: str
    is_complete: bool
    tool_results: Dict[str, Any]

# 工具使用节点
def tool_usage_node(state: ToolAgentState) -> ToolAgentState:
    """工具使用节点"""
    print("=== 使用工具 ===")
    
    llm = ChatOpenAI(
        model="gpt-4-turbo",
        temperature=0.3,
        max_tokens=1500
    )
    
    # 检查是否需要使用工具
    last_message = state["messages"][-1].content
    
    if "计算" in last_message:
        tool_result = calculator(last_message)
    else:
        tool_result = "无需使用工具"
    
    return {
        **state,
        "tool_results": {**state.get("tool_results", {}), "last_tool": tool_result},
        "current_step": "tool_used"
    }

def processing_with_tools(state: ToolAgentState) -> ToolAgentState:
    """使用工具处理节点"""
    print("=== 使用工具处理 ===")
    
    llm = ChatOpenAI(
        model="gpt-4-turbo",
        temperature=0.7,
        max_tokens=1000
    )
    
    # 构建提示词，包含工具结果
    system_prompt = f"""
你是一个智能助手，可以使用工具帮助用户解决问题。

最近的工具使用结果：{state['tool_results'].get('last_tool', '无')}

请根据用户需求，选择合适的工具并给出最终回答。
"""
    
    response = llm.invoke([
        SystemMessage(content=system_prompt),
        state["messages"][-1]
    ])
    
    return {
        **state,
        "messages": state["messages"] + [response],
        "current_step": "processed_with_tools"
    }

# 创建带工具的图
def create_tool_graph():
    """创建带工具的状态图"""
    
    # 定义初始状态
    initial_state: ToolAgentState = {
        "messages": [HumanMessage(content="帮我计算 25 + 38 等于多少")],
        "current_step": "initial",
        "is_complete": False,
        "tool_results": {}
    }
    
    # 创建图
    workflow = StateGraph(ToolAgentState)
    
    # 添加节点
    workflow.add_node("start", lambda s: {**s, "current_step": "started"})
    workflow.add_node("tool_usage", tool_usage_node)
    workflow.add_node("processing", processing_with_tools)
    workflow.add_node("end", lambda s: {**s, "is_complete": True, "current_step": "completed"})
    
    # 添加边
    workflow.add_edge(START, "start")
    workflow.add_edge("start", "tool_usage")
    workflow.add_edge("tool_usage", "processing")
    workflow.add_edge("processing", "end")
    workflow.add_edge("end", END)
    
    # 编译图
    app = workflow.compile()
    
    return app, initial_state

# 使用带工具的图
print("=== 带工具的 StateGraph 测试 ===")
app, initial_state = create_tool_graph()

# 运行图
result = app.invoke(initial_state)

print("\n最终结果：")
print(f"工具结果：{result['tool_results']}")
print(f"最后回复：{result['messages'][-1].content}")
print(f"完成状态：{result['is_complete']}")
​

完整示例

完整的多功能 Agent 状态图系统

import os
from typing import Dict, List, Any, Optional, TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain.tools import Tool

# 定义工具
@tool
def research_tool(query: str) -> str:
    """研究工具"""
    return f"关于'{query}'的研究结果..."

# 定义复杂状态类型
class ComplexAgentState(TypedDict):
    """复杂 Agent 状态"""
    messages: List[HumanMessage | AIMessage]
    current_step: str
    is_complete: bool
    task_progress: Dict[str, Any]
    tool_results: Dict[str, Any]
    final_result: Optional[str]

# 任务识别节点
def task_identification_node(state: ComplexAgentState) -> ComplexAgentState:
    """任务识别节点"""
    print("=== 任务识别 ===")
    
    llm = ChatOpenAI(
        model="gpt-4-turbo",
        temperature=0.1,
        max_tokens=500
    )
    
    prompt = f"""
请识别用户任务的类型：

用户消息：{state['messages'][-1].content}

任务类型选项：
- research: 研究类任务
- analysis: 分析类任务  
- writing: 写作类任务
- planning: 规划类任务

请只返回任务类型名称。
"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    task_type = response.content.strip().lower()
    
    return {
        **state,
        "task_progress": {"identified_type": task_type, "current_phase": "identification"},
        "current_step": "task_identified"
    }

# 工具选择节点
def tool_selection_node(state: ComplexAgentState) -> ComplexAgentState:
    """工具选择节点"""
    print("=== 工具选择 ===")
    
    task_type = state["task_progress"].get("identified_type", "research")
    
    tools = {
        "research": ["research_tool"],
        "analysis": ["research_tool"],
        "writing": ["research_tool"],
        "planning": ["research_tool"]
    }
    
    selected_tools = tools.get(task_type, ["research_tool"])
    
    return {
        **state,
        "task_progress": {**state["task_progress"], "selected_tools": selected_tools},
        "current_step": "tools_selected"
    }

# 执行节点
def execution_node(state: ComplexAgentState) -> ComplexAgentState:
    """执行节点"""
    print("=== 执行任务 ===")
    
    llm = ChatOpenAI(
        model="gpt-4-turbo",
        temperature=0.3,
        max_tokens=2000
    )
    
    task_type = state["task_progress"]["identified_type"]
    selected_tools = state["task_progress"]["selected_tools"]
    
    # 构建系统提示词
    system_prompt = f"""
你是一个专业助手，正在执行{task_type}任务。

可用工具：{', '.join(selected_tools)}
用户需求：{state['messages'][-1].content}

请根据任务类型选择合适的工具，并完成任务。
"""
    
    response = llm.invoke([
        SystemMessage(content=system_prompt),
        state["messages"][-1]
    ])
    
    return {
        **state,
        "tool_results": {**state.get("tool_results", {}), "execution_result": response.content},
        "final_result": response.content,
        "current_step": "execution_completed"
    }

# 创建复杂图
def create_complex_graph():
    """创建复杂状态图"""
    
    # 定义初始状态
    initial_state: ComplexAgentState = {
        "messages": [HumanMessage(content="我需要分析一个市场营销策略")],
        "current_step": "initial",
        "is_complete": False,
        "task_progress": {},
        "tool_results": {},
        "final_result": None
    }
    
    # 创建图
    workflow = StateGraph(ComplexAgentState)
    
    # 添加节点
    workflow.add_node("start", lambda s: {**s, "current_step": "started"})
    workflow.add_node("task_id", task_identification_node)
    workflow.add_node("tool_selection", tool_selection_node)
    workflow.add_node("execution", execution_node)
    workflow.add_node("end", lambda s: {**s, "is_complete": True, "current_step": "completed"})
    
    # 添加边
    workflow.add_edge(START, "start")
    workflow.add_edge("start", "task_id")
    workflow.add_edge("task_id", "tool_selection")
    workflow.add_edge("tool_selection", "execution")
    workflow.add_edge("execution", "end")
    workflow.add_edge("end", END)
    
    # 编译图
    app = workflow.compile()
    
    return app, initial_state

# 使用复杂图
print("=== 复杂 StateGraph 系统 ===")
app, initial_state = create_complex_graph()

# 运行图
result = app.invoke(initial_state)

print("\n最终结果：")
print(f"任务进度：{result['task_progress']}")
print(f"工具结果：{list(result['tool_results'].keys())}")
print(f"最终结果：{result['final_result'][:300]}...")
print(f"完成状态：{result['is_complete']}")
​

常见问题 FAQ

Q1：StateGraph 和传统 Chain 的主要区别是什么？

A：StateGraph 支持复杂的控制流（循环、分支、条件判断），有状态管理能力，支持图形化工作流，而 Chain 通常是线性的、无状态的，适合简单的任务处理。

Q2：如何调试 StateGraph 的执行过程？

A：可以添加日志节点记录状态变化，使用可视化工具查看执行流程，在关键节点添加状态检查，使用 print 或日志库输出中间结果。

Q3：如何处理 StateGraph 中的错误？

A：添加错误处理节点，使用 try-catch 包裹节点函数，设置错误恢复机制，添加重试逻辑，确保状态的一致性。

Q4：如何优化 StateGraph 的性能？

A：减少不必要的节点调用，优化节点函数的执行效率，使用缓存机制，批量处理状态更新，合理设置最大迭代次数。

最佳实践与避坑

最佳实践

1. 状态设计：保持状态结构简洁，只包含必要的数据
2. 节点职责：每个节点只负责一个特定任务，保持单一职责原则
3. 错误处理：完善的错误处理和恢复机制
4. 性能优化：避免频繁的状态更新和重复计算
5. 测试覆盖：为不同的执行路径编写测试用例

常见陷阱

1. 无限循环：循环条件设置不当导致无限循环
2. 状态混乱：状态结构复杂，难以维护和调试
3. 性能瓶颈：节点函数效率低下，导致整体性能问题
4. 错误处理不当：错误处理逻辑不完善，导致系统不稳定
5. 过度设计：工作流过于复杂，难以理解和维护

本节小结

本节详细介绍了 LangGraph StateGraph 的核心概念和实现方法，包括：

• StateGraph 基础：理解状态图的架构原理和核心组件
• 节点和边：掌握节点定义、边连接和条件边的使用方法
• 状态管理：学会定义和管理工作流状态
• 工具集成：实现工具与状态图的结合使用
• 完整实践：从基础图到复杂多功能 Agent 系统的完整实现

掌握 StateGraph 是构建复杂 Agent 工作流的关键基础。在下一节中，我们将继续学习 LangGraph 的条件分支与循环控制，进一步提升 Agent 的编排能力。

延伸阅读

• 官方文档：LangGraph StateGraph 官方文档
• 相关章节：本教程 3.2 节条件分支与循环控制
• 推荐资源：《工作流编排与状态管理》
• 开源项目：LangGraph 示例库
• 技术博客：Agent 工作流设计最佳实践

关键词：StateGraph,LangGraph,状态图,工作流编排,节点,边,状态管理,Agent架构,教程,实战
难度：进阶
预计阅读：50分钟