从零构建一个具备状态记忆、工具调用、多步推理能力的智能Agent系统

前言：为什么我们需要AI Agent？

在ChatGPT引爆AI热潮后，我们很快发现：单纯的大语言模型（LLM）只是"会说话的百科全书"。它能生成流畅的文本，却无法执行实际操作——不能查询实时数据、不能操作文件、不能与外部系统交互。

这正是AI Agent诞生的背景。Agent = LLM + 记忆 + 工具 + 规划能力，它让AI从"被动回答"进化为"主动解决问题"。

本文将带你深入LangGraph——LangChain团队推出的Agent开发框架，通过一个完整的自动化数据分析Agent项目，掌握现代AI Agent的核心开发技术。

本文你将学到：

• Agent的核心架构与设计模式
• LangGraph状态图编程模型
• 工具调用与Function Calling机制
• 多Agent协作系统设计
• 完整项目实战：从需求到部署

---

一、AI Agent核心概念解析

1.1 Agent ≠ ChatGPT

先通过一个例子理解Agent与传统LLM对话的本质区别：

传统LLM对话：

用户：帮我分析一下sales.csv文件，给出销售趋势
LLM：我无法访问您的本地文件，请提供文件内容或描述...

AI Agent对话：

用户：帮我分析一下sales.csv文件，给出销售趋势
Agent：[读取文件] → [解析数据结构] → [生成分析代码] → [执行分析] → [生成报告]
       我已完成分析，发现以下趋势：Q3销售额同比增长23%...

核心差异：Agent具备行动能力（Action），能调用工具完成实际任务。

1.2 Agent的四大核心组件

现代AI Agent架构遵循ReAct模式（Reasoning + Acting），包含四个关键模块：

┌─────────────────────────────────────────────┐
│              AI Agent Architecture           │
├─────────────────────────────────────────────┤
│  1. 🧠 大脑 (Brain/LLM)                      │
│     - 理解用户意图                            │
│     - 生成推理链条                            │
│     - 决策下一步行动                          │
├─────────────────────────────────────────────┤
│  2. 💾 记忆系统 (Memory)                     │
│     - 短期记忆：当前对话上下文                 │
│     - 长期记忆：历史经验、知识库               │
│     - 工作记忆：任务执行状态                   │
├─────────────────────────────────────────────┤
│  3. 🔧 工具库 (Tools)                        │
│     - 搜索引擎、数据库查询                     │
│     - 代码执行、文件操作                       │
│     - API调用、第三方服务集成                  │
├─────────────────────────────────────────────┤
│  4. 📋 规划器 (Planner)                      │
│     - 任务分解（复杂→简单子任务）              │
│     - 执行策略（顺序/并行/条件分支）           │
│     - 自我反思与纠错                          │
└─────────────────────────────────────────────┘

1.3 ReAct循环：思考-行动-观察

Agent的核心执行模式是ReAct循环：

# ReAct循环伪代码
while not task_complete:
    # 1. 思考（Reasoning）
    thought = llm.think(current_state, goal)
    
    # 2. 行动（Action）
    action = llm.decide_action(available_tools)
    result = execute_tool(action)
    
    # 3. 观察（Observation）
    observation = observe(result)
    
    # 4. 更新状态
    current_state.update(observation)
    
    # 5. 判断是否完成
    task_complete = llm.evaluate(current_state, goal)

LangGraph的价值：提供一套声明式的图结构API，让开发者可以用状态机思维设计Agent，而非编写复杂的循环控制逻辑。

---

二、LangGraph架构深度解析

2.1 为什么选择LangGraph？

在LangGraph之前，主流Agent框架存在明显痛点：

框架	优势	痛点
LangChain Agent	快速原型	难以定制循环逻辑，调试困难
AutoGPT/BabyAGI	自主性强	黑盒执行，不可控，成本高
自定义实现	完全控制	开发成本高，易出错

LangGraph的设计哲学：

• 显式状态管理：状态流转清晰可见
• 可控性：每个节点、边都可精确控制
• 循环支持：原生支持循环、分支结构
• 可视化调试：自动生成执行流程图

2.2 核心概念：图（Graph）、节点（Node）、边（Edge）

LangGraph基于**状态图（State Graph）**模型，三个核心概念：

from langgraph.graph import StateGraph, END

# 1. 定义状态（State）
class AgentState(TypedDict):
    messages: List[BaseMessage]  # 对话历史
    current_task: str             # 当前任务
    tool_results: List[dict]      # 工具执行结果
    next_action: str              # 下一步行动

# 2. 定义节点（Node）- 处理状态的函数
def think_node(state: AgentState) -> AgentState:
    """LLM推理节点"""
    # ... 处理逻辑
    return state

def act_node(state: AgentState) -> AgentState:
    """工具执行节点"""
    # ... 处理逻辑
    return state

# 3. 构建图（Graph）
graph = StateGraph(AgentState)
graph.add_node("think", think_node)
graph.add_node("act", act_node)

# 4. 定义边（Edge）- 状态流转路径
graph.add_edge("think", "act")
graph.add_conditional_edges(
    "act",
    should_continue,  # 条件函数
    {
        "continue": "think",  # 继续循环
        "end": END            # 结束
    }
)

# 5. 编译并运行
app = graph.compile()
result = app.invoke({"messages": [], "current_task": "分析数据"})

2.3 图的结构类型

LangGraph支持多种图结构，适配不同复杂度的Agent：

2.3.1 线性图（Linear Graph）

适用于简单流水线任务：

[Start] → [Node A] → [Node B] → [Node C] → [End]

graph = StateGraph(State)
graph.add_node("a", node_a)
graph.add_node("b", node_b)
graph.add_node("c", node_c)
graph.add_edge("a", "b")
graph.add_edge("b", "c")
graph.add_edge("c", END)
graph.set_entry_point("a")

2.3.2 循环图（Cyclic Graph）

适用于ReAct模式的Agent：

[Start] → [Think] ⟷ [Act] → [End]
              ↑________|

graph.add_edge("think", "act")
graph.add_conditional_edges(
    "act",
    lambda state: "continue" if not state["done"] else "end",
    {"continue": "think", "end": END}
)

2.3.3 分支图（Branching Graph）

适用于多工具选择的复杂场景：

                → [Search Tool] ↘
[Start] → [Router]                    → [Aggregate] → [End]
                → [Code Tool]   ↗

2.4 状态管理：Agent的"工作记忆"

状态是LangGraph的核心，所有节点共享同一状态对象：

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class AgentState(TypedDict):
    # Annotated类型：自动合并（而非覆盖）
    messages: Annotated[List[BaseMessage], add_messages]
    
    # 普通类型：每次更新会覆盖
    current_step: int
    task_list: List[str]
    results: dict

状态更新策略：

• 覆盖模式：state["current_step"] = 2
• 合并模式：add_messages自动将新消息追加到列表

---

三、实战基础：构建第一个Agent

3.1 环境准备

# 创建虚拟环境
python -m venv agent_env
source agent_env/bin/activate  # Windows: agent_env\Scripts\activate

# 安装依赖
pip install langgraph langchain langchain-openai python-dotenv

# 可选：安装工具相关库
pip install duckduckgo-search  # 搜索工具
pip install pandas matplotlib   # 数据分析工具

创建.env文件配置API密钥：

OPENAI_API_KEY=sk-xxx
# 或使用国内模型
# DEEPSEEK_API_KEY=sk-xxx

3.2 最简Agent：问答机器人

先实现一个最简单的图，理解基本流程：

"""
最简LangGraph Agent示例：基础问答机器人
演示图的基本构建和执行流程
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage

# 1. 定义状态
class ChatState(TypedDict):
    messages: list  # 对话历史

# 2. 定义节点
def chatbot_node(state: ChatState) -> ChatState:
    """对话节点：调用LLM生成回复"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    
    # 获取历史消息
    messages = state["messages"]
    
    # 调用LLM
    response = llm.invoke(messages)
    
    # 更新状态（追加AI消息）
    return {"messages": messages + [response]}

# 3. 构建图
graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot_node)
graph.set_entry_point("chatbot")
graph.add_edge("chatbot", END)

# 4. 编译
app = graph.compile()

# 5. 运行
if __name__ == "__main__":
    # 初始状态
    initial_state = {
        "messages": [HumanMessage(content="什么是LangGraph？用一句话解释")]
    }
    
    # 执行图
    result = app.invoke(initial_state)
    
    # 打印结果
    print("AI回复:", result["messages"][-1].content)

运行结果：

AI回复: LangGraph是一个基于图的AI Agent开发框架，让开发者可以用状态机思维构建具有记忆、工具调用和循环推理能力的智能助手。

3.3 进阶：添加工具调用能力

现在让Agent具备搜索能力，成为真正的"能行动"的助手：

"""
工具调用Agent：具备搜索能力的智能助手
演示LangGraph的工具绑定机制
"""
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
import duckduckgo_search

# 1. 定义搜索工具
@tool
def search_web(query: str) -> str:
    """搜索互联网获取实时信息
    
    Args:
        query: 搜索关键词
        
    Returns:
        搜索结果摘要
    """
    from duckduckgo_search import DDGS
    
    with DDGS() as ddgs:
        results = list(ddgs.text(query, max_results=3))
    
    if not results:
        return "未找到相关结果"
    
    # 格式化结果
    output = []
    for r in results:
        output.append(f"【{r['title']}】\n{r['body']}\n链接: {r['href']}\n")
    
    return "\n".join(output)

# 2. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, "add_messages"]  # 自动合并消息

# 3. 初始化LLM并绑定工具
tools = [search_web]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# 4. 定义节点
def agent_node(state: AgentState) -> AgentState:
    """Agent推理节点：决定是否调用工具"""
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    """路由函数：判断是否需要调用工具"""
    last_message = state["messages"][-1]
    
    # 如果LLM决定调用工具，则进入工具节点
    if last_message.tool_calls:
        return "tools"
    # 否则结束
    return "end"

# 5. 构建图
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))  # 内置工具节点

# 设置入口
graph.set_entry_point("agent")

# 添加条件边
graph.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "end": END
    }
)

# 工具执行后返回Agent继续推理
graph.add_edge("tools", "agent")

# 6. 编译
app = graph.compile()

# 7. 运行示例
if __name__ == "__main__":
    question = "2024年诺贝尔物理学奖获得者是谁？"
    
    result = app.invoke({
        "messages": [HumanMessage(content=question)]
    })
    
    # 打印执行过程
    print("=" * 50)
    print("用户问题:", question)
    print("=" * 50)
    
    for i, msg in enumerate(result["messages"]):
        msg_type = type(msg).__name__
        print(f"\n[步骤{i+1}] {msg_type}:")
        if hasattr(msg, 'tool_calls') and msg.tool_calls:
            print(f"  → 调用工具: {msg.tool_calls}")
        else:
            print(f"  → {msg.content[:200]}...")
    
    print("\n" + "=" * 50)
    print("最终回答:", result["messages"][-1].content)

执行流程可视化：

[步骤1] HumanMessage:
  → 2024年诺贝尔物理学奖获得者是谁？

[步骤2] AIMessage:
  → 调用工具: [{'name': 'search_web', 'args': {'query': '2024年诺贝尔物理学奖获得者'}}]

[步骤3] ToolMessage:
  → 【2024年诺贝尔物理学奖公布...】
  
[步骤4] AIMessage:
  → 2024年诺贝尔物理学奖授予了John Hopfield和Geoffrey Hinton...

---

四、状态管理深度实践

4.1 为什么状态管理至关重要？

在复杂Agent系统中，状态管理直接影响：

• 上下文连贯性：多轮对话、多步骤任务
• 错误恢复：任务失败后的状态回滚
• 调试追踪：定位问题节点
• 持久化：任务中断后的恢复

4.2 状态设计最佳实践

from typing import TypedDict, Annotated, Literal
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage

class DataAnalysisState(TypedDict):
    """数据分析Agent的状态定义
    
    设计原则：
    1. 区分"累加型"和"覆盖型"字段
    2. 明确每个字段的用途
    3. 为复杂字段定义子结构
    """
    
    # === 对话与推理 ===
    messages: Annotated[List[BaseMessage], add_messages]  # 对话历史（自动合并）
    reasoning_chain: List[str]  # 推理链条（手动追加）
    
    # === 任务管理 ===
    current_step: str  # 当前步骤标识
    task_plan: List[dict]  # 任务计划列表
    completed_steps: List[str]  # 已完成步骤
    
    # === 数据处理 ===
    input_file: str  # 输入文件路径
    data_schema: dict  # 数据结构信息
    analysis_results: dict  # 分析结果
    
    # === 输出控制 ===
    output_format: Literal["report", "chart", "json"]  # 输出格式
    final_output: str  # 最终输出内容
    
    # === 错误处理 ===
    errors: List[dict]  # 错误记录
    retry_count: int  # 重试计数

4.3 状态更新模式

模式1：直接覆盖

def update_step(state: DataAnalysisState) -> DataAnalysisState:
    """更新当前步骤"""
    return {
        "current_step": "data_cleaning"  # 直接覆盖
    }

模式2：追加到列表

def add_reasoning(state: DataAnalysisState) -> DataAnalysisState:
    """追加推理步骤"""
    new_thought = "发现数据包含空值，需要清洗"
    
    return {
        "reasoning_chain": state["reasoning_chain"] + [new_thought]
    }

模式3：合并字典

def update_results(state: DataAnalysisState) -> DataAnalysisState:
    """合并分析结果"""
    new_result = {"mean_price": 150.5, "std_price": 23.8}
    
    return {
        "analysis_results": {
            **state["analysis_results"],  # 保留现有结果
            **new_result  # 添加新结果
        }
    }

4.4 状态持久化：Checkpointing

LangGraph内置检查点机制，支持任务中断与恢复：

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

# 创建检查点存储
checkpointer = MemorySaver()

# 编译时启用检查点
app = graph.compile(checkpointer=checkpointer)

# 运行并指定thread_id（会话标识）
config = {"configurable": {"thread_id": "session-001"}}
result = app.invoke(initial_state, config=config)

# 后续可以从断点恢复
# checkpointer会自动保存每一步的状态

应用场景：

• 长时间运行的任务（数据分析、代码生成）
• 需要人工审核的步骤
• 多用户并发场景

---

五、多Agent协作系统

5.1 为什么需要多Agent？

单个Agent在面对复杂任务时存在局限：

• 能力边界：一个Agent难以精通所有领域
• 注意力分散：长任务链容易偏离目标
• 单点故障：一个工具失败可能导致整体失败

多Agent协作的优势：

• 专业化分工：每个Agent专注特定领域
• 并行执行：独立子任务可并行处理
• 相互校验：多Agent交叉验证提高准确性

5.2 协作模式

模式1：主从模式（Supervisor）

           ┌→ [Researcher Agent] →┐
[User] → [Supervisor Agent]        → [Final Report]
           └→ [Coder Agent]     →┘

"""
多Agent协作：主从模式示例
Supervisor负责任务分配和结果整合
"""
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage

# 定义状态
class MultiAgentState(TypedDict):
    messages: Annotated[list, "add_messages"]
    next_agent: str  # 下一个要执行的Agent
    research_result: str
    code_result: str
    final_report: str

# Supervisor节点
def supervisor_node(state: MultiAgentState) -> MultiAgentState:
    """主控Agent：分析任务并分配"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    
    prompt = f"""你是一个任务分配者。分析用户需求，决定下一步应该由哪个Agent处理：
    
当前状态：
- 已有研究结果: {state.get('research_result', '无')}
- 已有代码结果: {state.get('code_result', '无')}

可选Agent：
- researcher: 负责信息搜集和调研
- coder: 负责代码编写
- end: 任务完成，生成最终报告

输出格式（仅输出一个词）：
researcher / coder / end
"""
    
    response = llm.invoke([
        HumanMessage(content=prompt)
    ])
    
    decision = response.content.strip().lower()
    
    return {"next_agent": decision}

# Researcher节点
def researcher_node(state: MultiAgentState) -> MultiAgentState:
    """研究Agent：搜集信息"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    
    task = state["messages"][0].content
    
    prompt = f"""你是一个专业研究员。针对以下任务进行调研：

任务：{task}

请提供：
1. 关键概念解释
2. 相关技术或方法
3. 实施建议
"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    return {
        "research_result": response.content,
        "next_agent": "supervisor"  # 返回主控
    }

# Coder节点
def coder_node(state: MultiAgentState) -> MultiAgentState:
    """代码Agent：编写代码"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
    
    task = state["messages"][0].content
    research = state.get("research_result", "")
    
    prompt = f"""你是一个专业程序员。基于研究结果编写代码：

原始任务：{task}

研究结果：
{research}

请提供：
1. 完整可运行的Python代码
2. 代码注释
3. 使用说明
"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    return {
        "code_result": response.content,
        "next_agent": "supervisor"  # 返回主控
    }

# 最终报告节点
def final_report_node(state: MultiAgentState) -> MultiAgentState:
    """生成最终报告"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
    
    task = state["messages"][0].content
    
    prompt = f"""生成最终报告：

原始任务：{task}

研究结果：
{state.get('research_result', '')}

代码方案：
{state.get('code_result', '')}

请整合以上信息，生成一份完整的技术报告。
"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    return {
        "final_report": response.content,
        "next_agent": "end"
    }

# 路由函数
def route_agent(state: MultiAgentState) -> str:
    """根据next_agent字段路由"""
    next_agent = state["next_agent"]
    
    if next_agent == "end":
        return "final_report"
    return next_agent

# 构建图
graph = StateGraph(MultiAgentState)

# 添加节点
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("coder", coder_node)
graph.add_node("final_report", final_report_node)

# 设置入口
graph.set_entry_point("supervisor")

# 添加条件路由
graph.add_conditional_edges(
    "supervisor",
    route_agent,
    {
        "researcher": "researcher",
        "coder": "coder",
        "final_report": "final_report"
    }
)

# Agent执行后返回Supervisor
graph.add_edge("researcher", "supervisor")
graph.add_edge("coder", "supervisor")
graph.add_edge("final_report", END)

# 编译
app = graph.compile()

模式2：对等协作（Peer-to-Peer）

"""
对等协作模式：两个专家Agent互相讨论
适用于需要多角度分析的场景
"""
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END

class DebateState(TypedDict):
    messages: Annotated[list, "add_messages"]
    round_count: int
    conclusion: str

def optimist_agent(state: DebateState) -> DebateState:
    """乐观派Agent"""
    # ... 提供正面观点
    pass

def pessimist_agent(state: DebateState) -> DebateState:
    """悲观派Agent"""
    # ... 提供反面观点
    pass

def mediator_agent(state: DebateState) -> DebateState:
    """调解人：综合观点"""
    # ... 生成结论
    pass

# 构建对等协作图
# 两个Agent轮流发言，最后由调解人总结

5.3 Agent通信机制

多Agent之间需要共享信息，常用方法：

方法1：共享状态（推荐）

class SharedState(TypedDict):
    workspace: dict  # 共享工作空间
    agent_a_output: str
    agent_b_output: str

方法2：消息传递

def agent_a_send_message(state: SharedState) -> SharedState:
    """Agent A发送消息到工作空间"""
    message = {"from": "agent_a", "content": "数据分析完成"}
    
    return {
        "workspace": {
            **state["workspace"],
            "messages": state["workspace"].get("messages", []) + [message]
        }
    }

---

六、完整项目：自动化数据分析Agent

6.1 项目需求分析

**目标：**构建一个能自动完成数据分析全流程的Agent系统

功能需求：

1. 接收用户自然语言描述的分析需求
2. 自动读取数据文件（CSV/Excel）
3. 自动理解数据结构
4. 生成并执行分析代码
5. 输出可视化图表和分析报告

技术架构：

用户输入 → [需求理解] → [数据加载] → [数据分析] → [可视化] → [报告生成]
              ↓             ↓             ↓            ↓
           [状态管理] ← ← ← ← ← ← ← ← ← ← ← ← ← ←

6.2 项目结构

data_analysis_agent/
├── agent/
│   ├── __init__.py
│   ├── graph.py          # 主图定义
│   ├── nodes/            # 各节点实现
│   │   ├── planner.py    # 任务规划节点
│   │   ├── loader.py     # 数据加载节点
│   │   ├── analyzer.py   # 分析执行节点
│   │   └── reporter.py   # 报告生成节点
│   ├── state.py          # 状态定义
│   └── tools.py          # 工具函数
├── utils/
│   ├── code_executor.py  # 代码执行器
│   └── visualizer.py     # 可视化工具
├── main.py               # 入口文件
└── requirements.txt

6.3 核心代码实现

6.3.1 状态定义

# agent/state.py
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage

class DataAnalysisState(TypedDict):
    """数据分析Agent状态
    
    状态流转：
    用户需求 → 规划 → 加载 → 分析 → 可视化 → 报告
    """
    
    # === 输入 ===
    user_request: str  # 用户原始需求
    input_file: Optional[str]  # 数据文件路径
    
    # === 处理过程 ===
    messages: Annotated[List[BaseMessage], add_messages]  # 对话历史
    
    # 规划阶段
    analysis_plan: List[str]  # 分析计划步骤
    current_step_index: int  # 当前执行到第几步
    
    # 数据阶段
    data_info: dict  # 数据基本信息（列名、类型、统计）
    data_sample: str  # 数据预览（前5行）
    
    # 分析阶段
    generated_code: str  # 生成的分析代码
    code_output: str  # 代码执行输出
    analysis_results: dict  # 解析后的分析结果
    
    # 可视化阶段
    chart_paths: List[str]  # 生成的图表文件路径
    
    # === 输出 ===
    final_report: str  # 最终分析报告
    
    # === 控制 ===
    errors: List[dict]  # 错误记录
    retry_count: int  # 当前步骤重试次数
    max_retries: int  # 最大重试次数

6.3.2 工具函数

# agent/tools.py
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from pathlib import Path
from typing import Optional
import json

class DataTools:
    """数据分析工具集"""
    
    @staticmethod
    def load_data(file_path: str) -> dict:
        """加载数据文件并返回基本信息
        
        Args:
            file_path: CSV或Excel文件路径
            
        Returns:
            包含数据信息的字典
        """
        path = Path(file_path)
        
        if not path.exists():
            return {"error": f"文件不存在: {file_path}"}
        
        # 根据扩展名选择加载方式
        if path.suffix == '.csv':
            df = pd.read_csv(file_path)
        elif path.suffix in ['.xlsx', '.xls']:
            df = pd.read_excel(file_path)
        else:
            return {"error": f"不支持的文件格式: {path.suffix}"}
        
        # 生成数据信息
        info = {
            "file_name": path.name,
            "rows": len(df),
            "columns": list(df.columns),
            "dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},
            "missing_values": {col: int(df[col].isna().sum()) for col in df.columns},
            "statistics": df.describe().to_dict()
        }
        
        # 数据预览
        sample = df.head(5).to_string()
        
        return {
            "info": info,
            "sample": sample,
            "shape": df.shape,
            "dataframe": df  # 保留DataFrame供后续使用
        }
    
    @staticmethod
    def execute_analysis_code(code: str, df: pd.DataFrame) -> dict:
        """执行分析代码
        
        Args:
            code: Python分析代码
            df: 数据DataFrame
            
        Returns:
            执行结果
        """
        # 创建执行环境
        local_vars = {
            'df': df,
            'pd': pd,
            'plt': plt,
            'sns': sns,
            'result': {}
        }
        
        try:
            # 执行代码
            exec(code, local_vars)
            
            # 获取结果
            result = local_vars.get('result', {})
            
            return {
                "success": True,
                "output": str(result),
                "result": result
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
    
    @staticmethod
    def save_chart(figure: plt.Figure, output_dir: str, name: str) -> str:
        """保存图表
        
        Args:
            figure: matplotlib Figure对象
            output_dir: 输出目录
            name: 图表名称
            
        Returns:
            保存的文件路径
        """
        output_path = Path(output_dir)
        output_path.mkdir(parents=True, exist_ok=True)
        
        file_path = output_path / f"{name}.png"
        figure.savefig(file_path, dpi=150, bbox_inches='tight')
        plt.close(figure)
        
        return str(file_path)


# 定义LangGraph工具
from langchain_core.tools import tool

@tool
def analyze_data_info(file_path: str) -> str:
    """分析数据文件的基本信息
    
    Args:
        file_path: 数据文件路径
        
    Returns:
        数据信息JSON字符串
    """
    result = DataTools.load_data(file_path)
    return json.dumps(result["info"], indent=2, ensure_ascii=False)

@tool  
def generate_sample_code(data_info: str, analysis_goal: str) -> str:
    """根据数据信息和分析目标生成分析代码
    
    Args:
        data_info: 数据基本信息JSON
        analysis_goal: 分析目标描述
        
    Returns:
        Python分析代码
    """
    # 这个工具实际由LLM实现，这里只是占位
    return "# 代码生成由LLM完成"

6.3.3 节点实现

# agent/nodes/planner.py
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from typing import List
import json

def planner_node(state: dict) -> dict:
    """任务规划节点：分析用户需求，制定分析计划"""
    
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    
    system_prompt = """你是一个数据分析专家。根据用户需求，制定详细的分析计划。

输出JSON格式的计划列表，例如：
{
    "plan": [
        "1. 数据概览：查看数据基本统计信息",
        "2. 数据质量：检查缺失值和异常值",
        "3. 分布分析：分析关键数值列的分布",
        "4. 相关性分析：探索变量间的关系",
        "5. 可视化：生成关键图表",
        "6. 总结：输出分析报告"
    ]
}

只输出JSON，不要有其他内容。"""

    user_prompt = f"""用户需求：{state['user_request']}

请制定分析计划。"""

    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=user_prompt)
    ])
    
    # 解析计划
    try:
        plan_data = json.loads(response.content)
        plan = plan_data["plan"]
    except:
        # 如果解析失败，使用默认计划
        plan = [
            "1. 数据概览",
            "2. 数据质量检查",
            "3. 分析执行",
            "4. 生成报告"
        ]
    
    return {
        "analysis_plan": plan,
        "current_step_index": 0
    }


# agent/nodes/loader.py
def loader_node(state: dict) -> dict:
    """数据加载节点：读取数据文件并提取信息"""
    
    from agent.tools import DataTools
    
    file_path = state.get("input_file")
    
    if not file_path:
        return {"errors": state.get("errors", []) + [{"step": "loader", "error": "未指定数据文件"}]}
    
    result = DataTools.load_data(file_path)
    
    if "error" in result:
        return {
            "errors": state.get("errors", []) + [{"step": "loader", "error": result["error"]}]
        }
    
    return {
        "data_info": result["info"],
        "data_sample": result["sample"],
        "_dataframe": result.get("dataframe")  # 临时存储
    }


# agent/nodes/analyzer.py
def analyzer_node(state: dict) -> dict:
    """分析执行节点：生成并执行分析代码"""
    
    from agent.tools import DataTools
    
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
    
    # 构建提示词
    system_prompt = """你是一个数据分析程序员。根据数据信息和分析目标，编写Python代码进行分析。

要求：
1. 使用pandas进行数据处理
2. 使用matplotlib/seaborn生成图表
3. 将分析结果存入result字典
4. 代码要完整可运行
5. 添加必要的注释"""

    user_prompt = f"""数据信息：
{json.dumps(state['data_info'], indent=2, ensure_ascii=False)}

数据预览：
{state['data_sample']}

当前分析步骤：
{state['analysis_plan'][state['current_step_index']]}

请编写分析代码。"""

    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=user_prompt)
    ])
    
    generated_code = response.content
    
    # 提取代码块
    if "```python" in generated_code:
        code_start = generated_code.find("```python") + 9
        code_end = generated_code.find("```", code_start)
        code = generated_code[code_start:code_end].strip()
    elif "```" in generated_code:
        code_start = generated_code.find("```") + 3
        code_end = generated_code.find("```", code_start)
        code = generated_code[code_start:code_end].strip()
    else:
        code = generated_code
    
    # 执行代码
    df = state.get("_dataframe")
    execution_result = DataTools.execute_analysis_code(code, df)
    
    return {
        "generated_code": code,
        "code_output": execution_result.get("output", ""),
        "analysis_results": execution_result.get("result", {}),
        "current_step_index": state["current_step_index"] + 1
    }


# agent/nodes/reporter.py
def reporter_node(state: dict) -> dict:
    """报告生成节点：整合所有分析结果，生成最终报告"""
    
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
    
    system_prompt = """你是一个数据分析师，负责撰写分析报告。

报告要求：
1. 结构清晰：概述 → 数据概况 → 主要发现 → 建议
2. 突出关键发现
3. 提供可操作的建议
4. 使用Markdown格式"""

    user_prompt = f"""原始需求：{state['user_request']}

数据信息：
{json.dumps(state['data_info'], indent=2, ensure_ascii=False)}

分析计划：
{json.dumps(state['analysis_plan'], indent=2, ensure_ascii=False)}

代码执行输出：
{state.get('code_output', '无')}

分析结果：
{json.dumps(state.get('analysis_results', {}), indent=2, ensure_ascii=False)}

请撰写完整的分析报告。"""

    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=user_prompt)
    ])
    
    return {
        "final_report": response.content
    }

6.3.4 主图构建

# agent/graph.py
from langgraph.graph import StateGraph, END
from agent.state import DataAnalysisState
from agent.nodes.planner import planner_node
from agent.nodes.loader import loader_node
from agent.nodes.analyzer import analyzer_node
from agent.nodes.reporter import reporter_node

def should_continue_analysis(state: DataAnalysisState) -> str:
    """判断是否继续分析"""
    
    # 检查错误
    if state.get("errors"):
        return "report"  # 有错误则直接生成报告
    
    # 检查步骤进度
    current = state["current_step_index"]
    total = len(state["analysis_plan"])
    
    if current < total:
        return "analyze"  # 继续分析
    else:
        return "report"  # 进入报告生成

def build_analysis_graph():
    """构建数据分析Agent图"""
    
    # 创建图
    graph = StateGraph(DataAnalysisState)
    
    # 添加节点
    graph.add_node("planner", planner_node)
    graph.add_node("loader", loader_node)
    graph.add_node("analyzer", analyzer_node)
    graph.add_node("reporter", reporter_node)
    
    # 设置入口
    graph.set_entry_point("planner")
    
    # 定义边
    graph.add_edge("planner", "loader")
    graph.add_edge("loader", "analyzer")
    
    # 条件边：判断是否继续分析
    graph.add_conditional_edges(
        "analyzer",
        should_continue_analysis,
        {
            "analyze": "analyzer",  # 继续下一轮分析
            "report": "reporter"
        }
    )
    
    graph.add_edge("reporter", END)
    
    # 编译
    return graph.compile()


# main.py
from agent.graph import build_analysis_graph
from agent.state import DataAnalysisState

def main():
    """主入口"""
    
    # 构建Agent
    app = build_analysis_graph()
    
    # 用户输入
    user_request = input("请描述你的数据分析需求: ")
    file_path = input("请输入数据文件路径: ")
    
    # 初始状态
    initial_state = {
        "user_request": user_request,
        "input_file": file_path,
        "messages": [],
        "analysis_plan": [],
        "current_step_index": 0,
        "data_info": {},
        "data_sample": "",
        "generated_code": "",
        "code_output": "",
        "analysis_results": {},
        "chart_paths": [],
        "final_report": "",
        "errors": [],
        "retry_count": 0,
        "max_retries": 3
    }
    
    # 执行
    print("\n" + "="*50)
    print("开始执行数据分析...")
    print("="*50 + "\n")
    
    result = app.invoke(initial_state)
    
    # 输出结果
    print("\n" + "="*50)
    print("分析报告")
    print("="*50 + "\n")
    print(result["final_report"])
    
    # 保存报告
    output_file = "analysis_report.md"
    with open(output_file, "w", encoding="utf-8") as f:
        f.write(result["final_report"])
    
    print(f"\n报告已保存到: {output_file}")


if __name__ == "__main__":
    main()

6.4 运行示例

$ python main.py

请描述你的数据分析需求: 分析销售数据，找出销售趋势和最佳产品
请输入数据文件路径: ./sales_data.csv

==================================================
开始执行数据分析...
==================================================

[规划阶段] 制定分析计划...
[加载阶段] 读取数据文件...
[分析阶段] 执行步骤 1/5...
[分析阶段] 执行步骤 2/5...
[分析阶段] 执行步骤 3/5...

==================================================
分析报告
==================================================

# 销售数据分析报告

## 一、数据概况

本次分析的数据集包含1000条销售记录，时间跨度为2023年1月至2024年6月。
数据包含以下字段：日期、产品名称、销售额、数量、客户类型...

## 二、主要发现

### 1. 销售趋势
- 整体销售额呈上升趋势，月均增长率约5.2%
- Q4为销售旺季，销售额占全年的35%

### 2. 最佳产品
- 销售额Top3产品：产品A（￥125万）、产品B（￥98万）、产品C（￥76万）
- 产品A贡献了总销售额的25%

## 三、建议

1. 加大旺季营销投入
2. 优化产品A的供应链
...

---

七、高级主题与优化策略

7.1 错误处理与重试机制

from typing import Literal

def robust_analyzer_node(state: DataAnalysisState) -> DataAnalysisState:
    """具备错误处理的分析节点"""
    
    max_retries = state.get("max_retries", 3)
    current_retry = state.get("retry_count", 0)
    
    try:
        # 执行分析
        result = execute_analysis(state)
        
        if result["success"]:
            return {
                **result,
                "retry_count": 0  # 成功后重置计数器
            }
        else:
            raise Exception(result.get("error", "Unknown error"))
            
    except Exception as e:
        # 记录错误
        errors = state.get("errors", [])
        errors.append({
            "step": "analyzer",
            "error": str(e),
            "retry": current_retry
        })
        
        # 判断是否需要重试
        if current_retry < max_retries:
            # 生成修复后的代码
            fixed_code = fix_code_with_llm(state["generated_code"], str(e))
            
            return {
                "errors": errors,
                "retry_count": current_retry + 1,
                "generated_code": fixed_code
            }
        else:
            # 超过重试次数，跳过当前步骤
            return {
                "errors": errors,
                "current_step_index": state["current_step_index"] + 1,
                "retry_count": 0
            }

7.2 流式输出

LangGraph支持流式执行，实时展示中间结果：

from langgraph.graph import StateGraph

app = graph.compile()

# 流式执行
for event in app.stream(initial_state):
    # event是字典：{节点名: 节点输出}
    for node_name, node_output in event.items():
        print(f"[{node_name}] {node_output}")

7.3 可视化调试

from IPython.display import Image, display

# 生成流程图
try:
    display(Image(app.get_graph().draw_mermaid_png()))
except:
    # 保存为文件
    graph_image = app.get_graph().draw_mermaid_png()
    with open("agent_flow.png", "wb") as f:
        f.write(graph_image)

7.4 性能优化

优化1：并行执行

from langgraph.graph import StateGraph

# 定义并行节点
graph.add_node("task_a", node_a)
graph.add_node("task_b", node_b)

# 从同一起点分支
graph.add_edge("start", "task_a")
graph.add_edge("start", "task_b")

# 汇聚到同一节点
graph.add_edge("task_a", "merge")
graph.add_edge("task_b", "merge")

优化2：缓存机制

from functools import lru_cache

@lru_cache(maxsize=100)
def cached_llm_call(prompt_hash: str) -> str:
    """缓存LLM调用结果"""
    return llm.invoke(prompt)

优化3：Token管理

def truncate_messages(messages: list, max_tokens: int = 4000):
    """截断消息历史以控制token数"""
    # 使用tiktoken计算token数
    import tiktoken
    
    encoding = tiktoken.encoding_for_model("gpt-4")
    total_tokens = sum(len(encoding.encode(m.content)) for m in messages)
    
    if total_tokens > max_tokens:
        # 保留最新的消息
        while total_tokens > max_tokens and len(messages) > 1:
            messages.pop(0)
            total_tokens = sum(len(encoding.encode(m.content)) for m in messages)
    
    return messages

---

八、总结与展望

8.1 核心要点回顾

本文从理论到实践，完整介绍了LangGraph Agent开发：

1. Agent本质 = LLM + 记忆 + 工具 + 规划，实现"思考-行动"循环
2. LangGraph优势 = 显式状态管理 + 可控执行流程 + 循环/分支支持
3. 状态设计 = 区分累加型/覆盖型字段，合理组织数据结构
4. 多Agent协作 = 主从模式/对等模式，实现专业化分工
5. 完整项目 = 数据分析Agent展示真实场景应用

8.2 最佳实践总结

维度	建议
状态设计	粒度适中，避免过度复杂；区分临时状态与持久状态
节点实现	单一职责，每个节点只做一件事；幂等性设计
错误处理	多层重试机制；降级策略；错误隔离
工具选择	优先使用成熟库；合理抽象；文档完善
调试策略	使用检查点机制；可视化流程图；单元测试节点

8.3 未来发展方向

AI Agent技术正在快速演进，值得关注的方向：

1. 多模态Agent：图像、语音、视频理解与生成
2. 自主学习：从执行中学习，持续优化策略
3. 人机协作：人在回路（Human-in-the-loop）的混合智能
4. 安全可控：输出对齐、风险评估、可解释性

---

附录

A. 完整项目代码

项目已开源：github.com/yourname/data-analysis-agent

B. 参考资料

1. LangGraph官方文档
2. ReAct论文
3. LangChain文档
4. OpenAI Function Calling

C. 环境依赖

# requirements.txt
langgraph>=0.2.0
langchain>=0.3.0
langchain-openai>=0.2.0
pandas>=2.0.0
matplotlib>=3.7.0
seaborn>=0.12.0
duckduckgo-search>=4.0.0
python-dotenv>=1.0.0

---