Agent智能体

┌─────────────────────────────────────────────┐
│                用户界面                      │
├─────────────────────────────────────────────┤
│                对话管理                      │
├─────────────────────────────────────────────┤
│   推理引擎   │   规划模块   │   记忆系统    │
├─────────────────────────────────────────────┤
│              工具调用接口                    │
├─────────────────────────────────────────────┤
│   外部API   │   数据库    │   文件系统     │
└─────────────────────────────────────────────┘

# 示例：智能客服Agent
class CustomerServiceAgent:
    def __init__(self):
        self.tools = {
            "search_order": self.search_order_info,
            "process_refund": self.process_refund,
            "schedule_callback": self.schedule_callback
        }
    
    def handle_query(self, user_message):
        # 理解用户意图
        intent = self.analyze_intent(user_message)
        
        # 选择合适的工具
        if intent == "order_inquiry":
            return self.tools["search_order"](user_message)
        elif intent == "refund_request":
            return self.tools["process_refund"](user_message)

# 工具描述标准化
def search_database(query: str, table: str) -> dict:
    """
    在指定数据表中搜索信息
    
    Args:
        query: 搜索关键词
        table: 目标数据表名
    
    Returns:
        搜索结果字典
    """
    pass

## 系统提示示例

你是一个专业的数据分析助手。请遵循以下规则：

1. 总是先理解用户的分析需求
2. 选择合适的分析方法和工具
3. 提供清晰的分析步骤说明
4. 如果数据不足，主动询问补充信息
5. 结果要包含具体的数字和可视化建议

可用工具：
- pandas_analysis: 数据处理和统计分析
- plot_generator: 生成各种图表
- sql_executor: 执行数据库查询

class AgentErrorHandler:
    def handle_tool_error(self, error, tool_name):
        if "permission_denied" in str(error):
            return "对不起，我没有权限执行这个操作。"
        elif "invalid_parameter" in str(error):
            return f"参数错误，请检查{tool_name}的输入要求。"
        else:
            return "遇到了技术问题，让我换个方式尝试。"

import asyncio

async def parallel_tool_calls(tools_and_params):
    tasks = []
    for tool, params in tools_and_params:
        task = asyncio.create_task(tool(**params))
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    return results