# 大模型Agent通用基础框架
import json
from typing import List, Dict
# 大模型调用接口
def llm_invoke(prompt: str) -> str:
    return f"【大模型推理结果】基于输入指令完成任务解析与规划：{prompt[:30]}..."
class BaseAgent:
    """Agent基础父类，封装通用能力"""
    def __init__(self):
        self.short_memory: List[Dict] = []  # 短期记忆：当前任务上下文
        self.long_memory: List[Dict] = []   # 长期记忆：历史任务数据
        self.task_status = "idle"           # 任务状态：idle/working/finish/error
    def parse_task(self, user_input: str) -> Dict:
        """1. 任务解析：识别需求、场景、任务类型"""
        self.task_status = "working"
        prompt = f"请解析用户任务，输出JSON格式：任务目标、场景类型、任务复杂度、所需工具\n用户指令：{user_input}"
        res = llm_invoke(prompt)
        self.short_memory.append({"type": "parse", "content": user_input, "result": res})
        return {"task_target": user_input, "scene_type": "通用场景", "complex": "普通"}
    def plan_task(self, parse_result: Dict) -> List[str]:
        """2. 任务规划：拆解子任务"""
        prompt = f"基于任务解析结果，拆解为可执行子任务，分点输出：{json.dumps(parse_result)}"
        res = llm_invoke(prompt)
        sub_tasks = [f"子任务{i+1}" for i in range(3)]
        self.short_memory.append({"type": "plan", "content": "任务拆解", "result": sub_tasks})
        return sub_tasks
    def execute_task(self, sub_tasks: List[str]) -> List[str]:
        """3. 任务执行：逐个子任务落地"""
        execute_results = []
        for task in sub_tasks:
            res = llm_invoke(f"执行子任务：{task}")
            execute_results.append(res)
        self.short_memory.append({"type": "execute", "content": "子任务执行", "result": execute_results})
        return execute_results
    def check_result(self, execute_results: List[str]) -> bool:
        """4. 结果校验纠错"""
        prompt = f"校验任务执行结果是否合格：{json.dumps(execute_results)}"
        res = llm_invoke(prompt)
        return True
    def run(self, user_input: str) -> str:
        """Agent全流程运行入口"""
        # 步骤1：解析任务
        parse_res = self.parse_task(user_input)
        # 步骤2：规划任务
        plan_res = self.plan_task(parse_res)
        # 步骤3：执行任务
        exec_res = self.execute_task(plan_res)
        # 步骤4：校验结果
        check_res = self.check_result(exec_res)
        # 步骤5：整合输出
        self.task_status = "finish"
        self.long_memory.append(self.short_memory.copy())
        return f"任务执行完成！执行结果：{exec_res}"
# 测试通用Agent
if __name__ == "__main__":
    agent = BaseAgent()
    result = agent.run("帮我整理本周工作周报，汇总核心工作、问题与下周计划")
    print(result)

# 轻量交互场景Agent（问答/文案/润色专属）
from typing import Dict
from base_agent import BaseAgent
def llm_light_invoke(prompt: str) -> str:
    """轻量场景专属大模型调用，精简推理逻辑，提升响应速度"""
    return f"【轻量交互结果】已完成需求响应：{prompt[:40]}..."
class LightInteractiveAgent(BaseAgent):
    """轻量交互专属Agent，精简执行链路，适配简单问答、文案场景"""
    def parse_task(self, user_input: str) -> Dict:
        """重写任务解析：快速判定轻量场景，精简解析逻辑"""
        self.task_status = "working"
        # 轻量场景固定解析规则，无需复杂推理
        parse_result = {
            "task_target": user_input,
            "scene_type": "轻量交互",
            "complex": "简单",
            "need_tool": False,
            "need_iter": False
        }
        self.short_memory.append({"type": "parse", "result": parse_result})
        return parse_result
    def plan_task(self, parse_result: Dict):
        """重写任务规划：轻量任务无需拆解子任务，直接返回单任务"""
        return [parse_result["task_target"]]
    def check_result(self, execute_results) -> bool:
        """重写结果校验：仅合规性基础校验"""
        if len(execute_results) == 0:
            return False
        return True
    def run(self, user_input: str) -> str:
        """轻量化执行流程，缩减冗余步骤"""
        parse_res = self.parse_task(user_input)
        task = self.plan_task(parse_res)[0]
        exec_res = llm_light_invoke(task)
        self.check_result([exec_res])
        self.task_status = "finish"
        return f"✅ 轻量任务完成\n输出结果：{exec_res}"
# 场景测试
if __name__ == "__main__":
    agent = LightInteractiveAgent()
    # 测试文案润色场景
    print(agent.run("帮我润色这句话：今天工作很忙，收获很多，学到了很多知识"))
    # 测试知识问答场景
    print(agent.run("简单解释一下大模型Agent的核心作用"))

# 任务自动化场景Agent（周报自动生成专属）
from typing import List, Dict
from base_agent import BaseAgent
# 模拟工作数据接口（实际对接企业OA/日志系统）
def get_weekly_work_data() -> Dict:
    """获取本周工作原始数据"""
    return {
        "daily_work": ["完成项目需求开发", "参与需求评审会议", "修复线上bug", "整理技术文档"],
        "problems": ["部分需求变更频繁", "线上偶发性能问题"],
        "achievement": "完成核心模块迭代，系统稳定性提升20%",
        "next_plan": ["优化系统性能", "完成剩余需求开发", "学习新技术框架"]
    }
def llm_auto_invoke(prompt: str) -> str:
    """自动化场景大模型调用，固定输出风格"""
    return f"【自动化生成结果】{prompt[:50]}：内容规整，符合周报输出标准"
class AutoTaskAgent(BaseAgent):
    """任务自动化专属Agent，适配固定流程批量任务"""
    def parse_task(self, user_input: str) -> Dict:
        """解析自动化任务，匹配业务模板"""
        self.task_status = "working"
        parse_res = {
            "task_target": user_input,
            "scene_type": "任务自动化",
            "complex": "中等",
            "need_tool": True,
            "tool_list": ["数据读取工具", "内容整理工具"],
            "need_iter": True
        }
        self.short_memory.append({"type": "parse", "result": parse_res})
        return parse_res
    def plan_task(self, parse_result: Dict) -> List[str]:
        """固定流程任务拆解，周报生成专属子任务"""
        return [
            "读取本周工作原始数据",
            "清洗筛选有效工作内容",
            "梳理工作成果与现存问题",
            "规划下周工作计划",
            "按照标准周报模板整合内容"
        ]
    def execute_task(self, sub_tasks: List[str]) -> List[str]:
        """按固定流程执行自动化任务"""
        results = []
        # 步骤1：读取数据
        work_data = get_weekly_work_data()
        results.append(f"数据读取完成：{work_data}")
        # 步骤2-5：大模型整合生成周报
        prompt = f"基于以下数据生成标准工作周报：{work_data}，分工作内容、工作成果、存在问题、下周计划四部分输出"
        results.append(llm_auto_invoke(prompt))
        self.short_memory.append({"type": "execute", "result": results})
        return results
    def check_result(self, execute_results: List[str]) -> bool:
        """自动化任务规则化校验"""
        key_check = ["工作内容", "工作成果", "存在问题", "下周计划"]
        result_str = str(execute_results)
        # 校验核心模块是否齐全
        for key in key_check:
            if key not in result_str:
                return False
        return True
    def run(self, user_input: str) -> str:
        """自动化任务专属执行链路"""
        parse_res = self.parse_task(user_input)
        plan_tasks = self.plan_task(parse_res)
        exec_res = self.execute_task(plan_tasks)
        # 校验失败则重试一次
        if not self.check_result(exec_res):
            exec_res = self.execute_task(plan_tasks)
        self.task_status = "finish"
        return f"✅ 自动化周报生成完成\n详细内容：{exec_res[-1]}"
# 场景测试
if __name__ == "__main__":
    agent = AutoTaskAgent()
    print(agent.run("自动生成本周工作周报"))

# 复杂推理场景Agent（迭代推理+回溯纠错专属）
from typing import List, Dict
from base_agent import BaseAgent
def llm_reason_invoke(prompt: str) -> str:
    """复杂推理专属大模型调用，输出分步推理逻辑"""
    return f"【分步推理】{prompt[:60]}：已完成单步逻辑推导，结论合理"
class ReasonAgent(BaseAgent):
    """复杂推理专属Agent，支持动态规划、迭代推理、回溯纠错"""
    def __init__(self):
        super().__init__()
        self.reason_history = []  # 推理历史记录，用于回溯
        self.max_iter = 3         # 最大迭代次数，防止死循环
    def parse_task(self, user_input: str) -> Dict:
        """深度解析复杂推理任务"""
        self.task_status = "working"
        parse_res = {
            "task_target": user_input,
            "scene_type": "复杂推理",
            "complex": "高",
            "need_tool": True,
            "need_iter": True,
            "max_step": 5
        }
        self.short_memory.append({"type": "parse", "result": parse_res})
        return parse_res
    def plan_dynamic_task(self, last_result: str, iter_num: int) -> str:
        """动态规划下一步推理任务，基于上一轮结果生成"""
        if iter_num >= self.max_iter:
            return "结束推理，整合最终结论"
        prompt = f"基于上一轮推理结果：{last_result}，生成下一步推理子任务"
        return llm_reason_invoke(prompt)
    def execute_reason(self, task: str) -> str:
        """单步推理执行"""
        res = llm_reason_invoke(f"执行推理任务：{task}")
        self.reason_history.append(res)
        return res
    def check_reason_step(self, step_result: str) -> bool:
        """单步推理逻辑校验"""
        # 简单规则校验，实际可扩展逻辑校验、数据校验
        if "错误" in step_result or "逻辑漏洞" in step_result:
            return False
        return True
    def backtrack(self):
        """推理回溯：删除错误步骤，返回上一步结果"""
        if len(self.reason_history) > 0:
            self.reason_history.pop()
        return self.reason_history[-1] if self.reason_history else ""
    def run(self, user_input: str) -> str:
        """复杂推理迭代执行链路"""
        self.reason_history = []
        self.parse_task(user_input)
        iter_num = 0
        current_result = ""
        # 多轮迭代推理
        while iter_num < self.max_iter:
            # 动态生成子任务
            next_task = self.plan_dynamic_task(current_result, iter_num)
            # 执行单步推理
            step_res = self.execute_reason(next_task)
            # 校验单步结果
            if not self.check_reason_step(step_res):
                current_result = self.backtrack()
                continue
            current_result = step_res
            iter_num += 1
        self.task_status = "finish"
        return f"✅ 复杂推理完成\n推理迭代次数：{iter_num}\n完整推理链路：{self.reason_history}\n最终结论：{current_result}"
# 场景测试
if __name__ == "__main__":
    agent = ReasonAgent()
    # 测试复杂分析推理场景
    print(agent.run("分析项目线上性能卡顿问题，推导核心成因并给出优化方案"))

# 工具协同场景Agent（多工具联动专属）
from typing import List, Dict
from base_agent import BaseAgent
# 模拟外部工具能力
class SearchTool:
    """联网搜索工具"""
    def search(self, keyword: str) -> str:
        return f"【搜索结果】关键词{keyword}：获取最新行业数据、政策信息、实时资讯"
class CalcTool:
    """数据计算工具"""
    def calc(self, formula: str) -> str:
        return f"【计算结果】公式{formula}：计算完成，数据准确无误"
class FileTool:
    """文件处理工具"""
    def save_file(self, content: str, file_name: str) -> str:
        return f"【文件操作】已将内容保存至{file_name}"
def llm_tool_invoke(prompt: str) -> str:
    """工具协同场景大模型推理"""
    return f"【工具协同推理】{prompt[:50]}：完成数据整合分析"
class ToolCooperateAgent(BaseAgent):
    """多工具协同专属Agent，支持多工具联动迭代"""
    def __init__(self):
        super().__init__()
        # 注册工具
        self.tools = {
            "search": SearchTool(),
            "calc": CalcTool(),
            "file": FileTool()
        }
        self.tool_call_log = []  # 工具调用日志
    def parse_task(self, user_input: str) -> Dict:
        """解析任务，匹配所需工具"""
        self.task_status = "working"
        parse_res = {
            "task_target": user_input,
            "scene_type": "工具协同",
            "complex": "高",
            "need_tool": True,
            "tool_list": ["search", "calc", "file"],
            "need_iter": True
        }
        self.short_memory.append({"type": "parse", "result": parse_res})
        return parse_res
    def plan_tool_task(self, parse_res: Dict) -> List[Dict]:
        """规划工具调用顺序与参数"""
        return [
            {"tool": "search", "param": "2026年大模型行业发展数据"},
            {"tool": "calc", "param": "行业增长率、市场规模核算"},
            {"tool": "file", "param": "行业分析报告保存"}
        ]
    def execute_tool_task(self, tool_tasks: List[Dict]) -> List[str]:
        """执行多工具协同任务"""
        tool_results = []
        for task in tool_tasks:
            tool_name = task["tool"]
            param = task["param"]
            # 调用对应工具
            if tool_name == "search":
                res = self.tools["search"].search(param)
            elif tool_name == "calc":
                res = self.tools["calc"].calc(param)
            elif tool_name == "file":
                res = self.tools["file"].save_file("行业分析内容", "2026大模型行业报告.md")
            else:
                res = "工具不存在"
            tool_results.append(res)
            self.tool_call_log.append({tool_name: param})
        # 工具结果回流推理
        llm_res = llm_tool_invoke(f"基于工具结果{tool_results}，生成完整行业分析报告")
        tool_results.append(llm_res)
        return tool_results
    def check_result(self, execute_results: List[str]) -> bool:
        """校验工具协同执行结果"""
        if len(execute_results) < 3:
            return False
        return True
    def run(self, user_input: str) -> str:
        """工具协同全流程执行"""
        parse_res = self.parse_task(user_input)
        tool_tasks = self.plan_tool_task(parse_res)
        exec_res = self.execute_tool_task(tool_tasks)
        # 失败重试
        if not self.check_result(exec_res):
            exec_res = self.execute_tool_task(tool_tasks)
        self.task_status = "finish"
        return f"✅ 多工具协同任务完成\n工具调用日志：{self.tool_call_log}\n最终输出：{exec_res[-1]}"
# 场景测试
if __name__ == "__main__":
    agent = ToolCooperateAgent()
    print(agent.run("调研2026年大模型行业数据，核算市场规模，生成并保存行业分析报告"))