AI Agent工业化落地：任务拆解 + 工具调用 + 反馈优化三板斧

from typing import List
import openai

# 初始化LLM客户端（以OpenAI为例，可替换为本地化模型）
openai.api_key = "your-api-key"

def decompose_task(main_task: str) -> List[str]:
    """
    任务自动拆解函数：规则兜底+LLM优化
    :param main_task: 原始复杂业务任务
    :return: 原子化子任务列表
    """
    # 1. 基于规则的标准化任务拆解（适用于固定流程场景）
    rule_based_tasks = {
        "生成车间月度能耗分析报告并同步至ERP系统": [
            "查询车间近30天各设备能耗数据",
            "计算能耗均值、峰值及异常波动值",
            "生成标准化能耗分析报表（Excel格式）",
            "调用ERP系统API同步报表文件",
            "发送同步完成通知至指定邮箱"
        ],
        "排查生产线设备异常停机问题": [
            "获取设备停机时间及故障代码",
            "查询设备历史运行日志（近24小时）",
            "分析故障代码对应原因",
            "生成故障排查方案",
            "执行方案并验证设备恢复状态"
        ]
    }
    
    # 规则匹配成功则直接返回拆解结果
    if main_task in rule_based_tasks:
        return rule_based_tasks[main_task]
    
    # 2. 规则匹配失败，调用LLM进行智能拆解
    prompt = f"""
    请将以下工业场景业务任务拆解为原子化子任务，要求：
    1. 每个子任务粒度适中，可直接调用工具执行；
    2. 子任务间无循环依赖，按执行顺序排列；
    3. 子任务数量控制在3-8个，避免冗余。
    原始任务：{main_task}
    输出格式：仅返回子任务列表，每行一个，无需额外说明。
    """
    
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    
    # 解析LLM输出，生成子任务列表
    subtasks = [task.strip() for task in response.choices[0].message.content.split("\n") if task.strip()]
    return subtasks

# 测试示例
if __name__ == "__main__":
    main_task = "生成车间月度能耗分析报告并同步至ERP系统"
    subtasks = decompose_task(main_task)
    print("原子化子任务列表：")
    for idx, task in enumerate(subtasks, 1):
        print(f"{idx}. {task}")

from typing import Dict, Any, Optional
import time

# ------------------------------
# 1. 工具定义（模拟工业场景常用工具）
# ------------------------------
def query_energy_db(start_date: str, end_date: str, equipment_id: str) -> Dict[str, Any]:
    """模拟能耗数据库查询工具：查询指定设备在时间范围内的能耗数据"""
    print(f"查询设备{equipment_id} {start_date}至{end_date}的能耗数据...")
    # 模拟数据库查询结果
    return {
        "equipment_id": equipment_id,
        "start_date": start_date,
        "end_date": end_date,
        "energy_data": [120.5, 118.3, 122.1, 119.8],  # 单位：kWh
        "status": "success"
    }

def call_erp_api(file_path: str, erp_url: str) -> Dict[str, Any]:
    """模拟ERP系统API调用工具：同步文件至ERP系统"""
    print(f"同步文件{file_path}至ERP系统{erp_url}...")
    # 模拟API调用延迟与结果
    time.sleep(2)
    return {"status": "success", "message": "文件同步完成", "erp_task_id": "ERP-20260120-001"}

def rpa_execute(report_path: str) -> Dict[str, Any]:
    """模拟RPA工具：生成标准化Excel报表"""
    print(f"通过RPA生成报表：{report_path}...")
    return {"status": "success", "report_path": report_path, "file_size": "2.5MB"}

# ------------------------------
# 2. 工具注册与调度核心逻辑
# ------------------------------
class ToolDispatcher:
    def __init__(self):
        self.tool_registry: Dict[str, Dict] = {}  # 工具注册库
    
    def register_tool(self, tool_name: str, tool_func, desc: str, params: List[str]):
        """
        注册工具至工具库
        :param tool_name: 工具名称（唯一标识）
        :param tool_func: 工具函数对象
        :param desc: 工具功能描述（用于任务匹配）
        :param params: 工具所需参数列表
        """
        self.tool_registry[tool_name] = {
            "func": tool_func,
            "desc": desc,
            "params": params
        }
        print(f"工具[{tool_name}]注册成功")
    
    def dispatch_tool(self, tool_name: str, args: Dict[str, Any], retry=2) -> Optional[Dict[str, Any]]:
        """
        工具调度执行函数，支持重试机制
        :param tool_name: 工具名称
        :param args: 工具执行参数（键值对）
        :param retry: 重试次数
        :return: 工具执行结果，失败返回None
        """
        # 检查工具是否已注册
        if tool_name not in self.tool_registry:
            print(f"错误：工具[{tool_name}]未注册")
            return None
        
        tool_info = self.tool_registry[tool_name]
        # 检查参数是否完整
        missing_params = [p for p in tool_info["params"] if p not in args]
        if missing_params:
            print(f"错误：工具[{tool_name}]缺少必要参数：{missing_params}")
            return None
        
        # 执行工具并处理重试
        for attempt in range(retry + 1):
            try:
                result = tool_info["func"](**args)
                return result
            except Exception as e:
                print(f"工具[{tool_name}]第{attempt+1}次执行失败，错误信息：{str(e)}")
                if attempt == retry:
                    print(f"工具[{tool_name}]执行失败，已达到最大重试次数")
                    return None
                time.sleep(1)  # 重试间隔1秒

# ------------------------------
# 测试示例
# ------------------------------
if __name__ == "__main__":
    # 初始化工具调度器
    dispatcher = ToolDispatcher()
    
    # 注册工具
    dispatcher.register_tool(
        tool_name="query_energy_db",
        tool_func=query_energy_db,
        desc="查询设备能耗数据库，获取指定时间范围的能耗数据",
        params=["start_date", "end_date", "equipment_id"]
    )
    dispatcher.register_tool(
        tool_name="call_erp_api",
        tool_func=call_erp_api,
        desc="调用ERP系统API，同步文件至ERP平台",
        params=["file_path", "erp_url"]
    )
    dispatcher.register_tool(
        tool_name="rpa_execute",
        tool_func=rpa_execute,
        desc="通过RPA生成标准化Excel报表",
        params=["report_path"]
    )
    
    # 调度工具执行子任务
    task_args = {"start_date": "2026-01-01", "end_date": "2026-01-30", "equipment_id": "EQ-001"}
    result = dispatcher.dispatch_tool("query_energy_db", task_args)
    print("工具执行结果：", result)

from typing import List, Dict
import json

# 模拟执行日志（实际落地中可从日志系统读取）
execution_logs = [
    {
        "main_task": "生成车间月度能耗分析报告并同步至ERP系统",
        "subtasks": ["查询能耗数据", "计算波动值", "生成报表", "同步ERP", "发送通知"],
        "tool_calls": [
            {"tool_name": "query_energy_db", "args": {"start_date": "2026-01-01", "end_date": "2026-01-30", "equipment_id": "EQ-001"}, "status": "success"},
            {"tool_name": "call_erp_api", "args": {"file_path": "/report/energy.xlsx", "erp_url": "https://erp.example.com/upload"}, "status": "fail", "error": "API超时"}
        ],
        "final_status": "partial_success",
        "feedback": "ERP API超时，需切换备用RPA工具同步"
    }
]

def analyze_logs(logs: List[Dict]) -> Dict[str, Any]:
    """
    日志分析与错误归因
    :param logs: 执行日志列表
    :return: 归因结果与优化建议
    """
    optimization_suggestions = {
        "task_decompose": [],
        "tool_selection": [],
        "error_handling": []
    }
    
    for log in logs:
        if log["final_status"] in ["fail", "partial_success"]:
            # 分析工具调用错误
            for tool_call in log["tool_calls"]:
                if tool_call["status"] == "fail":
                    if tool_call["error"] == "API超时":
                        # 归因：工具选择单一，无备用方案
                        optimization_suggestions["tool_selection"].append(
                            f"工具[{tool_call['tool_name']}]调用超时，建议增加RPA作为备用工具，优先级：API> RPA"
                        )
                        # 建议补充异常处理规则
                        optimization_suggestions["error_handling"].append(
                            f"新增规则：{tool_call['tool_name']}调用超时（超过3秒），自动切换至RPA工具执行"
                        )
    
    return optimization_suggestions

def update_strategies(suggestions: Dict[str, Any], tool_dispatcher, decompose_rules):
    """
    基于优化建议更新系统策略
    :param suggestions: 优化建议
    :param tool_dispatcher: 工具调度器实例
    :param decompose_rules: 任务拆解规则库
    """
    # 更新工具选择优先级（示例：为ERP同步任务增加备用工具）
    for suggestion in suggestions["tool_selection"]:
        if "call_erp_api" in suggestion and "RPA" in suggestion:
            # 为ERP同步子任务添加备用工具配置
            print("更新工具选择策略：ERP同步任务增加RPA作为备用工具")
            # 实际落地中可修改工具调度器的优先级配置
            tool_dispatcher.tool_registry["call_erp_api"]["backup_tool"] = "rpa_execute"
    
    # 更新异常处理规则
    for suggestion in suggestions["error_handling"]:
        print(f"新增异常处理规则：{suggestion}")
        # 实际落地中可将规则写入配置文件，供工具调度器读取
    
    return tool_dispatcher, decompose_rules

# 执行日志分析与策略更新
if __name__ == "__main__":
    # 模拟任务拆解规则库
    decompose_rules = {"生成车间月度能耗分析报告并同步至ERP系统": [...]}\
    
    # 初始化工具调度器（复用前文定义的ToolDispatcher）
    from tool_dispatcher import ToolDispatcher, call_erp_api, rpa_execute
    dispatcher = ToolDispatcher()
    dispatcher.register_tool("call_erp_api", call_erp_api, "同步文件至ERP", ["file_path", "erp_url"])
    dispatcher.register_tool("rpa_execute", rpa_execute, "生成报表并同步", ["report_path"])
    
    # 分析日志并获取建议
    suggestions = analyze_logs(execution_logs)
    print("优化建议：", json.dumps(suggestions, indent=2, ensure_ascii=False))
    
    # 更新策略
    updated_dispatcher, updated_rules = update_strategies(suggestions, dispatcher, decompose_rules)
    print("策略更新完成")