目标：工具、ToolNode、agent执行器、工具调用状态、函数调用

以上是逻辑和状态的一部分

环境准备和导入

输入控制：TypedDict，Literal，List和Dict

数据处理：requests处理HTTP网络请求

AI工具：ToolNode和tools_condition处理工具调用逻辑，tool装饰器创建可调用工具，BaseModel和Field用于定义数据验证模型。

# 安装必要的库
# !pip install langgraph langchain-core langchain-openai langchain-community

import os
from typing import Annotated, Literal, TypedDict, List, Dict, Any
from langgraph import StateGraph, START, END
from langgraph.graph import MessagesState
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_core.pydantic_v1 import BaseModel, Field
import json
import random
import time
import requests
from datetime import datetime, timedelta

print("环境准备完成")

1. 基础工具定义和使用

首先学习如何定义和使用基本的工具

@tool：类似springboot的@Component，告诉ai你可以用

# 使用装饰器定义工具
@tool
def calculator(expression: str) -> str:
    """计算数学表达式的结果。
    
    Args:
        expression: 要计算的数学表达式，例如 "2 + 3 * 4"
        
    Returns:
        计算结果的字符串表示
    """
    try:
        # 安全的数学表达式计算
        allowed_chars = '0123456789+-*/().'
        if all(c in allowed_chars or c.isspace() for c in expression):
            result = eval(expression)
            return f"计算结果: {result}"
        else:
            return "错误: 表达式包含不允许的字符"
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool  
def get_current_time() -> str:
    """获取当前的日期和时间。
    
    Returns:
        当前日期和时间的字符串
    """
    current_time = datetime.now()
    return f"当前时间: {current_time.strftime('%Y-%m-%d %H:%M:%S')}"

@tool
def search_weather(city: str) -> str:
    """查询指定城市的天气信息（模拟）。
    
    Args:
        city: 要查询的城市名称
        
    Returns:
        天气信息字符串
    """
    # 模拟天气数据
    weather_options = [
        "晴天，气温25°C",
        "多云，气温22°C", 
        "小雨，气温18°C",
        "阴天，气温20°C"
    ]
    weather = random.choice(weather_options)
    return f"{city}的天气: {weather}"

@tool
def generate_random_number(min_val: int = 1, max_val: int = 100) -> str:
    """生成指定范围内的随机数。
    
    Args:
        min_val: 最小值（默认1）
        max_val: 最大值（默认100）
        
    Returns:
        随机数字符串
    """
    number = random.randint(min_val, max_val)
    return f"随机数: {number} (范围: {min_val}-{max_val})"

# 创建工具列表
basic_tools = [calculator, get_current_time, search_weather, generate_random_number]

# 测试工具
print("基础工具测试：")
print(calculator.invoke({"expression": "2 + 3 * 4"}))
print(get_current_time.invoke({}))
print(search_weather.invoke({"city": "北京"}))
print(generate_random_number.invoke({"min_val": 10, "max_val": 50}))

2. ToolNode的使用

ToolNode是LangGraph提供的专门处理工具调用的节点类型

🆚 核心区别：

方面	@tool 装饰器	ToolNode
作用	定义工具函数	执行工具调用
层级	函数级别	节点级别
使用	装饰普通函数	在StateGraph中添加节点
输入	函数参数	AI的tool_calls请求
输出	函数返回值	格式化的ToolMessage

🎯 简单理解：

• @tool = 制作工具（锤子、螺丝刀）

• ToolNode = 工具包（收纳所有工具，知道什么时候用什么工具）

AI说："我要用锤子" → ToolNode听到 → 从工具包拿出锤子(@tool函数) → 使用 → 把结果告诉AI

# 定义带有工具调用的状态
class AgentState(TypedDict):
    messages: Annotated[list, "消息列表"]
    
# 模拟LLM节点（实际应用中会使用真实的LLM）
def mock_llm_node(state: AgentState) -> AgentState:
    """模拟LLM决策工具调用的节点"""
    messages = state.get("messages", [])
    last_message = messages[-1] if messages else None
    
    if isinstance(last_message, HumanMessage):
        content = last_message.content.lower()
        
        # 根据用户输入决定调用哪个工具
        if "计算" in content or "算" in content:
            # 提取数学表达式（简单示例）
            expression = "2 + 3"  # 实际应用中需要更智能的提取
            if "*" in content or "乘" in content:
                expression = "5 * 6"
            elif "/" in content or "除" in content:
                expression = "20 / 4"
            
            ai_message = AIMessage(
                content="我来帮你计算",
                tool_calls=[
                    {
                        "name": "calculator",
                        "args": {"expression": expression},
                        "id": "calc_" + str(random.randint(1000, 9999))
                    }
                ]
            )
        elif "时间" in content or "几点" in content:
            ai_message = AIMessage(
                content="我来查询当前时间",
                tool_calls=[
                    {
                        "name": "get_current_time",
                        "args": {},
                        "id": "time_" + str(random.randint(1000, 9999))
                    }
                ]
            )
        elif "天气" in content:
            # 简单提取城市名（实际应用需要更复杂的NER）
            city = "北京"  # 默认城市
            if "上海" in content:
                city = "上海"
            elif "深圳" in content:
                city = "深圳"
                
            ai_message = AIMessage(
                content=f"我来查询{city}的天气",
                tool_calls=[
                    {
                        "name": "search_weather",
                        "args": {"city": city},
                        "id": "weather_" + str(random.randint(1000, 9999))
                    }
                ]
            )
        elif "随机" in content:
            ai_message = AIMessage(
                content="我来生成一个随机数",
                tool_calls=[
                    {
                        "name": "generate_random_number",
                        "args": {"min_val": 1, "max_val": 100},
                        "id": "random_" + str(random.randint(1000, 9999))
                    }
                ]
            )
        else:
            ai_message = AIMessage(content="我没有找到合适的工具来处理您的请求。请尝试询问时间、天气、计算或随机数。")
    else:
        # 处理工具调用结果
        tool_messages = [msg for msg in messages if isinstance(msg, ToolMessage)]
        if tool_messages:
            latest_tool_result = tool_messages[-1].content
            ai_message = AIMessage(content=f"根据工具调用结果：{latest_tool_result}")
        else:
            ai_message = AIMessage(content="请告诉我您需要什么帮助。")
    
    return {"messages": messages + [ai_message]}

# 创建ToolNode
tool_node = ToolNode(basic_tools)

# 决定是否需要调用工具
def should_continue(state: AgentState) -> Literal["tools", "end"]:
    messages = state.get("messages", [])
    last_message = messages[-1] if messages else None
    
    # 检查最后一条消息是否包含工具调用
    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        return "tools"
    return "end"

# 构建带有工具调用的图
tool_workflow = StateGraph(AgentState)

# 添加节点
tool_workflow.add_node("llm", mock_llm_node)
tool_workflow.add_node("tools", tool_node)

# 设置入口点
tool_workflow.set_entry_point("llm")

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

# 工具执行后回到LLM
tool_workflow.add_edge("tools", "llm")

# 编译图
tool_app = tool_workflow.compile()

# 测试工具调用
test_queries = [
    "请帮我计算5乘以8",
    "现在几点了？",
    "北京今天天气怎么样？",
    "给我一个随机数",
    "你好"  # 不需要工具调用的查询
]

print("\n工具调用测试：")
for query in test_queries:
    print(f"\n用户: {query}")
    result = tool_app.invoke({"messages": [HumanMessage(content=query)]})
    
    # 显示对话历史
    for msg in result["messages"]:
        if isinstance(msg, AIMessage):
            print(f"AI: {msg.content}")
            if msg.tool_calls:
                print(f"  工具调用: {msg.tool_calls}")
        elif isinstance(msg, ToolMessage):
            print(f"  工具结果: {msg.content}")

3. 复杂工具系统

构建更复杂的工具系统，包含多种类型的工具

# 文件操作工具
@tool
def read_file(file_path: str) -> str:
    """读取文件内容（模拟）。
    
    Args:
        file_path: 文件路径
        
    Returns:
        文件内容或错误信息
    """
    # 模拟文件内容
    mock_files = {
        "data.txt": "这是数据文件的内容\n包含一些重要信息",
        "config.json": '{"api_key": "xxx", "timeout": 30}',
        "log.txt": "2023-01-01 10:00:00 - 系统启动\n2023-01-01 10:01:00 - 加载配置"
    }
    
    if file_path in mock_files:
        return f"文件 {file_path} 的内容:\n{mock_files[file_path]}"
    else:
        return f"错误: 找不到文件 {file_path}"

@tool
def write_file(file_path: str, content: str) -> str:
    """写入文件（模拟）。
    
    Args:
        file_path: 文件路径
        content: 要写入的内容
        
    Returns:
        操作结果
    """
    # 模拟写入操作
    return f"成功将内容写入文件 {file_path}\n内容长度: {len(content)} 字符"

# 数据处理工具
@tool
def analyze_data(data: str) -> str:
    """分析数据并提供统计信息。
    
    Args:
        data: 要分析的数据（用逗号分隔的数字）
        
    Returns:
        数据分析结果
    """
    try:
        numbers = [float(x.strip()) for x in data.split(',')]
        if not numbers:
            return "错误: 没有找到有效的数字"
            
        count = len(numbers)
        total = sum(numbers)
        average = total / count
        minimum = min(numbers)
        maximum = max(numbers)
        
        return f"""数据分析结果:
数量: {count}
总和: {total}
平均值: {average:.2f}
最小值: {minimum}
最大值: {maximum}"""
    except Exception as e:
        return f"数据分析错误: {str(e)}"

# 网络请求工具
@tool
def fetch_url(url: str) -> str:
    """获取URL内容（模拟）。
    
    Args:
        url: 要获取的URL
        
    Returns:
        URL内容或错误信息
    """
    # 模拟网络请求
    mock_responses = {
        "https://api.example.com/status": '{"status": "online", "version": "1.0"}',
        "https://api.example.com/data": '{"data": [1, 2, 3, 4, 5]}',
        "https://news.example.com": "今日头条新闻内容"
    }
    
    if url in mock_responses:
        return f"成功获取 {url}:\n{mock_responses[url]}"
    else:
        return f"模拟请求 {url}\n返回: 404 - 页面未找到"

# 邮件发送工具
@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件（模拟）。
    
    Args:
        to: 收件人邮箱
        subject: 邮件主题
        body: 邮件内容
        
    Returns:
        发送结果
    """
    return f"""邮件发送成功!
收件人: {to}
主题: {subject}
内容长度: {len(body)} 字符
发送时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"""

# 数据库查询工具
@tool
def query_database(table: str, condition: str = "") -> str:
    """查询数据库（模拟）。
    
    Args:
        table: 表名
        condition: 查询条件（可选）
        
    Returns:
        查询结果
    """
    # 模拟数据库数据
    mock_data = {
        "users": [
            {"id": 1, "name": "张三", "age": 25},
            {"id": 2, "name": "李四", "age": 30},
            {"id": 3, "name": "王五", "age": 35}
        ],
        "orders": [
            {"id": 101, "user_id": 1, "amount": 100.0},
            {"id": 102, "user_id": 2, "amount": 200.0}
        ],
        "products": [
            {"id": 1, "name": "笔记本电脑", "price": 5000},
            {"id": 2, "name": "手机", "price": 3000}
        ]
    }
    
    if table in mock_data:
        data = mock_data[table]
        result_count = len(data)
        
        return f"""查询表 '{table}' 成功:
条件: {condition if condition else '无'}
结果数量: {result_count}
数据: {json.dumps(data, ensure_ascii=False, indent=2)}"""
    else:
        return f"错误: 找不到表 '{table}'"

# 创建复杂工具列表
advanced_tools = [
    calculator, get_current_time, search_weather, generate_random_number,
    read_file, write_file, analyze_data, fetch_url, send_email, query_database
]

print("复杂工具系统测试：")
print("\n1. 文件操作:")
print(read_file.invoke({"file_path": "data.txt"}))

print("\n2. 数据分析:")
print(analyze_data.invoke({"data": "10, 20, 30, 40, 50"}))

print("\n3. 网络请求:")
print(fetch_url.invoke({"url": "https://api.example.com/status"}))

print("\n4. 数据库查询:")
print(query_database.invoke({"table": "users", "condition": "age > 25"}))

4. Agent执行器构建

构建一个完整的Agent执行器，能够智能地选择和执行工具

# 增强的Agent状态
class EnhancedAgentState(TypedDict):
    messages: Annotated[list, "消息历史"]
    current_task: str
    tool_results: Dict[str, Any]
    execution_step: int
    max_steps: int
    error_count: int
    context: Dict[str, Any]

# 智能LLM节点（更复杂的决策逻辑）
def intelligent_llm_node(state: EnhancedAgentState) -> EnhancedAgentState:
    """智能LLM节点，能够根据上下文做出更好的决策"""
    messages = state.get("messages", [])
    execution_step = state.get("execution_step", 0)
    max_steps = state.get("max_steps", 10)
    tool_results = state.get("tool_results", {})
    context = state.get("context", {})
    
    # 检查是否超过最大步数
    if execution_step >= max_steps:
        ai_message = AIMessage(content=f"已达到最大执行步数({max_steps})，停止执行。")
        return {
            **state,
            "messages": messages + [ai_message]
        }
    
    last_message = messages[-1] if messages else None
    
    if isinstance(last_message, HumanMessage):
        content = last_message.content.lower()
        task_type = analyze_task_type(content)
        
        # 根据任务类型制定执行计划
        ai_message, new_context = plan_execution(content, task_type, context)
        
    elif isinstance(last_message, ToolMessage):
        # 处理工具执行结果
        ai_message = process_tool_result(last_message, messages, tool_results, context)
        new_context = context
    else:
        ai_message = AIMessage(content="请告诉我您需要什么帮助。")
        new_context = context
    
    return {
        **state,
        "messages": messages + [ai_message],
        "execution_step": execution_step + 1,
        "context": new_context
    }

def analyze_task_type(content: str) -> str:
    """分析任务类型"""
    if any(word in content for word in ["计算", "算", "数学"]):
        return "calculation"
    elif any(word in content for word in ["文件", "读取", "写入", "保存"]):
        return "file_operation"
    elif any(word in content for word in ["数据", "分析", "统计"]):
        return "data_analysis"
    elif any(word in content for word in ["邮件", "发送", "通知"]):
        return "communication"
    elif any(word in content for word in ["查询", "搜索", "数据库"]):
        return "query"
    elif any(word in content for word in ["天气", "时间", "信息"]):
        return "information"
    else:
        return "general"

def plan_execution(content: str, task_type: str, context: Dict) -> tuple:
    """制定执行计划"""
    
    if task_type == "calculation":
        # 提取表达式
        expression = extract_math_expression(content)
        ai_message = AIMessage(
            content=f"我来计算表达式: {expression}",
            tool_calls=[
                {
                    "name": "calculator",
                    "args": {"expression": expression},
                    "id": f"calc_{random.randint(1000, 9999)}"
                }
            ]
        )
        new_context = {**context, "task_type": "calculation", "expression": expression}
        
    elif task_type == "file_operation":
        if "读取" in content or "read" in content:
            file_path = extract_file_path(content)
            ai_message = AIMessage(
                content=f"我来读取文件: {file_path}",
                tool_calls=[
                    {
                        "name": "read_file",
                        "args": {"file_path": file_path},
                        "id": f"read_{random.randint(1000, 9999)}"
                    }
                ]
            )
        else:
            # 写入文件的例子
            ai_message = AIMessage(
                content="我来创建一个示例文件",
                tool_calls=[
                    {
                        "name": "write_file",
                        "args": {
                            "file_path": "example.txt",
                            "content": "这是一个示例文件内容"
                        },
                        "id": f"write_{random.randint(1000, 9999)}"
                    }
                ]
            )
        new_context = {**context, "task_type": "file_operation"}
        
    elif task_type == "data_analysis":
        # 提取数据或使用示例数据
        data = extract_data(content) or "1,2,3,4,5,6,7,8,9,10"
        ai_message = AIMessage(
            content=f"我来分析数据: {data}",
            tool_calls=[
                {
                    "name": "analyze_data",
                    "args": {"data": data},
                    "id": f"analyze_{random.randint(1000, 9999)}"
                }
            ]
        )
        new_context = {**context, "task_type": "data_analysis", "data": data}
        
    elif task_type == "query":
        table = extract_table_name(content)
        ai_message = AIMessage(
            content=f"我来查询数据库表: {table}",
            tool_calls=[
                {
                    "name": "query_database",
                    "args": {"table": table},
                    "id": f"query_{random.randint(1000, 9999)}"
                }
            ]
        )
        new_context = {**context, "task_type": "query", "table": table}
        
    elif task_type == "information":
        if "时间" in content:
            ai_message = AIMessage(
                content="我来获取当前时间",
                tool_calls=[
                    {
                        "name": "get_current_time",
                        "args": {},
                        "id": f"time_{random.randint(1000, 9999)}"
                    }
                ]
            )
        else:  # 天气查询
            city = extract_city(content)
            ai_message = AIMessage(
                content=f"我来查询{city}的天气",
                tool_calls=[
                    {
                        "name": "search_weather",
                        "args": {"city": city},
                        "id": f"weather_{random.randint(1000, 9999)}"
                    }
                ]
            )
        new_context = {**context, "task_type": "information"}
        
    else:
        ai_message = AIMessage(content="请更具体地描述您需要什么帮助。我可以帮您：计算、文件操作、数据分析、查询数据库、获取信息等。")
        new_context = context
        
    return ai_message, new_context

# 辅助函数
def extract_math_expression(content: str) -> str:
    """从文本中提取数学表达式（简化版本）"""
    # 简单的表达式提取逻辑
    import re
    # 寻找数学表达式模式
    patterns = [
        r'(\d+(?:\.\d+)?\s*[+\-*/]\s*\d+(?:\.\d+)?(?:\s*[+\-*/]\s*\d+(?:\.\d+)?)*)',
        r'(\d+(?:\.\d+)?)\s*[+加]\s*(\d+(?:\.\d+)?)',
        r'(\d+(?:\.\d+)?)\s*[\-减]\s*(\d+(?:\.\d+)?)',
        r'(\d+(?:\.\d+)?)\s*[*×乘]\s*(\d+(?:\.\d+)?)',
        r'(\d+(?:\.\d+)?)\s*[/÷除]\s*(\d+(?:\.\d+)?)'
    ]
    
    for pattern in patterns:
        match = re.search(pattern, content)
        if match:
            if len(match.groups()) == 1:
                return match.group(1)
            else:
                # 处理中文运算符
                if '加' in content:
                    return f"{match.group(1)} + {match.group(2)}"
                elif '减' in content:
                    return f"{match.group(1)} - {match.group(2)}"
                elif any(x in content for x in ['乘', '×']):
                    return f"{match.group(1)} * {match.group(2)}"
                elif any(x in content for x in ['除', '÷']):
                    return f"{match.group(1)} / {match.group(2)}"
    
    return "2 + 2"  # 默认表达式

def extract_file_path(content: str) -> str:
    """提取文件路径"""
    if "data.txt" in content:
        return "data.txt"
    elif "config" in content:
        return "config.json"
    elif "log" in content:
        return "log.txt"
    return "data.txt"

def extract_data(content: str) -> str:
    """提取数据"""
    import re
    # 寻找逗号分隔的数字
    match = re.search(r'([0-9, ]+)', content)
    if match:
        return match.group(1).strip()
    return None

def extract_table_name(content: str) -> str:
    """提取表名"""
    if "用户" in content or "user" in content:
        return "users"
    elif "订单" in content or "order" in content:
        return "orders"
    elif "产品" in content or "product" in content:
        return "products"
    return "users"

def extract_city(content: str) -> str:
    """提取城市名"""
    cities = ["北京", "上海", "深圳", "广州", "杭州", "成都"]
    for city in cities:
        if city in content:
            return city
    return "北京"

def process_tool_result(tool_message: ToolMessage, messages: list, tool_results: Dict, context: Dict) -> AIMessage:
    """处理工具执行结果"""
    result = tool_message.content
    task_type = context.get("task_type", "general")
    
    # 根据任务类型提供不同的响应
    if task_type == "calculation":
        return AIMessage(content=f"计算完成！{result}")
    elif task_type == "file_operation":
        return AIMessage(content=f"文件操作完成：{result}")
    elif task_type == "data_analysis":
        return AIMessage(content=f"数据分析结果：\n{result}")
    elif task_type == "query":
        return AIMessage(content=f"数据库查询结果：\n{result}")
    else:
        return AIMessage(content=f"操作完成：{result}")

# 错误处理节点
def error_handler(state: EnhancedAgentState) -> EnhancedAgentState:
    """处理执行错误"""
    messages = state.get("messages", [])
    error_count = state.get("error_count", 0) + 1
    
    if error_count >= 3:
        ai_message = AIMessage(content="抱歉，连续出现多次错误，请稍后重试或联系管理员。")
    else:
        ai_message = AIMessage(content=f"执行过程中出现错误，正在重试... (第{error_count}次)")
    
    return {
        **state,
        "messages": messages + [ai_message],
        "error_count": error_count
    }

# 构建增强的Agent执行器
def enhanced_should_continue(state: EnhancedAgentState) -> Literal["tools", "error", "end"]:
    messages = state.get("messages", [])
    error_count = state.get("error_count", 0)
    last_message = messages[-1] if messages else None
    
    # 检查错误次数
    if error_count >= 3:
        return "error"
    
    # 检查是否需要调用工具
    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        return "tools"
    
    return "end"

# 创建增强的工具节点
enhanced_tool_node = ToolNode(advanced_tools)

# 构建增强的Agent图
enhanced_agent_workflow = StateGraph(EnhancedAgentState)

# 添加节点
enhanced_agent_workflow.add_node("llm", intelligent_llm_node)
enhanced_agent_workflow.add_node("tools", enhanced_tool_node)
enhanced_agent_workflow.add_node("error_handler", error_handler)

# 设置入口
enhanced_agent_workflow.set_entry_point("llm")

# 添加条件边
enhanced_agent_workflow.add_conditional_edges(
    "llm",
    enhanced_should_continue,
    {
        "tools": "tools",
        "error": "error_handler",
        "end": END
    }
)

# 工具执行后回到LLM
enhanced_agent_workflow.add_edge("tools", "llm")
enhanced_agent_workflow.add_edge("error_handler", END)

# 编译增强Agent
enhanced_agent_app = enhanced_agent_workflow.compile()

# 测试增强Agent
complex_queries = [
    "请计算15乘以23加上67",
    "帮我读取data.txt文件的内容",
    "分析这些数据：10, 15, 20, 25, 30, 35, 40",
    "查询用户表的信息",
    "现在几点了？",
    "上海今天天气如何？"
]

print("\n增强Agent执行器测试：")
for query in complex_queries:
    print(f"\n========== 查询: {query} ==========")
    
    initial_state = {
        "messages": [HumanMessage(content=query)],
        "current_task": query,
        "tool_results": {},
        "execution_step": 0,
        "max_steps": 5,
        "error_count": 0,
        "context": {}
    }
    
    result = enhanced_agent_app.invoke(initial_state)
    
    # 显示执行过程
    for i, msg in enumerate(result["messages"]):
        if isinstance(msg, HumanMessage):
            print(f"👤 用户: {msg.content}")
        elif isinstance(msg, AIMessage):
            print(f"🤖 AI: {msg.content}")
            if msg.tool_calls:
                for tool_call in msg.tool_calls:
                    print(f"   🔧 调用工具: {tool_call['name']}({tool_call['args']})")
        elif isinstance(msg, ToolMessage):
            print(f"   ⚙️ 工具结果: {msg.content[:100]}..." if len(msg.content) > 100 else f"   ⚙️ 工具结果: {msg.content}")
    
    print(f"执行步数: {result['execution_step']}, 错误次数: {result['error_count']}")

5. 错误处理和异常恢复

实现健壮的错误处理机制

# 带有错误处理的工具
@tool
def unreliable_service(task: str, fail_rate: float = 0.3) -> str:
    """模拟不可靠的服务调用。
    
    Args:
        task: 要执行的任务描述
        fail_rate: 失败率（0.0-1.0）
        
    Returns:
        任务结果或错误信息
    """
    if random.random() < fail_rate:
        error_types = [
            "网络超时",
            "服务不可用", 
            "权限不足",
            "请求格式错误",
            "系统维护中"
        ]
        error = random.choice(error_types)
        raise Exception(f"服务调用失败: {error}")
    
    return f"任务 '{task}' 执行成功！结果：模拟数据处理完成"

@tool
def validate_input(input_data: str, data_type: str = "string") -> str:
    """验证输入数据。
    
    Args:
        input_data: 要验证的数据
        data_type: 期望的数据类型 (string, number, email, url)
        
    Returns:
        验证结果
    """
    import re
    
    if data_type == "number":
        try:
            float(input_data)
            return f"✅ 数字验证通过: {input_data}"
        except ValueError:
            raise ValueError(f"❌ 无效的数字格式: {input_data}")
            
    elif data_type == "email":
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if re.match(pattern, input_data):
            return f"✅ 邮箱格式验证通过: {input_data}"
        else:
            raise ValueError(f"❌ 无效的邮箱格式: {input_data}")
            
    elif data_type == "url":
        pattern = r'^https?://[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'
        if re.match(pattern, input_data):
            return f"✅ URL格式验证通过: {input_data}"
        else:
            raise ValueError(f"❌ 无效的URL格式: {input_data}")
    
    return f"✅ 字符串验证通过: {input_data}"

# 错误恢复状态
class ErrorRecoveryState(TypedDict):
    messages: Annotated[list, "消息历史"]
    current_task: str
    retry_count: Dict[str, int]
    max_retries: int
    error_history: list
    fallback_strategies: Dict[str, str]
    execution_status: str  # running, error, recovered, failed

# 错误恢复LLM节点
def error_recovery_llm(state: ErrorRecoveryState) -> ErrorRecoveryState:
    """带有错误恢复能力的LLM节点"""
    messages = state.get("messages", [])
    retry_count = state.get("retry_count", {})
    max_retries = state.get("max_retries", 3)
    error_history = state.get("error_history", [])
    
    last_message = messages[-1] if messages else None

    #新加的赋值，不然报错
    new_retry_count = retry_count.copy()  # 先赋初始值
    new_error_history = error_history.copy()
    new_status = "running"
    
    if isinstance(last_message, HumanMessage):
        content = last_message.content.lower()
        
        if "验证" in content:
            # 输入验证任务
            if "邮箱" in content:
                test_email = "invalid-email"  # 故意使用无效邮箱来触发错误
                ai_message = AIMessage(
                    content=f"我来验证邮箱格式: {test_email}",
                    tool_calls=[
                        {
                            "name": "validate_input",
                            "args": {"input_data": test_email, "data_type": "email"},
                            "id": f"validate_{random.randint(1000, 9999)}"
                        }
                    ]
                )
            else:
                ai_message = AIMessage(
                    content="我来验证输入数据",
                    tool_calls=[
                        {
                            "name": "validate_input",
                            "args": {"input_data": "test123", "data_type": "number"},
                            "id": f"validate_{random.randint(1000, 9999)}"
                        }
                    ]
                )
        
        elif "服务" in content or "调用" in content:
            # 不可靠服务调用
            ai_message = AIMessage(
                content="我来调用服务",
                tool_calls=[
                    {
                        "name": "unreliable_service",
                        "args": {"task": "数据处理", "fail_rate": 0.6},
                        "id": f"service_{random.randint(1000, 9999)}"
                    }
                ]
            )
        
        else:
            ai_message = AIMessage(content="请告诉我您需要验证什么数据或调用什么服务。")
          
    
    elif isinstance(last_message, ToolMessage):
        # 检查工具调用是否成功
        if "错误" in last_message.content or "失败" in last_message.content:
            # 处理错误
            tool_name = last_message.name or "unknown_tool"
            current_retries = retry_count.get(tool_name, 0)
            
            if current_retries < max_retries:
                # 重试
                ai_message = AIMessage(content=f"工具调用失败，正在重试... (第{current_retries + 1}次)")
                new_retry_count = {**retry_count, tool_name: current_retries + 1}
                new_error_history = error_history + [{
                    "tool": tool_name,
                    "error": last_message.content,
                    "timestamp": datetime.now().isoformat(),
                    "retry_count": current_retries + 1
                }]
                new_status = "error"
            else:
                # 达到最大重试次数，使用备用策略
                ai_message = AIMessage(content=f"工具 {tool_name} 多次调用失败，使用备用策略。")
                new_retry_count = retry_count
                new_error_history = error_history + [{
                    "tool": tool_name,
                    "error": "达到最大重试次数",
                    "timestamp": datetime.now().isoformat(),
                    "retry_count": current_retries
                }]
                new_status = "failed"
        else:
            # 成功执行
            ai_message = AIMessage(content=f"操作成功完成：{last_message.content}")
            new_retry_count = retry_count
            new_error_history = error_history
            new_status = "recovered" if error_history else "running"
    
    else:
        ai_message = AIMessage(content="请告诉我您需要什么帮助。")
        new_retry_count = retry_count
        new_error_history = error_history
        new_status = "running"
    
    return {
        **state,
        "messages": messages + [ai_message],
        "retry_count": new_retry_count,
        "error_history": new_error_history,
        "execution_status": new_status
    }

# 带错误处理的工具节点
class ErrorHandlingToolNode:
    def __init__(self, tools):
        self.tools = {tool.name: tool for tool in tools}
    
    def __call__(self, state: ErrorRecoveryState) -> ErrorRecoveryState:
        messages = state.get("messages", [])
        last_message = messages[-1] if messages else None
        
        if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
            return state
        
        new_messages = messages.copy()
        
        for tool_call in last_message.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            call_id = tool_call["id"]
            
            if tool_name in self.tools:
                try:
                    # 执行工具
                    result = self.tools[tool_name].invoke(tool_args)
                    tool_message = ToolMessage(
                        content=result,
                        tool_call_id=call_id,
                        name=tool_name
                    )
                except Exception as e:
                    # 捕获工具执行错误
                    error_msg = f"工具 {tool_name} 执行失败: {str(e)}"
                    tool_message = ToolMessage(
                        content=error_msg,
                        tool_call_id=call_id,
                        name=tool_name
                    )
                
                new_messages.append(tool_message)
            else:
                # 未知工具
                error_msg = f"未知工具: {tool_name}"
                tool_message = ToolMessage(
                    content=error_msg,
                    tool_call_id=call_id,
                    name=tool_name
                )
                new_messages.append(tool_message)
        
        return {**state, "messages": new_messages}

# 创建错误处理工具
error_tools = [unreliable_service, validate_input, calculator]
error_tool_node = ErrorHandlingToolNode(error_tools)

# 决策是否继续
def error_recovery_continue(state: ErrorRecoveryState) -> Literal["tools", "end"]:
    messages = state.get("messages", [])
    execution_status = state.get("execution_status", "running")
    last_message = messages[-1] if messages else None
    
    # 如果状态为失败，结束执行
    if execution_status == "failed":
        return "end"
    
    # 检查是否需要调用工具
    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        return "tools"
    
    return "end"

# 构建错误恢复图
error_recovery_workflow = StateGraph(ErrorRecoveryState)

# 添加节点
error_recovery_workflow.add_node("llm", error_recovery_llm)
error_recovery_workflow.add_node("tools", error_tool_node)

# 设置入口
error_recovery_workflow.set_entry_point("llm")

# 添加条件边
error_recovery_workflow.add_conditional_edges(
    "llm",
    error_recovery_continue,
    {
        "tools": "tools",
        "end": END
    }
)

# 工具执行后回到LLM
error_recovery_workflow.add_edge("tools", "llm")

# 编译错误恢复图
error_recovery_app = error_recovery_workflow.compile()

# 测试错误处理和恢复
error_test_cases = [
    "请调用不可靠的服务",
    "请验证邮箱格式",
    "请验证数字格式"
]

print("\n错误处理和恢复测试：")
for test_case in error_test_cases:
    print(f"\n========== 测试: {test_case} ==========")
    
    initial_state = {
        "messages": [HumanMessage(content=test_case)],
        "current_task": test_case,
        "retry_count": {},
        "max_retries": 2,
        "error_history": [],
        "fallback_strategies": {},
        "execution_status": "running"
    }
    
    result = error_recovery_app.invoke(initial_state)
    
    # 显示执行结果
    for msg in result["messages"]:
        if isinstance(msg, HumanMessage):
            print(f"👤 用户: {msg.content}")
        elif isinstance(msg, AIMessage):
            print(f"🤖 AI: {msg.content}")
            if msg.tool_calls:
                for tool_call in msg.tool_calls:
                    print(f"   🔧 调用工具: {tool_call['name']}")
        elif isinstance(msg, ToolMessage):
            status_icon = "❌" if ("错误" in msg.content or "失败" in msg.content) else "✅"
            print(f"   {status_icon} 工具结果: {msg.content}")
    
    print(f"最终状态: {result['execution_status']}")
    print(f"重试计数: {result['retry_count']}")
    if result['error_history']:
        print(f"错误历史: {len(result['error_history'])} 个错误")
        for error in result['error_history'][-2:]:  # 只显示最后2个错误
            print(f"  - {error['tool']}: {error['error'][:50]}...")

6. 实践案例：智能数据处理Agent

构建一个完整的智能数据处理Agent系统

# 数据处理专用工具
@tool
def load_dataset(source: str, format_type: str = "csv") -> str:
    """加载数据集（模拟）。
    
    Args:
        source: 数据源名称或路径
        format_type: 数据格式 (csv, json, excel)
        
    Returns:
        数据加载结果
    """
    mock_datasets = {
        "sales_data": {
            "rows": 1000,
            "columns": ["date", "product", "sales", "region"],
            "size": "2.5MB"
        },
        "user_data": {
            "rows": 5000,
            "columns": ["id", "name", "age", "city", "registration_date"],
            "size": "1.8MB"
        },
        "log_data": {
            "rows": 10000,
            "columns": ["timestamp", "level", "message", "source"],
            "size": "5.2MB"
        }
    }
    
    if source in mock_datasets:
        data_info = mock_datasets[source]
        return f"""数据集 '{source}' 加载成功！
格式: {format_type.upper()}
行数: {data_info['rows']:,}
列数: {len(data_info['columns'])}
列名: {', '.join(data_info['columns'])}
文件大小: {data_info['size']}
状态: 已加载到内存"""
    else:
        return f"错误: 找不到数据集 '{source}'"

@tool
def clean_data(dataset: str, operations: str) -> str:
    """清理数据（模拟）。
    
    Args:
        dataset: 数据集名称
        operations: 清理操作列表，用逗号分隔
        
    Returns:
        数据清理结果
    """
    operations_list = [op.strip() for op in operations.split(',')]
    
    results = []
    for op in operations_list:
        if "空值" in op or "null" in op.lower():
            results.append("删除了125个空值记录")
        elif "重复" in op or "duplicate" in op.lower():
            results.append("删除了43个重复记录")
        elif "格式" in op or "format" in op.lower():
            results.append("标准化了日期格式")
        elif "异常" in op or "outlier" in op.lower():
            results.append("识别并标记了15个异常值")
        else:
            results.append(f"执行了操作: {op}")
    
    return f"""数据集 '{dataset}' 清理完成：
执行的操作:
{chr(10).join([f"- {result}" for result in results])}
清理后数据质量得分: 94/100"""

@tool
def statistical_analysis(dataset: str, analysis_type: str) -> str:
    """执行统计分析（模拟）。
    
    Args:
        dataset: 数据集名称
        analysis_type: 分析类型 (basic, correlation, trend, distribution)
        
    Returns:
        统计分析结果
    """
    if analysis_type == "basic":
        return f"""基础统计分析结果 - {dataset}:
平均值: 245.67
中位数: 230.00
标准差: 78.23
最小值: 45.00
最大值: 987.00
四分位数范围: [180.00, 320.00]"""
    
    elif analysis_type == "correlation":
        return f"""相关性分析结果 - {dataset}:
强正相关 (r > 0.7):
- sales & marketing_spend: r = 0.85
- age & experience: r = 0.72

强负相关 (r < -0.7):
- price & demand: r = -0.78

弱相关 (-0.3 < r < 0.3):
- region & satisfaction: r = 0.12"""
    
    elif analysis_type == "trend":
        return f"""趋势分析结果 - {dataset}:
时间序列趋势:
- 整体趋势: 上升 (+12.5%)
- 季节性模式: 检测到季度周期
- 异常点: 发现3个异常时期
- 预测准确度: 87.3%
- 下一周期预测: 预计增长8.2%"""
    
    else:  # distribution
        return f"""分布分析结果 - {dataset}:
数据分布特征:
- 分布类型: 近似正态分布
- 偏度: 0.23 (轻微右偏)
- 峰度: -0.47 (较平坦)
- 正态性检验: p-value = 0.082
- 异常值比例: 2.1%"""

@tool
def generate_visualization(dataset: str, chart_type: str, variables: str) -> str:
    """生成可视化图表（模拟）。
    
    Args:
        dataset: 数据集名称
        chart_type: 图表类型 (bar, line, scatter, heatmap, histogram)
        variables: 要可视化的变量，用逗号分隔
        
    Returns:
        可视化生成结果
    """
    chart_types = {
        "bar": "柱状图",
        "line": "折线图",
        "scatter": "散点图",
        "heatmap": "热力图",
        "histogram": "直方图",
        "pie": "饼图"
    }
    
    chart_name = chart_types.get(chart_type, chart_type)
    variables_list = [v.strip() for v in variables.split(',')]
    
    return f"""可视化图表生成成功！
数据集: {dataset}
图表类型: {chart_name}
变量: {', '.join(variables_list)}
图表尺寸: 1200x800 像素
保存路径: ./charts/{dataset}_{chart_type}_chart.png
交互式版本: ./charts/{dataset}_{chart_type}_interactive.html
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"""

@tool
def export_report(dataset: str, format_type: str = "pdf") -> str:
    """导出分析报告（模拟）。
    
    Args:
        dataset: 数据集名称
        format_type: 报告格式 (pdf, html, docx)
        
    Returns:
        报告导出结果
    """
    report_sections = [
        "执行摘要",
        "数据概览",
        "数据质量评估",
        "统计分析结果",
        "可视化图表",
        "主要发现",
        "建议和结论",
        "附录"
    ]
    
    return f"""分析报告导出完成！
数据集: {dataset}
报告格式: {format_type.upper()}
包含章节: {len(report_sections)} 个
页数: 25 页
文件大小: 3.2 MB
保存路径: ./reports/{dataset}_analysis_report.{format_type}
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

报告章节:
{chr(10).join([f"{i+1}. {section}" for i, section in enumerate(report_sections)])}"""

# 数据处理Agent状态
class DataProcessingAgentState(TypedDict):
    messages: Annotated[list, "消息历史"]
    current_dataset: str
    processing_pipeline: list
    completed_steps: list
    analysis_results: Dict[str, Any]
    workflow_status: str
    next_action: str

# 数据处理规划器
def data_processing_planner(state: DataProcessingAgentState) -> DataProcessingAgentState:
    """数据处理任务规划器"""
    messages = state.get("messages", [])
    last_message = messages[-1] if messages else None
    
    if isinstance(last_message, HumanMessage):
        content = last_message.content.lower()
        
        # 分析用户意图并制定处理计划
        if "分析" in content and "数据" in content:
            # 确定数据集
            dataset = determine_dataset(content)
            
            # 制定处理流水线
            pipeline = create_processing_pipeline(content)
            
            ai_message = AIMessage(
                content=f"我将为您分析 {dataset} 数据集。处理流程包括：{', '.join(pipeline)}",
                tool_calls=[
                    {
                        "name": "load_dataset",
                        "args": {"source": dataset, "format_type": "csv"},
                        "id": f"load_{random.randint(1000, 9999)}"
                    }
                ]
            )
            
            new_state = {
                **state,
                "messages": messages + [ai_message],
                "current_dataset": dataset,
                "processing_pipeline": pipeline,
                "completed_steps": [],
                "workflow_status": "data_loading",
                "next_action": "load_data"
            }
        
        else:
            ai_message = AIMessage(content="请告诉我您想要分析什么数据。我可以帮您分析销售数据、用户数据或日志数据。")
            new_state = {**state, "messages": messages + [ai_message]}
    
    elif isinstance(last_message, ToolMessage):
        # 根据当前工作流状态决定下一步
        new_state = process_workflow_step(state, last_message)
    
    else:
        ai_message = AIMessage(content="请告诉我您需要什么数据分析帮助。")
        new_state = {**state, "messages": messages + [ai_message]}
    
    return new_state

def determine_dataset(content: str) -> str:
    """确定要使用的数据集"""
    if "销售" in content or "sale" in content:
        return "sales_data"
    elif "用户" in content or "user" in content:
        return "user_data"
    elif "日志" in content or "log" in content:
        return "log_data"
    else:
        return "sales_data"  # 默认数据集

def create_processing_pipeline(content: str) -> list:
    """创建处理流水线"""
    pipeline = ["加载数据"]
    
    if "清理" in content or "clean" in content:
        pipeline.append("数据清理")
    
    if any(word in content for word in ["统计", "分析", "statistical"]):
        pipeline.append("统计分析")
    
    if any(word in content for word in ["可视化", "图表", "visualization", "chart"]):
        pipeline.append("生成图表")
    
    if "报告" in content or "report" in content:
        pipeline.append("导出报告")
    
    # 如果没有指定具体操作，添加默认流程
    if len(pipeline) == 1:
        pipeline.extend(["数据清理", "统计分析", "生成图表"])
    
    return pipeline

def process_workflow_step(state: DataProcessingAgentState, tool_result: ToolMessage) -> DataProcessingAgentState:
    """处理工作流步骤"""
    messages = state.get("messages", [])
    current_dataset = state.get("current_dataset", "")
    pipeline = state.get("processing_pipeline", [])
    completed_steps = state.get("completed_steps", [])
    workflow_status = state.get("workflow_status", "")
    
    # 记录完成的步骤
    if workflow_status not in completed_steps:
        completed_steps = completed_steps + [workflow_status]
    
    # 确定下一步
    next_step_map = {
        "data_loading": "data_cleaning",
        "data_cleaning": "statistical_analysis", 
        "statistical_analysis": "visualization",
        "visualization": "report_generation",
        "report_generation": "completed"
    }
    
    next_status = next_step_map.get(workflow_status, "completed")
    
    if next_status == "data_cleaning" and "数据清理" in pipeline:
        ai_message = AIMessage(
            content="数据加载完成，现在开始清理数据",
            tool_calls=[
                {
                    "name": "clean_data",
                    "args": {
                        "dataset": current_dataset,
                        "operations": "删除空值, 删除重复项, 格式标准化"
                    },
                    "id": f"clean_{random.randint(1000, 9999)}"
                }
            ]
        )
    
    elif next_status == "statistical_analysis" and "统计分析" in pipeline:
        ai_message = AIMessage(
            content="数据清理完成，开始统计分析",
            tool_calls=[
                {
                    "name": "statistical_analysis",
                    "args": {"dataset": current_dataset, "analysis_type": "basic"},
                    "id": f"stats_{random.randint(1000, 9999)}"
                }
            ]
        )
    
    elif next_status == "visualization" and "生成图表" in pipeline:
        ai_message = AIMessage(
            content="统计分析完成，生成可视化图表",
            tool_calls=[
                {
                    "name": "generate_visualization",
                    "args": {
                        "dataset": current_dataset,
                        "chart_type": "bar",
                        "variables": "sales, region"
                    },
                    "id": f"viz_{random.randint(1000, 9999)}"
                }
            ]
        )
    
    elif next_status == "report_generation" and "导出报告" in pipeline:
        ai_message = AIMessage(
            content="图表生成完成，导出分析报告",
            tool_calls=[
                {
                    "name": "export_report",
                    "args": {"dataset": current_dataset, "format_type": "pdf"},
                    "id": f"report_{random.randint(1000, 9999)}"
                }
            ]
        )
    
    else:
        # 工作流完成
        progress = f"完成步骤: {len(completed_steps)}/{len(pipeline)}"
        ai_message = AIMessage(content=f"数据分析工作流已完成！\n{progress}\n所有分析结果已保存。")
        next_status = "completed"
    
    return {
        **state,
        "messages": messages + [ai_message],
        "completed_steps": completed_steps,
        "workflow_status": next_status
    }

# 创建数据处理工具节点
data_tools = [load_dataset, clean_data, statistical_analysis, generate_visualization, export_report]
data_tool_node = ToolNode(data_tools)

# 决策是否继续
def data_processing_continue(state: DataProcessingAgentState) -> Literal["tools", "end"]:
    messages = state.get("messages", [])
    workflow_status = state.get("workflow_status", "")
    last_message = messages[-1] if messages else None
    
    if workflow_status == "completed":
        return "end"
    
    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        return "tools"
    
    return "end"

# 构建数据处理Agent
data_processing_workflow = StateGraph(DataProcessingAgentState)

# 添加节点
data_processing_workflow.add_node("planner", data_processing_planner)
data_processing_workflow.add_node("tools", data_tool_node)

# 设置入口
data_processing_workflow.set_entry_point("planner")

# 添加条件边
data_processing_workflow.add_conditional_edges(
    "planner",
    data_processing_continue,
    {
        "tools": "tools",
        "end": END
    }
)

# 工具执行后回到规划器
data_processing_workflow.add_edge("tools", "planner")

# 编译数据处理Agent
data_processing_app = data_processing_workflow.compile()

# 测试数据处理Agent
data_queries = [
    "请分析销售数据，包括数据清理、统计分析和可视化",
    "帮我处理用户数据，生成分析报告",
    "分析日志数据的趋势和异常"
]

print("\n智能数据处理Agent测试：")
for query in data_queries:
    print(f"\n{'='*60}")
    print(f"📊 数据分析请求: {query}")
    print('='*60)
    
    initial_state = {
        "messages": [HumanMessage(content=query)],
        "current_dataset": "",
        "processing_pipeline": [],
        "completed_steps": [],
        "analysis_results": {},
        "workflow_status": "planning",
        "next_action": ""
    }
    
    result = data_processing_app.invoke(initial_state)
    
    # 显示处理过程
    step_counter = 1
    for msg in result["messages"]:
        if isinstance(msg, HumanMessage):
            print(f"👤 用户请求: {msg.content}")
        elif isinstance(msg, AIMessage):
            print(f"\n🤖 步骤 {step_counter}: {msg.content}")
            if msg.tool_calls:
                for tool_call in msg.tool_calls:
                    print(f"   🔧 执行: {tool_call['name']}({', '.join([f'{k}={v}' for k, v in tool_call['args'].items()])})") 
            step_counter += 1
        elif isinstance(msg, ToolMessage):
            print(f"   ✅ 结果: {msg.content[:150]}..." if len(msg.content) > 150 else f"   ✅ 结果: {msg.content}")
    
    print(f"\n📈 工作流状态: {result.get('workflow_status', 'N/A')}")
    print(f"📋 处理流程: {' → '.join(result.get('processing_pipeline', []))}")
    print(f"✔️  已完成步骤: {len(result.get('completed_steps', []))}/{len(result.get('processing_pipeline', []))}")
    
    if result.get('workflow_status') == 'completed':
        print("🎉 数据分析工作流全部完成！")