Ver Fonte

练习模块提交

wangyang há 1 mês atrás
pai
commit
7d1ef68514
7 ficheiros alterados com 1703 adições e 1 exclusões
  1. 484 0
      PRACTICE_GUIDE.md
  2. 10 1
      README
  3. 20 0
      examples/__init__.py
  4. 351 0
      examples/advanced_agent.py
  5. 140 0
      examples/basic_agent.py
  6. 417 0
      examples/multi_agent.py
  7. 281 0
      examples/state_machine.py

+ 484 - 0
PRACTICE_GUIDE.md

@@ -0,0 +1,484 @@
+# 交易流水分析项目 - LangChain & LangGraph Agent开发练习手册
+
+## 📚 练习目标
+
+通过这个项目,掌握以下核心技能:
+- ✅ LangChain基础使用(模型、提示词、输出解析)
+- ✅ LangGraph状态机工作流设计
+- ✅ Agent架构设计与实现
+- ✅ 多Agent协作系统
+- ✅ API调用与错误处理
+- ✅ 数据处理与文件管理
+- ✅ 日志记录与调试
+
+## 🏗️ 项目架构概览
+
+```
+交易流水分析系统
+├── 规划Agent (PlanningAgent)
+│   ├── 分析用户需求
+│   ├── 决策下一步行动
+│   └── 分配计算任务
+├── 大纲生成Agent (OutlineAgent)
+│   ├── 生成分析报告结构
+│   ├── 选择相关指标
+│   └── 输出JSON格式大纲
+├── 指标计算Agent (MetricCalculationAgent)
+│   ├── 调用外部API
+│   ├── 处理计算结果
+│   └── 错误处理与重试
+└── 工作流编排器 (CompleteAgentFlow)
+    ├── 状态管理
+    ├── 条件路由
+    └── 执行控制
+```
+
+## 🚀 练习路线图
+
+### Phase 1: 环境搭建与基础概念
+
+#### 练习1.1: 环境配置
+**目标**: 掌握Python项目环境搭建
+
+**任务**:
+1. 创建新的虚拟环境
+2. 安装基础依赖包
+3. 配置API密钥环境变量
+4. 验证安装成功
+
+**代码示例**:
+```bash
+# 创建虚拟环境
+python -m venv tx_flow_env
+source tx_flow_env/bin/activate  # Linux/Mac
+# 或 tx_flow_env\Scripts\activate  # Windows
+
+# 安装依赖
+pip install langchain langgraph pydantic python-dotenv
+
+# 设置环境变量
+echo "DEEPSEEK_API_KEY=your-key-here" > .env
+```
+
+**验证代码**:
+```python
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+api_key = os.getenv('DEEPSEEK_API_KEY')
+print(f"API Key loaded: {api_key is not None}")
+```
+
+#### 练习1.2: LangChain基础
+**目标**: 理解LangChain核心组件
+
+**任务**:
+1. 创建ChatOpenAI实例
+2. 设计简单的提示词模板
+3. 发送请求并处理响应
+4. 解析JSON输出
+
+**代码示例**:
+```python
+from langchain_openai import ChatOpenAI
+from langchain_core.prompts import ChatPromptTemplate
+
+# 初始化模型
+llm = ChatOpenAI(
+    model="deepseek-chat",
+    api_key="your-api-key",
+    base_url="https://api.deepseek.com",
+    temperature=0.1
+)
+
+# 创建提示词模板
+prompt = ChatPromptTemplate.from_messages([
+    ("system", "你是一个专业的助手,请用JSON格式回答。"),
+    ("user", "请分析这个数据: {data}")
+])
+
+# 链式调用
+chain = prompt | llm
+result = chain.invoke({"data": "示例数据"})
+print(result.content)
+```
+
+### Phase 2: Agent开发基础
+
+#### 练习2.1: 简单Agent实现
+**目标**: 掌握Agent基本结构
+
+**任务**:
+1. 创建基础Agent类
+2. 实现初始化方法
+3. 添加API调用跟踪
+4. 实现简单的问答功能
+
+**代码示例**:
+```python
+from typing import List, Dict, Any
+from datetime import datetime
+from langchain_openai import ChatOpenAI
+
+class BaseAgent:
+    """基础Agent类"""
+
+    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com"):
+        self.llm = ChatOpenAI(
+            model="deepseek-chat",
+            api_key=api_key,
+            base_url=base_url,
+            temperature=0.1
+        )
+        # API调用跟踪
+        self.api_calls: List[Dict[str, Any]] = []
+
+    def call_llm(self, messages: List[Dict[str, str]]) -> str:
+        """调用大模型并记录"""
+        start_time = datetime.now()
+
+        try:
+            response = self.llm.invoke(messages)
+            end_time = datetime.now()
+
+            # 记录API调用
+            self.api_calls.append({
+                "timestamp": end_time.isoformat(),
+                "duration": (end_time - start_time).total_seconds(),
+                "success": True,
+                "response": response.content
+            })
+
+            return response.content
+
+        except Exception as e:
+            end_time = datetime.now()
+            self.api_calls.append({
+                "timestamp": end_time.isoformat(),
+                "duration": (end_time - start_time).total_seconds(),
+                "success": False,
+                "error": str(e)
+            })
+            raise
+```
+
+#### 练习2.2: 提示词工程
+**目标**: 掌握提示词设计技巧
+
+**任务**:
+1. 设计结构化输出提示词
+2. 使用JSON模式输出
+3. 处理输出解析错误
+4. 实现重试机制
+
+**代码示例**:
+```python
+from pydantic import BaseModel, Field
+from langchain_core.output_parsers import JsonOutputParser
+
+class AnalysisResult(BaseModel):
+    """分析结果结构"""
+    summary: str = Field(description="分析摘要")
+    key_points: List[str] = Field(description="关键要点")
+    confidence: float = Field(description="置信度", ge=0, le=1)
+
+def create_analysis_prompt() -> ChatPromptTemplate:
+    """创建分析提示词"""
+
+    parser = JsonOutputParser(pydantic_object=AnalysisResult)
+
+    template = """你是一个专业的数据分析师,请分析以下数据并按JSON格式输出。
+
+数据: {data}
+
+要求:
+1. 提供简洁的分析摘要
+2. 列出3-5个关键要点
+3. 给出分析置信度(0-1之间)
+
+{format_instructions}
+
+请确保输出有效的JSON格式。"""
+
+    return ChatPromptTemplate.from_template(
+        template,
+        partial_variables={"format_instructions": parser.get_format_instructions()}
+    )
+```
+
+### Phase 3: LangGraph工作流
+
+#### 练习3.1: 基础状态机
+**目标**: 理解状态机概念
+
+**任务**:
+1. 定义状态结构
+2. 创建基础节点
+3. 实现简单路由
+4. 测试状态转换
+
+**代码示例**:
+```python
+from typing import TypedDict, Annotated, List
+from langgraph.graph import StateGraph, START, END
+
+class WorkflowState(TypedDict):
+    """工作流状态"""
+    question: str
+    current_step: str
+    results: Annotated[List[str], "add"]  # 使用注解支持列表追加
+
+def planning_node(state: WorkflowState) -> WorkflowState:
+    """规划节点"""
+    print(f"📋 规划阶段 - 问题: {state['question']}")
+
+    # 简单的决策逻辑
+    if "分析" in state["question"]:
+        next_step = "analyze"
+    else:
+        next_step = "answer"
+
+    return {
+        **state,
+        "current_step": next_step,
+        "results": ["规划完成"]
+    }
+
+def analysis_node(state: WorkflowState) -> WorkflowState:
+    """分析节点"""
+    print("🔍 分析阶段")
+    return {
+        **state,
+        "results": ["分析完成"]
+    }
+
+def answer_node(state: WorkflowState) -> WorkflowState:
+    """回答节点"""
+    print("💬 回答阶段")
+    return {
+        **state,
+        "results": ["回答完成"]
+    }
+
+# 创建工作流
+workflow = StateGraph(WorkflowState)
+
+# 添加节点
+workflow.add_node("planning", planning_node)
+workflow.add_node("analysis", analysis_node)
+workflow.add_node("answer", answer_node)
+
+# 设置入口
+workflow.set_entry_point("planning")
+
+# 添加边
+workflow.add_edge("analysis", END)
+workflow.add_edge("answer", END)
+
+# 添加条件边
+def route_from_planning(state: WorkflowState) -> str:
+    return state["current_step"]
+
+workflow.add_conditional_edges(
+    "planning",
+    route_from_planning,
+    {
+        "analyze": "analysis",
+        "answer": "answer"
+    }
+)
+
+# 编译并运行
+app = workflow.compile()
+
+result = app.invoke({
+    "question": "请分析这个数据",
+    "current_step": "",
+    "results": []
+})
+
+print("最终结果:", result)
+```
+
+#### 练习3.2: 条件路由与循环
+**目标**: 掌握复杂工作流设计
+
+**任务**:
+1. 实现条件判断路由
+2. 添加循环逻辑
+3. 处理错误状态
+4. 实现状态回退
+
+### Phase 4: 多Agent协作系统
+
+#### 练习4.1: Agent通信
+**目标**: 实现Agent间数据传递
+
+**任务**:
+1. 设计Agent接口
+2. 实现数据传递机制
+3. 处理依赖关系
+4. 实现结果聚合
+
+#### 练习4.2: 错误处理与重试
+**目标**: 构建健壮的Agent系统
+
+**任务**:
+1. 实现重试机制
+2. 处理部分失败
+3. 实现降级策略
+4. 添加监控和告警
+
+### Phase 5: 实际项目实现
+
+#### 练习5.1: 数据管理Agent
+**目标**: 实现数据加载和预处理
+
+基于项目中的 `data_manager.py`,实现:
+1. CSV文件读取
+2. 数据验证
+3. 格式转换
+4. 错误处理
+
+#### 练习5.2: 规划Agent
+**目标**: 实现决策制定逻辑
+
+基于项目中的 `planning_agent.py`,实现:
+1. 状态分析
+2. 决策制定
+3. 任务分配
+4. 结果输出
+
+#### 练习5.3: 大纲生成Agent
+**目标**: 实现结构化内容生成
+
+基于项目中的 `outline_agent.py`,实现:
+1. 提示词设计
+2. JSON输出解析
+3. 内容验证
+4. 错误重试
+
+#### 练习5.4: 指标计算Agent
+**目标**: 实现外部API调用
+
+基于项目中的 `rules_engine_metric_calculation_agent.py`,实现:
+1. HTTP请求处理
+2. 响应解析
+3. 错误处理
+4. 结果缓存
+
+### Phase 6: 系统集成与优化
+
+#### 练习6.1: 工作流编排
+**目标**: 集成所有组件
+
+基于项目中的 `complete_agent_flow_rule.py`,实现:
+1. 组件组装
+2. 状态管理
+3. 执行控制
+4. 结果汇总
+
+#### 练习6.2: 日志与监控
+**目标**: 添加可观测性
+
+基于项目中的日志系统,实现:
+1. 结构化日志
+2. 性能监控
+3. 错误追踪
+4. 结果统计
+
+## 🏆 进阶挑战
+
+### 挑战1: 扩展新功能
+- 添加数据可视化Agent
+- 实现报告生成Agent
+- 添加用户交互界面
+
+### 挑战2: 性能优化
+- 实现并发处理
+- 添加缓存机制
+- 优化API调用
+
+### 挑战3: 错误恢复
+- 实现自动重试策略
+- 添加备用方案
+- 实现优雅降级
+
+### 挑战4: 配置管理
+- 支持动态配置加载
+- 实现环境切换
+- 添加配置验证
+
+## 📚 学习资源
+
+### 官方文档
+- [LangChain文档](https://python.langchain.com/)
+- [LangGraph文档](https://langchain-ai.github.io/langgraph/)
+- [Pydantic文档](https://docs.pydantic.dev/)
+
+### 推荐书籍
+- "Building LLM Applications with LangChain"
+- "Prompt Engineering Guide"
+- "Designing Data-Intensive Applications"
+
+### 社区资源
+- LangChain GitHub Issues
+- Stack Overflow
+- Reddit r/LangChain
+
+## 🎯 练习建议
+
+1. **循序渐进**: 按照Phase顺序逐步学习
+2. **动手实践**: 每学一个概念就动手实现
+3. **调试练习**: 学会使用断点和日志调试
+4. **代码审查**: 学习阅读和理解现有代码
+5. **文档编写**: 为自己的代码编写文档
+6. **测试驱动**: 学习编写单元测试
+
+## 🚀 快速开始
+
+```bash
+# 1. 克隆项目
+git clone <repository-url>
+cd tx_flow_analysis
+
+# 2. 创建环境
+python -m venv venv
+source venv/bin/activate
+
+# 3. 安装依赖
+pip install -r requirements.txt
+
+# 4. 配置环境变量
+cp env_example.txt .env
+# 编辑 .env 文件
+
+# 5. 开始练习!
+```bash
+# 运行基础Agent示例
+python examples/basic_agent.py
+
+# 运行状态机示例
+python examples/state_machine.py
+```
+
+# 6. 练习项目结构
+```
+examples/
+├── basic_agent.py          # 基础Agent示例(LLM调用、提示词设计)
+├── state_machine.py        # LangGraph状态机示例(工作流设计、条件路由)
+├── advanced_agent.py       # 高级Agent功能(结构化输出、错误处理、重试机制)
+└── multi_agent.py          # 多Agent协作系统(任务分配、结果传递)
+```
+```
+
+## 💡 学习提示
+
+- **不懂就问**: 遇到问题及时求助
+- **多动手**: 理论学习后立即实践
+- **记笔记**: 记录学习心得和坑点
+- **分享交流**: 加入技术社区交流
+- **持之以恒**: Agent开发需要时间积累
+
+祝你在LangChain和LangGraph的学习之路上取得成功!🚀

+ 10 - 1
README

@@ -46,4 +46,13 @@ pip install pdfplumber   (pdf交易流水单解析需要)
 - api_results_3/     # 第三次运行
 - ...
 
-这样便于整理和对比不同运行的结果。
+这样便于整理和对比不同运行的结果。
+
+五、学习资源
+==========
+
+📚 **练习手册**: [PRACTICE_GUIDE.md](PRACTICE_GUIDE.md)
+- LangChain & LangGraph Agent开发详细教程
+- 从基础概念到项目实践的完整学习路径
+- 包含6个学习阶段,20+个具体练习
+- 适合想要深入学习Agent开发的开发者

+ 20 - 0
examples/__init__.py

@@ -0,0 +1,20 @@
+"""
+交易流水分析项目 - 学习示例
+==========================
+
+这个包包含了LangChain和LangGraph的学习示例代码。
+
+示例列表:
+- basic_agent.py: 基础Agent实现与LLM调用
+- state_machine.py: LangGraph状态机工作流
+- advanced_agent.py: 高级Agent功能(结构化输出、错误处理、重试机制)
+- multi_agent.py: 多Agent协作系统(任务分配与结果传递)
+
+快速开始:
+    python examples/basic_agent.py
+    python examples/state_machine.py
+    python examples/advanced_agent.py
+    python examples/multi_agent.py
+"""
+
+__version__ = "1.0.0"

+ 351 - 0
examples/advanced_agent.py

@@ -0,0 +1,351 @@
+#!/usr/bin/env python3
+"""
+高级Agent示例 - 结构化输出与错误处理
+===================================
+
+这个文件展示了高级Agent功能,包含:
+1. Pydantic数据模型
+2. 结构化JSON输出
+3. 错误处理与重试
+4. 结果验证
+5. 日志记录
+
+运行方法:
+python examples/advanced_agent.py
+"""
+
+import os
+import sys
+import json
+from typing import Dict, Any, List, Optional
+from datetime import datetime
+from dotenv import load_dotenv
+
+# 加载环境变量
+load_dotenv()
+
+try:
+    from langchain_openai import ChatOpenAI
+    from langchain_core.prompts import ChatPromptTemplate
+    from langchain_core.output_parsers import JsonOutputParser
+    from pydantic import BaseModel, Field, ValidationError
+except ImportError as e:
+    print(f"❌ 缺少依赖包: {e}")
+    print("请运行: pip install langchain langchain-openai pydantic python-dotenv")
+    sys.exit(1)
+
+
+class AnalysisMetrics(BaseModel):
+    """分析指标数据模型"""
+    total_records: int = Field(description="总记录数")
+    valid_records: int = Field(description="有效记录数")
+    invalid_records: int = Field(description="无效记录数")
+    completeness_rate: float = Field(description="完整性比率", ge=0, le=1)
+    unique_values: int = Field(description="唯一值数量")
+
+
+class DataQualityReport(BaseModel):
+    """数据质量报告"""
+    dataset_name: str = Field(description="数据集名称")
+    analysis_date: str = Field(description="分析日期")
+    overall_score: float = Field(description="整体质量评分", ge=0, le=1)
+    metrics: AnalysisMetrics = Field(description="详细指标")
+    recommendations: List[str] = Field(description="改进建议")
+    warnings: List[str] = Field(description="警告信息")
+
+
+class AdvancedAgent:
+    """高级Agent - 支持结构化输出和错误处理"""
+
+    def __init__(self, max_retries: int = 3):
+        """初始化高级Agent"""
+        api_key = os.getenv('DEEPSEEK_API_KEY')
+        if not api_key:
+            raise ValueError("请在.env文件中设置DEEPSEEK_API_KEY")
+
+        # 初始化LLM
+        self.llm = ChatOpenAI(
+            model="deepseek-chat",
+            api_key=api_key,
+            base_url="https://api.deepseek.com",
+            temperature=0.1
+        )
+
+        self.max_retries = max_retries
+        self.call_history = []
+
+        print("✅ AdvancedAgent初始化完成")
+
+    def create_quality_analysis_prompt(self) -> ChatPromptTemplate:
+        """创建数据质量分析提示词"""
+
+        parser = JsonOutputParser(pydantic_object=DataQualityReport)
+
+        template = """你是一个专业的数据质量分析师,请分析提供的数据集并生成详细的质量报告。
+
+数据集信息:
+名称: {dataset_name}
+记录数量: {record_count}
+数据样例: {data_sample}
+
+请按以下JSON格式输出分析报告:
+
+{format_instructions}
+
+要求:
+1. 计算总记录数、有效记录数、无效记录数
+2. 评估数据完整性(0-1之间的分数)
+3. 识别唯一值数量
+4. 给出整体质量评分(0-1之间)
+5. 提供至少2条改进建议
+6. 如果发现问题,请在warnings中列出
+
+确保所有数值字段都是数字类型,字符串字段是字符串类型。"""
+
+        return ChatPromptTemplate.from_template(
+            template,
+            partial_variables={"format_instructions": parser.get_format_instructions()}
+        )
+
+    def analyze_data_quality(self, dataset_name: str, data: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """
+        分析数据质量
+
+        Args:
+            dataset_name: 数据集名称
+            data: 数据列表
+
+        Returns:
+            分析结果
+        """
+        start_time = datetime.now()
+
+        try:
+            # 准备数据样例
+            data_sample = json.dumps(data[:3], ensure_ascii=False, indent=2) if data else "无数据"
+
+            # 创建提示词
+            prompt = self.create_quality_analysis_prompt()
+            chain = prompt | self.llm | JsonOutputParser(pydantic_object=DataQualityReport)
+
+            # 执行分析(带重试机制)
+            result = None
+            last_error = None
+
+            for attempt in range(self.max_retries):
+                try:
+                    print(f"🔍 执行数据质量分析 (尝试 {attempt + 1}/{self.max_retries})")
+
+                    raw_result = chain.invoke({
+                        "dataset_name": dataset_name,
+                        "record_count": len(data),
+                        "data_sample": data_sample
+                    })
+
+                    # 验证和转换结果
+                    result = DataQualityReport(**raw_result)
+                    break
+
+                except (ValidationError, json.JSONDecodeError) as e:
+                    last_error = f"解析错误: {str(e)}"
+                    print(f"⚠️ 尝试 {attempt + 1} 失败: {last_error}")
+
+                    if attempt < self.max_retries - 1:
+                        continue
+
+                except Exception as e:
+                    last_error = f"执行错误: {str(e)}"
+                    print(f"❌ 尝试 {attempt + 1} 失败: {last_error}")
+
+                    if attempt < self.max_retries - 1:
+                        continue
+
+            # 记录调用历史
+            end_time = datetime.now()
+            call_record = {
+                "timestamp": end_time.isoformat(),
+                "duration": (end_time - start_time).total_seconds(),
+                "function": "analyze_data_quality",
+                "dataset": dataset_name,
+                "success": result is not None,
+                "attempts": attempt + 1 if 'attempt' in locals() else 1,
+                "error": last_error if result is None else None
+            }
+            self.call_history.append(call_record)
+
+            if result:
+                print("✅ 数据质量分析完成")
+                return {
+                    "success": True,
+                    "result": result.dict(),
+                    "call_info": call_record
+                }
+            else:
+                print(f"❌ 数据质量分析失败: {last_error}")
+                return {
+                    "success": False,
+                    "error": last_error,
+                    "call_info": call_record
+                }
+
+        except Exception as e:
+            end_time = datetime.now()
+            error_msg = f"意外错误: {str(e)}"
+
+            call_record = {
+                "timestamp": end_time.isoformat(),
+                "duration": (end_time - start_time).total_seconds(),
+                "function": "analyze_data_quality",
+                "dataset": dataset_name,
+                "success": False,
+                "attempts": 1,
+                "error": error_msg
+            }
+            self.call_history.append(call_record)
+
+            print(f"❌ 数据质量分析异常: {error_msg}")
+            return {
+                "success": False,
+                "error": error_msg,
+                "call_info": call_record
+            }
+
+    def generate_summary_report(self) -> Dict[str, Any]:
+        """生成调用历史摘要报告"""
+        if not self.call_history:
+            return {"message": "暂无调用历史"}
+
+        total_calls = len(self.call_history)
+        successful_calls = sum(1 for call in self.call_history if call["success"])
+        failed_calls = total_calls - successful_calls
+
+        total_duration = sum(call["duration"] for call in self.call_history)
+        avg_duration = total_duration / total_calls if total_calls > 0 else 0
+
+        return {
+            "total_calls": total_calls,
+            "successful_calls": successful_calls,
+            "failed_calls": failed_calls,
+            "success_rate": successful_calls / total_calls if total_calls > 0 else 0,
+            "total_duration": round(total_duration, 2),
+            "average_duration": round(avg_duration, 2),
+            "call_history": self.call_history[-5:]  # 最近5次调用
+        }
+
+
+def create_sample_data() -> List[Dict[str, Any]]:
+    """创建示例数据"""
+    return [
+        {
+            "id": 1,
+            "name": "张三",
+            "age": 25,
+            "city": "北京",
+            "salary": 5000,
+            "department": "技术部"
+        },
+        {
+            "id": 2,
+            "name": "李四",
+            "age": 30,
+            "city": "上海",
+            "salary": 6000,
+            "department": "销售部"
+        },
+        {
+            "id": 3,
+            "name": "王五",
+            "age": None,  # 缺失数据
+            "city": "广州",
+            "salary": None,  # 缺失数据
+            "department": "技术部"
+        },
+        {
+            "id": 4,
+            "name": "赵六",
+            "age": 35,
+            "city": "深圳",
+            "salary": 7000,
+            "department": "财务部"
+        },
+        {
+            "id": 5,
+            "name": "张三",  # 重复数据
+            "age": 25,
+            "city": "北京",
+            "salary": 5000,
+            "department": "技术部"
+        }
+    ]
+
+
+def main():
+    """主函数 - 演示高级Agent功能"""
+    print("🚀 高级Agent示例 - 结构化输出与错误处理")
+    print("=" * 60)
+
+    try:
+        # 创建Agent实例
+        agent = AdvancedAgent(max_retries=2)
+
+        # 准备测试数据
+        sample_data = create_sample_data()
+
+        print(f"\n🧪 测试数据:")
+        print(f"数据集: 示例员工数据")
+        print(f"记录数: {len(sample_data)}")
+        print(f"数据样例: {json.dumps(sample_data[0], ensure_ascii=False, indent=2)}")
+
+        # 执行数据质量分析
+        print("\n🔍 开始数据质量分析...")
+        result = agent.analyze_data_quality("员工数据集", sample_data)
+
+        if result["success"]:
+            analysis_result = result["result"]
+            print("\n✅ 分析结果:")
+            print(f"整体质量评分: {analysis_result['overall_score']:.2f}")
+            print(f"完整性比率: {analysis_result['metrics']['completeness_rate']:.2f}")
+            print(f"唯一值数量: {analysis_result['metrics']['unique_values']}")
+
+            print(f"\n📋 改进建议:")
+            for i, rec in enumerate(analysis_result['recommendations'][:3], 1):
+                print(f"{i}. {rec}")
+
+            if analysis_result['warnings']:
+                print(f"\n⚠️ 警告信息:")
+                for warning in analysis_result['warnings'][:2]:
+                    print(f"• {warning}")
+        else:
+            print(f"❌ 分析失败: {result['error']}")
+
+        # 显示调用历史摘要
+        print("\n📊 调用历史摘要:")
+        summary = agent.generate_summary_report()
+        print(f"总调用次数: {summary['total_calls']}")
+        print(f"成功率: {summary['success_rate']:.1%}")
+        print(f"平均耗时: {summary['average_duration']:.2f}秒")
+
+        print("\n🎉 高级Agent示例完成!")
+        print("\n💡 学习要点:")
+        print("1. Pydantic数据模型: 使用BaseModel定义结构化数据")
+        print("2. 输出解析器: JsonOutputParser自动解析JSON输出")
+        print("3. 错误处理: 捕获ValidationError和网络异常")
+        print("4. 重试机制: 自动重试失败的请求")
+        print("5. 调用跟踪: 记录所有API调用的历史")
+        print("6. 结果验证: 使用Pydantic验证输出格式")
+
+        print("\n📚 下一步学习:")
+        print("- 查看项目中的实际Agent代码")
+        print("- 学习PRACTICE_GUIDE.md中的Phase 4内容")
+        print("- 尝试修改示例代码,添加新功能")
+
+    except Exception as e:
+        print(f"❌ 运行出错: {e}")
+        print("\n🔧 故障排除:")
+        print("1. 检查.env文件中的API密钥")
+        print("2. 确认网络连接正常")
+        print("3. 检查pydantic版本: pip show pydantic")
+
+
+if __name__ == "__main__":
+    main()

+ 140 - 0
examples/basic_agent.py

@@ -0,0 +1,140 @@
+#!/usr/bin/env python3
+"""
+基础Agent示例 - LangChain学习起点
+=================================
+
+这个文件展示了如何创建一个简单的Agent,包含:
+1. 基础的LLM调用
+2. 简单的提示词设计
+3. 基本的结果处理
+
+运行方法:
+python examples/basic_agent.py
+"""
+
+import os
+import sys
+from typing import Dict, Any
+from dotenv import load_dotenv
+
+# 加载环境变量
+load_dotenv()
+
+try:
+    from langchain_openai import ChatOpenAI
+    from langchain_core.prompts import ChatPromptTemplate
+except ImportError as e:
+    print(f"❌ 缺少依赖包: {e}")
+    print("请运行: pip install langchain langchain-openai python-dotenv")
+    sys.exit(1)
+
+
+class BasicAgent:
+    """基础Agent示例"""
+
+    def __init__(self):
+        """初始化Agent"""
+        api_key = os.getenv('DEEPSEEK_API_KEY')
+        if not api_key:
+            raise ValueError("请在.env文件中设置DEEPSEEK_API_KEY")
+
+        # 初始化LLM
+        self.llm = ChatOpenAI(
+            model="deepseek-chat",
+            api_key=api_key,
+            base_url="https://api.deepseek.com",
+            temperature=0.1
+        )
+
+        # 创建提示词模板
+        self.prompt_template = ChatPromptTemplate.from_messages([
+            ("system", "你是一个专业的数据分析师,请简洁地回答用户的问题。"),
+            ("user", "{question}")
+        ])
+
+        print("✅ BasicAgent初始化完成")
+
+    def analyze_data(self, question: str, data_sample: str = None) -> Dict[str, Any]:
+        """
+        分析数据并返回结果
+
+        Args:
+            question: 用户问题
+            data_sample: 数据样本(可选)
+
+        Returns:
+            分析结果字典
+        """
+        try:
+            # 构建完整的查询
+            full_question = question
+            if data_sample:
+                full_question += f"\n\n数据样本: {data_sample}"
+
+            # 创建链并调用
+            chain = self.prompt_template | self.llm
+            response = chain.invoke({"question": full_question})
+
+            return {
+                "success": True,
+                "question": question,
+                "data_sample": data_sample,
+                "answer": response.content,
+                "model": "deepseek-chat"
+            }
+
+        except Exception as e:
+            return {
+                "success": False,
+                "question": question,
+                "error": str(e)
+            }
+
+
+def main():
+    """主函数 - 演示基本Agent功能"""
+    print("🚀 基础Agent示例")
+    print("=" * 50)
+
+    try:
+        # 创建Agent实例
+        agent = BasicAgent()
+
+        # 示例问题
+        questions = [
+            "请解释什么是数据分析",
+            "如何提高数据分析的准确性?",
+            "数据可视化的重要性是什么?"
+        ]
+
+        print("\n🧪 测试Agent功能:")
+        print("-" * 30)
+
+        # 逐个测试问题
+        for i, question in enumerate(questions, 1):
+            print(f"\n📋 问题 {i}: {question}")
+
+            result = agent.analyze_data(question)
+
+            if result["success"]:
+                print(f"✅ 回答: {result['answer'][:100]}...")
+            else:
+                print(f"❌ 错误: {result['error']}")
+
+        print("\n🎉 基础Agent示例完成!")
+        print("\n💡 下一步学习建议:")
+        print("1. 查看 examples/state_machine.py - 学习LangGraph状态机")
+        print("2. 查看 examples/advanced_agent.py - 学习高级Agent功能")
+        print("3. 阅读 PRACTICE_GUIDE.md - 获取完整学习路径")
+
+    except Exception as e:
+        print(f"❌ 运行出错: {e}")
+        print("\n🔧 故障排除:")
+        print("1. 检查 .env 文件是否存在")
+        print("2. 确认 DEEPSEEK_API_KEY 已正确设置")
+        print("3. 确认网络连接正常")
+        print("4. 运行: pip install -r requirements.txt")
+
+
+if __name__ == "__main__":
+    main()

+ 417 - 0
examples/multi_agent.py

@@ -0,0 +1,417 @@
+#!/usr/bin/env python3
+"""
+多Agent协作示例 - Agent间通信与任务分配
+=====================================
+
+这个文件展示了多个Agent如何协作完成复杂任务,包含:
+1. 任务分解与分配
+2. Agent间结果传递
+3. 结果聚合与整合
+4. 协作工作流设计
+
+运行方法:
+python examples/multi_agent.py
+"""
+
+import os
+import sys
+import asyncio
+from typing import Dict, Any, List, Optional
+from datetime import datetime
+from dotenv import load_dotenv
+
+# 加载环境变量
+load_dotenv()
+
+try:
+    from langchain_openai import ChatOpenAI
+    from langchain_core.prompts import ChatPromptTemplate
+except ImportError as e:
+    print(f"❌ 缺少依赖包: {e}")
+    print("请运行: pip install langchain langchain-openai python-dotenv")
+    sys.exit(1)
+
+
+class TaskResult:
+    """任务结果类"""
+    def __init__(self, agent_name: str, task_name: str, success: bool,
+                 result: Any = None, error: str = None):
+        self.agent_name = agent_name
+        self.task_name = task_name
+        self.success = success
+        self.result = result
+        self.error = error
+        self.timestamp = datetime.now()
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "agent": self.agent_name,
+            "task": self.task_name,
+            "success": self.success,
+            "result": self.result,
+            "error": self.error,
+            "timestamp": self.timestamp.isoformat()
+        }
+
+
+class BaseAgent:
+    """基础Agent类"""
+
+    def __init__(self, name: str, specialty: str):
+        self.name = name
+        self.specialty = specialty
+
+        api_key = os.getenv('DEEPSEEK_API_KEY')
+        if not api_key:
+            raise ValueError("请在.env文件中设置DEEPSEEK_API_KEY")
+
+        self.llm = ChatOpenAI(
+            model="deepseek-chat",
+            api_key=api_key,
+            base_url="https://api.deepseek.com",
+            temperature=0.1
+        )
+
+        self.completed_tasks = []
+
+    async def execute_task(self, task_name: str, **kwargs) -> TaskResult:
+        """执行任务的通用方法"""
+        raise NotImplementedError("子类必须实现execute_task方法")
+
+    def record_task(self, task_result: TaskResult):
+        """记录完成的任务"""
+        self.completed_tasks.append(task_result)
+
+
+class DataAnalyzerAgent(BaseAgent):
+    """数据分析Agent"""
+
+    def __init__(self):
+        super().__init__("DataAnalyzer", "数据分析与统计")
+        self.analysis_prompt = ChatPromptTemplate.from_messages([
+            ("system", "你是一个专业的数据分析师,擅长发现数据中的模式和趋势。"),
+            ("user", "请分析以下数据:\n\n{data}\n\n请提供关键发现和趋势分析。")
+        ])
+
+    async def execute_task(self, task_name: str, **kwargs) -> TaskResult:
+        try:
+            data = kwargs.get("data", "")
+            chain = self.analysis_prompt | self.llm
+
+            result = await chain.ainvoke({"data": data})
+
+            task_result = TaskResult(
+                agent_name=self.name,
+                task_name=task_name,
+                success=True,
+                result=result.content
+            )
+
+            self.record_task(task_result)
+            return task_result
+
+        except Exception as e:
+            task_result = TaskResult(
+                agent_name=self.name,
+                task_name=task_name,
+                success=False,
+                error=str(e)
+            )
+            self.record_task(task_result)
+            return task_result
+
+
+class ReportGeneratorAgent(BaseAgent):
+    """报告生成Agent"""
+
+    def __init__(self):
+        super().__init__("ReportGenerator", "报告撰写与格式化")
+        self.report_prompt = ChatPromptTemplate.from_messages([
+            ("system", "你是一个专业的报告撰写专家,擅长将分析结果整理成清晰的报告。"),
+            ("user", "基于以下分析结果生成一份完整的分析报告:\n\n{analysis_result}\n\n请包含:执行摘要、详细分析、结论和建议。")
+        ])
+
+    async def execute_task(self, task_name: str, **kwargs) -> TaskResult:
+        try:
+            analysis_result = kwargs.get("analysis_result", "")
+            chain = self.report_prompt | self.llm
+
+            result = await chain.ainvoke({"analysis_result": analysis_result})
+
+            task_result = TaskResult(
+                agent_name=self.name,
+                task_name=task_name,
+                success=True,
+                result=result.content
+            )
+
+            self.record_task(task_result)
+            return task_result
+
+        except Exception as e:
+            task_result = TaskResult(
+                agent_name=self.name,
+                task_name=task_name,
+                success=False,
+                error=str(e)
+            )
+            self.record_task(task_result)
+            return task_result
+
+
+class QualityCheckerAgent(BaseAgent):
+    """质量检查Agent"""
+
+    def __init__(self):
+        super().__init__("QualityChecker", "质量评估与验证")
+        self.quality_prompt = ChatPromptTemplate.from_messages([
+            ("system", "你是一个严格的质量检查专家,负责评估分析结果的质量和准确性。"),
+            ("user", "请检查以下分析报告的质量:\n\n{report}\n\n请评估:1)准确性 2)完整性 3)清晰度 4)实用性。给出评分(1-10)和改进建议。")
+        ])
+
+    async def execute_task(self, task_name: str, **kwargs) -> TaskResult:
+        try:
+            report = kwargs.get("report", "")
+            chain = self.quality_prompt | self.llm
+
+            result = await chain.ainvoke({"report": report})
+
+            task_result = TaskResult(
+                agent_name=self.name,
+                task_name=task_name,
+                success=True,
+                result=result.content
+            )
+
+            self.record_task(task_result)
+            return task_result
+
+        except Exception as e:
+            task_result = TaskResult(
+                agent_name=self.name,
+                task_name=task_name,
+                success=False,
+                error=str(e)
+            )
+            self.record_task(task_result)
+            return task_result
+
+
+class MultiAgentSystem:
+    """多Agent协作系统"""
+
+    def __init__(self):
+        self.agents = {
+            "analyzer": DataAnalyzerAgent(),
+            "reporter": ReportGeneratorAgent(),
+            "checker": QualityCheckerAgent()
+        }
+        self.workflow_history = []
+
+    def get_agent(self, agent_type: str) -> BaseAgent:
+        """获取指定类型的Agent"""
+        return self.agents.get(agent_type)
+
+    async def execute_workflow(self, data: str) -> Dict[str, Any]:
+        """
+        执行完整的工作流:
+        1. 数据分析
+        2. 报告生成
+        3. 质量检查
+        """
+        print("🚀 开始多Agent协作工作流")
+        print(f"📊 输入数据: {data[:50]}...")
+
+        workflow_start = datetime.now()
+        results = {}
+
+        try:
+            # 步骤1: 数据分析
+            print("\n1️⃣ 执行数据分析...")
+            analyzer = self.get_agent("analyzer")
+            analysis_result = await analyzer.execute_task("data_analysis", data=data)
+            results["analysis"] = analysis_result.to_dict()
+
+            if not analysis_result.success:
+                raise Exception(f"数据分析失败: {analysis_result.error}")
+
+            print("✅ 数据分析完成")
+
+            # 步骤2: 报告生成
+            print("\n2️⃣ 生成分析报告...")
+            reporter = self.get_agent("reporter")
+            report_result = await reporter.execute_task(
+                "report_generation",
+                analysis_result=analysis_result.result
+            )
+            results["report"] = report_result.to_dict()
+
+            if not report_result.success:
+                raise Exception(f"报告生成失败: {report_result.error}")
+
+            print("✅ 报告生成完成")
+
+            # 步骤3: 质量检查
+            print("\n3️⃣ 执行质量检查...")
+            checker = self.get_agent("checker")
+            quality_result = await checker.execute_task(
+                "quality_check",
+                report=report_result.result
+            )
+            results["quality"] = quality_result.to_dict()
+
+            if not quality_result.success:
+                raise Exception(f"质量检查失败: {quality_result.error}")
+
+            print("✅ 质量检查完成")
+
+            # 记录成功的工作流
+            workflow_end = datetime.now()
+            workflow_record = {
+                "success": True,
+                "start_time": workflow_start.isoformat(),
+                "end_time": workflow_end.isoformat(),
+                "duration": (workflow_end - workflow_start).total_seconds(),
+                "steps_completed": 3,
+                "results": results
+            }
+
+            self.workflow_history.append(workflow_record)
+
+            print(f"\n🎉 工作流执行成功!总耗时: {workflow_record['duration']:.2f}秒")
+            return {
+                "success": True,
+                "workflow": workflow_record,
+                "final_report": report_result.result,
+                "quality_assessment": quality_result.result
+            }
+
+        except Exception as e:
+            workflow_end = datetime.now()
+            error_msg = str(e)
+
+            workflow_record = {
+                "success": False,
+                "start_time": workflow_start.isoformat(),
+                "end_time": workflow_end.isoformat(),
+                "duration": (workflow_end - workflow_start).total_seconds(),
+                "error": error_msg,
+                "results": results
+            }
+
+            self.workflow_history.append(workflow_record)
+
+            print(f"\n❌ 工作流执行失败: {error_msg}")
+            return {
+                "success": False,
+                "error": error_msg,
+                "workflow": workflow_record
+            }
+
+    def get_system_status(self) -> Dict[str, Any]:
+        """获取系统状态"""
+        agent_status = {}
+        for agent_type, agent in self.agents.items():
+            agent_status[agent_type] = {
+                "name": agent.name,
+                "specialty": agent.specialty,
+                "tasks_completed": len(agent.completed_tasks)
+            }
+
+        return {
+            "agents": agent_status,
+            "total_workflows": len(self.workflow_history),
+            "successful_workflows": sum(1 for w in self.workflow_history if w["success"]),
+            "failed_workflows": sum(1 for w in self.workflow_history if not w["success"])
+        }
+
+
+def create_sample_data() -> str:
+    """创建示例数据"""
+    data = """
+销售数据分析:
+- 2024年第一季度总销售额:150万元
+- 各月销售额:1月45万、2月50万、3月55万
+- 主要产品:A产品(40%)、B产品(35%)、C产品(25%)
+- 客户数量:新增客户120个,回头客80个
+- 地区分布:华北35%、华东30%、华南20%、其他15%
+
+趋势观察:
+- 销售额逐月上升,增长率约11%
+- A产品销售占比略有下降
+- 新客户获取率稳步提升
+"""
+    return data.strip()
+
+
+async def main():
+    """主函数 - 演示多Agent协作"""
+    print("🚀 多Agent协作示例 - Agent间通信与任务分配")
+    print("=" * 70)
+
+    try:
+        # 创建多Agent系统
+        system = MultiAgentSystem()
+
+        # 显示系统状态
+        status = system.get_system_status()
+        print("🤖 系统初始化完成")
+        print(f"📊 可用Agent: {len(status['agents'])}个")
+        for agent_type, info in status['agents'].items():
+            print(f"   • {info['name']}: {info['specialty']}")
+
+        # 准备测试数据
+        sample_data = create_sample_data()
+        print("\n📋 测试数据:")
+        print(sample_data)
+        print("-" * 50)
+
+        # 执行协作工作流
+        result = await system.execute_workflow(sample_data)
+
+        if result["success"]:
+            print("\n📄 最终分析报告:")
+            report_preview = result["final_report"][:300] + "..." if len(result["final_report"]) > 300 else result["final_report"]
+            print(report_preview)
+
+            print("\n⭐ 质量评估:")
+            quality_preview = result["quality_assessment"][:200] + "..." if len(result["quality_assessment"]) > 200 else result["quality_assessment"]
+            print(quality_preview)
+
+            # 显示系统最终状态
+            final_status = system.get_system_status()
+            print("\n📊 执行统计:")
+            print(f"总工作流数: {final_status['total_workflows']}")
+            print(f"成功执行: {final_status['successful_workflows']}")
+            print(f"失败执行: {final_status['failed_workflows']}")
+
+            for agent_type, info in final_status['agents'].items():
+                print(f"{info['name']}完成任务数: {info['tasks_completed']}")
+
+        else:
+            print(f"❌ 执行失败: {result['error']}")
+
+        print("\n🎉 多Agent协作示例完成!")
+        print("\n💡 多Agent协作学习要点:")
+        print("1. 任务分解: 将复杂任务拆分为多个专门步骤")
+        print("2. Agent分工: 每个Agent负责特定的专业任务")
+        print("3. 结果传递: Agent间通过TaskResult传递数据")
+        print("4. 错误处理: 任何一个环节失败都会终止整个流程")
+        print("5. 状态跟踪: 记录每个Agent的执行历史")
+        print("6. 协作编排: 设计合理的执行顺序和依赖关系")
+
+        print("\n📚 下一步学习:")
+        print("- 查看项目中的CompleteAgentFlow实现")
+        print("- 学习PRACTICE_GUIDE.md中的Phase 4和5")
+        print("- 尝试添加新的Agent类型到协作系统中")
+
+    except Exception as e:
+        print(f"❌ 运行出错: {e}")
+        print("\n🔧 故障排除:")
+        print("1. 检查.env文件中的API密钥")
+        print("2. 确认网络连接正常")
+        print("3. 检查Agent初始化是否成功")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

+ 281 - 0
examples/state_machine.py

@@ -0,0 +1,281 @@
+#!/usr/bin/env python3
+"""
+LangGraph状态机示例 - 学习工作流设计
+==================================
+
+这个文件展示了如何使用LangGraph创建状态机工作流,包含:
+1. 状态定义
+2. 节点函数
+3. 条件路由
+4. 工作流执行
+
+运行方法:
+python examples/state_machine.py
+"""
+
+import sys
+from typing import TypedDict, Annotated, List
+
+try:
+    from langgraph.graph import StateGraph, START, END
+except ImportError as e:
+    print(f"❌ 缺少依赖包: {e}")
+    print("请运行: pip install langgraph")
+    sys.exit(1)
+
+
+class AnalysisState(TypedDict):
+    """分析工作流的状态"""
+    question: str
+    data: str
+    current_step: str
+    analysis_result: str
+    report: str
+    steps_completed: Annotated[List[str], "add"]  # 使用注解支持列表追加
+
+
+def planning_node(state: AnalysisState) -> AnalysisState:
+    """
+    规划节点 - 分析用户需求并制定计划
+
+    这是一个决策节点,决定下一步的分析方向
+    """
+    print("📋 规划阶段: 分析用户需求")
+
+    question = state["question"].lower()
+
+    # 简单的决策逻辑
+    if "趋势" in question or "变化" in question:
+        next_step = "trend_analysis"
+        plan = "执行趋势分析"
+    elif "对比" in question or "比较" in question:
+        next_step = "comparison_analysis"
+        plan = "执行对比分析"
+    elif "汇总" in question or "总结" in question:
+        next_step = "summary_analysis"
+        plan = "执行汇总分析"
+    else:
+        next_step = "general_analysis"
+        plan = "执行一般性分析"
+
+    print(f"   决策: {plan}")
+
+    return {
+        **state,
+        "current_step": next_step,
+        "steps_completed": ["planning"]
+    }
+
+
+def data_processing_node(state: AnalysisState) -> AnalysisState:
+    """数据处理节点 - 准备数据用于分析"""
+    print("🔧 数据处理阶段: 准备分析数据")
+
+    # 模拟数据处理
+    processed_data = f"已处理的数据: {state['data'][:50]}..."
+
+    print("   数据处理完成")
+
+    return {
+        **state,
+        "data": processed_data,
+        "steps_completed": ["data_processing"]
+    }
+
+
+def trend_analysis_node(state: AnalysisState) -> AnalysisState:
+    """趋势分析节点 - 专门处理趋势分析"""
+    print("📈 趋势分析阶段: 识别数据变化趋势")
+
+    # 模拟趋势分析
+    result = f"趋势分析结果: 数据显示上升趋势,增长率为15%"
+
+    print(f"   分析结果: {result}")
+
+    return {
+        **state,
+        "analysis_result": result,
+        "steps_completed": ["trend_analysis"]
+    }
+
+
+def comparison_analysis_node(state: AnalysisState) -> AnalysisState:
+    """对比分析节点 - 专门处理对比分析"""
+    print("⚖️ 对比分析阶段: 比较不同数据集")
+
+    # 模拟对比分析
+    result = f"对比分析结果: A组数据优于B组数据,差异显著"
+
+    print(f"   分析结果: {result}")
+
+    return {
+        **state,
+        "analysis_result": result,
+        "steps_completed": ["comparison_analysis"]
+    }
+
+
+def general_analysis_node(state: AnalysisState) -> AnalysisState:
+    """一般分析节点 - 处理通用分析需求"""
+    print("🔍 一般分析阶段: 执行标准数据分析")
+
+    # 模拟一般分析
+    result = f"一般分析结果: 数据整体表现良好,符合预期"
+
+    print(f"   分析结果: {result}")
+
+    return {
+        **state,
+        "analysis_result": result,
+        "steps_completed": ["general_analysis"]
+    }
+
+
+def report_generation_node(state: AnalysisState) -> AnalysisState:
+    """报告生成节点 - 生成最终分析报告"""
+    print("📄 报告生成阶段: 整理分析结果")
+
+    # 生成报告
+    report = f"""
+📊 数据分析报告
+================
+
+问题: {state['question']}
+数据: {state['data']}
+分析结果: {state['analysis_result']}
+
+执行步骤: {', '.join(state['steps_completed'])}
+
+结论: 分析完成,建议根据具体情况采取相应措施。
+"""
+
+    print("   报告生成完成")
+
+    return {
+        **state,
+        "report": report.strip(),
+        "steps_completed": ["report_generation"]
+    }
+
+
+def route_from_planning(state: AnalysisState) -> str:
+    """
+    从规划节点路由到具体分析节点
+
+    这是一个条件路由函数,根据规划结果决定下一个节点
+    """
+    return state["current_step"]
+
+
+def create_analysis_workflow() -> StateGraph:
+    """创建分析工作流"""
+
+    # 创建状态图
+    workflow = StateGraph(AnalysisState)
+
+    # 添加节点
+    workflow.add_node("planning", planning_node)
+    workflow.add_node("data_processing", data_processing_node)
+    workflow.add_node("trend_analysis", trend_analysis_node)
+    workflow.add_node("comparison_analysis", comparison_analysis_node)
+    workflow.add_node("general_analysis", general_analysis_node)
+    workflow.add_node("report_generation", report_generation_node)
+
+    # 设置入口点
+    workflow.set_entry_point("planning")
+
+    # 添加固定边
+    workflow.add_edge("data_processing", "report_generation")
+    workflow.add_edge("trend_analysis", "report_generation")
+    workflow.add_edge("comparison_analysis", "report_generation")
+    workflow.add_edge("general_analysis", "report_generation")
+    workflow.add_edge("report_generation", END)
+
+    # 添加条件边
+    workflow.add_conditional_edges(
+        "planning",
+        route_from_planning,
+        {
+            "trend_analysis": "data_processing",  # 趋势分析需要先处理数据
+            "comparison_analysis": "data_processing",  # 对比分析也需要先处理数据
+            "general_analysis": "data_processing",  # 一般分析也需要先处理数据
+        }
+    )
+
+    return workflow
+
+
+def main():
+    """主函数 - 演示状态机工作流"""
+    print("🚀 LangGraph状态机示例")
+    print("=" * 50)
+
+    try:
+        # 创建工作流
+        workflow = create_analysis_workflow()
+        app = workflow.compile()
+
+        # 测试用例
+        test_cases = [
+            {
+                "question": "这个季度的数据趋势如何?",
+                "data": "Q1: 100, Q2: 120, Q3: 140, Q4: 160"
+            },
+            {
+                "question": "比较A产品和B产品的销量",
+                "data": "A产品: 500件, B产品: 450件"
+            },
+            {
+                "question": "请分析这份数据",
+                "data": "各类数据指标: 正常范围"
+            }
+        ]
+
+        print("\n🧪 测试不同类型的分析:")
+        print("-" * 40)
+
+        for i, test_case in enumerate(test_cases, 1):
+            print(f"\n📋 测试用例 {i}:")
+            print(f"问题: {test_case['question']}")
+            print(f"数据: {test_case['data']}")
+
+            # 执行工作流
+            result = app.invoke({
+                "question": test_case["question"],
+                "data": test_case["data"],
+                "current_step": "",
+                "analysis_result": "",
+                "report": "",
+                "steps_completed": []
+            })
+
+            print("\n✅ 执行完成:")
+            print(f"执行步骤: {result['steps_completed']}")
+            print(f"分析结果: {result['analysis_result']}")
+
+            # 显示报告预览
+            report_preview = result['report'][:200] + "..." if len(result['report']) > 200 else result['report']
+            print(f"报告预览: {report_preview}")
+
+        print("\n🎉 状态机示例完成!")
+        print("\n💡 LangGraph学习要点:")
+        print("1. 状态定义: 使用TypedDict定义工作流状态")
+        print("2. 节点函数: 每个节点处理特定逻辑")
+        print("3. 条件路由: 根据状态动态选择下一节点")
+        print("4. 状态传递: 通过return更新状态")
+        print("5. 工作流编译: 调用compile()创建可执行应用")
+
+        print("\n📚 下一步学习:")
+        print("- 查看 examples/advanced_agent.py - 学习高级Agent功能")
+        print("- 阅读 llmops/ 目录下的实际Agent代码")
+        print("- 学习 PRACTICE_GUIDE.md 中的Phase 3内容")
+
+    except Exception as e:
+        print(f"❌ 运行出错: {e}")
+        print("\n🔧 故障排除:")
+        print("1. 确认已安装langgraph: pip install langgraph")
+        print("2. 检查Python版本是否支持(推荐3.10+)")
+
+
+if __name__ == "__main__":
+    main()