08. 新一代AI Agent：从工具使用到自主执行¶

class ReActAgent:
    """
    ReAct: Synergizing Reasoning and Acting in Language Models

    核心思想：将推理(Reasoning)和行动(Acting)交织进行
    """
    def __init__(self, llm, tools, max_iterations=10):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        self.max_iterations = max_iterations
        self.memory = []

    def run(self, query: str) -> str:
        """执行ReAct循环"""
        self.memory.append(f"Task: {query}")

        for i in range(self.max_iterations):
            # 1. 思考 (Thought)
            thought = self.think()

            # 2. 行动 (Action)
            action = self.act(thought)

            # 3. 观察 (Observation)
            observation = self.execute_action(action)

            # 4. 更新记忆
            self.memory.extend([
                f"Thought {i+1}: {thought}",
                f"Action {i+1}: {action}",
                f"Observation {i+1}: {observation}"
            ])

            # 5. 检查是否完成
            if self.is_complete(observation):
                return self.generate_answer()

        return "Max iterations reached"

    def think(self) -> str:
        """生成思考过程"""
        prompt = self._build_react_prompt()
        response = self.llm.generate(prompt)
        return self._parse_thought(response)

    def act(self, thought: str) -> Dict:
        """根据思考选择行动"""
        prompt = f"""Based on the thought: {thought}

Available tools: {list(self.tools.keys())}

Choose the next action in format:
Action: [tool_name]
Action Input: [input]"""

        response = self.llm.generate(prompt)
        return self._parse_action(response)

    def execute_action(self, action: Dict) -> str:
        """执行选定的工具"""
        tool_name = action['tool']
        tool_input = action['input']

        if tool_name not in self.tools:
            return f"Error: Tool {tool_name} not found"

        try:  # try/except捕获异常，防止程序崩溃
            result = self.tools[tool_name].run(tool_input)
            return str(result)
        except Exception as e:
            return f"Error: {str(e)}"

class ManusAgent:
    """
    Manus通用AI Agent架构（基于公开信息推断）

    核心组件：
    1. 任务规划器 (Task Planner)
    2. 执行引擎 (Execution Engine)
    3. 工具集成层 (Tool Integration)
    4. 记忆系统 (Memory System)
    5. 验证器 (Validator)
    """

    def __init__(self):
        self.planner = TaskPlanner(llm="claude-3-5-sonnet")
        self.executor = ExecutionEngine()
        self.tool_registry = ToolRegistry()
        self.memory = HierarchicalMemory()
        self.validator = OutputValidator()

    def execute_task(self, user_request: str) -> TaskResult:
        """
        端到端任务执行流程

        Args:
            user_request: 用户的自然语言请求

        Returns:
            TaskResult: 包含执行结果、中间步骤、日志等
        """
        # Phase 1: 任务理解与规划
        task_plan = self.planner.create_plan(user_request)

        # Phase 2: 执行循环
        execution_log = []
        for step in task_plan.steps:
            # 执行步骤
            result = self._execute_step(step)
            execution_log.append(result)

            # 动态重规划（如果需要）
            if result.needs_replanning:
                task_plan = self.planner.replan(task_plan, execution_log)

            # 更新记忆
            self.memory.add_execution_step(step, result)

        # Phase 3: 结果验证与交付
        final_output = self._assemble_output(execution_log)
        validation = self.validator.validate(final_output, user_request)

        return TaskResult(
            output=final_output,
            execution_log=execution_log,
            validation_score=validation.score,
            artifacts=self._collect_artifacts()
        )

    def _execute_step(self, step: ExecutionStep) -> StepResult:
        """执行单个步骤"""
        if step.type == "web_browsing":
            return self._browse_web(step.params)
        elif step.type == "code_execution":
            return self._execute_code(step.params)
        elif step.type == "file_operation":
            return self._operate_file(step.params)
        elif step.type == "api_call":
            return self._call_api(step.params)
        elif step.type == "human_confirmation":
            return self._request_confirmation(step.params)
        else:
            raise ValueError(f"Unknown step type: {step.type}")

class TaskPlanner:
    """任务规划器：将复杂请求分解为可执行步骤"""

    def create_plan(self, request: str) -> TaskPlan:
        """
        创建任务执行计划

        使用Chain-of-Thought进行任务分解
        """
        planning_prompt = f"""You are an expert task planner. Break down the following request into clear, executable steps.

Request: {request}

For each step, specify:
1. Step type (web_browsing, code_execution, file_operation, api_call, human_confirmation)
2. Required parameters
3. Expected output
4. Dependencies on previous steps

Output the plan in structured JSON format."""

        plan_json = self.llm.generate(planning_prompt, format="json")
        return TaskPlan.from_json(plan_json)

    def replan(self, current_plan: TaskPlan, execution_log: List[StepResult]) -> TaskPlan:
        """根据执行反馈动态调整计划"""
        context = self._build_replanning_context(current_plan, execution_log)

        replan_prompt = f"""The current plan needs adjustment based on execution results.

Current Plan: {current_plan.to_json()}
Execution Log: {execution_log}

Please provide an updated plan that:
1. Addresses the issues encountered
2. Incorporates new information discovered
3. Optimizes remaining steps

Output the revised plan."""

        new_plan_json = self.llm.generate(replan_prompt, format="json")
        return TaskPlan.from_json(new_plan_json)

class HierarchicalMemory:
    """分层记忆系统"""

    def __init__(self):
        self.working_memory = []  # 当前任务上下文
        self.episodic_memory = []  # 历史任务经验
        self.semantic_memory = {}  # 知识库

    def add_execution_step(self, step: ExecutionStep, result: StepResult):
        """添加执行步骤到工作记忆"""
        self.working_memory.append({
            "step": step,
            "result": result,
            "timestamp": time.time()
        })

    def get_relevant_context(self, current_step: ExecutionStep) -> str:
        """检索与当前步骤相关的上下文"""
        # 检索工作记忆中的相关步骤
        relevant_working = self._retrieve_from_working_memory(current_step)

        # 检索历史经验中的相似案例
        similar_episodes = self._retrieve_from_episodic_memory(current_step)

        # 检索相关知识
        relevant_knowledge = self._retrieve_from_semantic_memory(current_step)

        return self._format_context(relevant_working, similar_episodes, relevant_knowledge)

# 示例：Manus执行复杂任务的流程

# 场景1：股票分析报告生成
def stock_analysis_example():
    """
    用户请求："分析特斯拉股票过去一个月的表现，
              生成一份包含技术面和基本面的投资报告"
    """

    manus = ManusAgent()

    result = manus.execute_task("""
    分析特斯拉(TSLA)股票过去一个月的表现，包括：
    1. 股价走势和技术指标分析
    2. 近期新闻和事件影响
    3. 财务数据对比
    4. 生成PDF格式的投资分析报告
    """)

    # Manus的执行流程：
    # 1. 搜索特斯拉股价数据
    # 2. 计算技术指标(MA, RSI, MACD等)
    # 3. 搜索相关新闻
    # 4. 查询财务报表
    # 5. 编写Python代码生成可视化图表
    # 6. 撰写分析报告
    # 7. 转换为PDF格式
    # 8. 交付最终报告

    return result.artifacts['report.pdf']

# 场景2：网站搭建
def website_building_example():
    """
    用户请求："为我创建一个个人博客网站"
    """

    manus = ManusAgent()

    result = manus.execute_task("""
    创建一个现代化的个人博客网站，要求：
    1. 响应式设计
    2. 支持Markdown文章
    3. 包含关于页面和联系表单
    4. 部署到Vercel
    """)

    # Manus的执行流程：
    # 1. 设计网站架构
    # 2. 创建Next.js项目
    # 3. 编写组件代码
    # 4. 配置样式和主题
    # 5. 实现Markdown渲染
    # 6. 添加联系表单功能
    # 7. 测试和调试
    # 8. 部署到Vercel
    # 9. 提供访问链接

    return result.artifacts['deployment_url']

# 场景3：数据处理流水线
def data_pipeline_example():
    """
    用户请求："从多个数据源整合销售数据并生成仪表板"
    """

    manus = ManusAgent()

    result = manus.execute_task("""
    构建销售数据整合流水线：
    1. 从CSV文件读取线下销售数据
    2. 通过API获取线上销售数据
    3. 清洗和标准化数据格式
    4. 计算关键指标(销售额、增长率等)
    5. 创建交互式仪表板
    """)

    return result.artifacts['dashboard.html']

class ClaudeCode:
    """
    Claude Code核心架构

    特点：
    1. 深度代码理解
    2. 安全的代码执行环境
    3. 版本控制集成
    4. 上下文感知编辑
    """

    def __init__(self, project_path: str):
        self.project = ProjectContext(project_path)
        self.code_index = CodeIndex(project_path)
        self.tool_executor = SecureToolExecutor()
        self.git_integration = GitIntegration(project_path)

    def process_request(self, user_input: str) -> CodeAction:
        """处理用户编程请求"""

        # 1. 理解意图
        intent = self._understand_intent(user_input)

        # 2. 收集上下文
        context = self._gather_context(intent)

        # 3. 规划行动
        plan = self._plan_actions(intent, context)

        # 4. 执行并验证
        results = []
        for action in plan:
            result = self._execute_action(action)
            results.append(result)

            # 实时反馈
            if action.requires_confirmation:
                self._show_diff_and_confirm(result)

        return self._summarize_results(results)

    def _gather_context(self, intent: Intent) -> Context:
        """智能上下文收集"""
        context = Context()

        # 检索相关文件
        if intent.target_files:
            for file_path in intent.target_files:
                context.add_file(self.project.read_file(file_path))
        else:
            # 语义搜索相关代码
            relevant_files = self.code_index.semantic_search(intent.description)
            for file in relevant_files[:5]:  # Top-5相关文件
                context.add_file(file)

        # 获取项目结构
        context.project_structure = self.project.get_structure()

        # 获取依赖信息
        context.dependencies = self.project.get_dependencies()

        # 获取Git状态
        context.git_status = self.git_integration.get_status()

        return context

    def _plan_actions(self, intent: Intent, context: Context) -> List[CodeAction]:
        """规划代码修改行动"""

        planning_prompt = f"""You are an expert software engineer.

Task: {intent.description}

Context:
{context.to_prompt()}

Plan the necessary code changes. For each change, specify:
1. File path
2. Change type (create, modify, delete, rename)
3. Detailed description of changes
4. Dependencies on other changes

Output as a structured action plan."""

        plan_response = self.llm.generate(planning_prompt, format="json")
        return self._parse_action_plan(plan_response)

class CodeIndex:
    """代码索引系统：支持语义搜索"""

    def __init__(self, project_path: str):
        self.project_path = project_path
        self.index = self._build_index()

    def _build_index(self) -> VectorStore:
        """构建代码向量索引"""
        code_chunks = []

        for file_path in self._get_code_files():
            chunks = self._parse_and_chunk(file_path)
            for chunk in chunks:
                embedding = self.embedder.encode(chunk.content)
                code_chunks.append({
                    "embedding": embedding,
                    "content": chunk.content,
                    "metadata": {
                        "file": file_path,
                        "line_start": chunk.line_start,
                        "line_end": chunk.line_end,
                        "type": chunk.type  # function, class, etc.
                    }
                })

        return VectorStore.from_documents(code_chunks)

    def semantic_search(self, query: str, top_k: int = 10) -> List[CodeChunk]:
        """语义搜索相关代码"""
        query_embedding = self.embedder.encode(query)
        results = self.index.similarity_search(query_embedding, k=top_k)
        return [CodeChunk.from_result(r) for r in results]

class ClaudeCowork:
    """
    Claude Cowork: 通用AI助手

    目标用户：非技术人员
    核心场景：日常办公、研究、内容创作
    """

    def __init__(self):
        self.browser = BrowserController()
        self.file_manager = FileManager()
        self.document_processor = DocumentProcessor()
        self.calendar = CalendarIntegration()
        self.email = EmailIntegration()

    def assist(self, request: UserRequest) -> AssistanceResult:
        """处理用户协助请求"""

        # 分类请求类型
        request_type = self._classify_request(request)

        if request_type == "research":
            return self._handle_research(request)
        elif request_type == "document_creation":
            return self._handle_document_creation(request)
        elif request_type == "data_analysis":
            return self._handle_data_analysis(request)
        elif request_type == "scheduling":
            return self._handle_scheduling(request)
        elif request_type == "communication":
            return self._handle_communication(request)
        else:
            return self._handle_general(request)

    def _handle_research(self, request: UserRequest) -> ResearchResult:
        """处理研究类请求"""

        # 1. 理解研究主题
        topic = request.extract_topic()

        # 2. 多源信息收集
        sources = []

        # 网页搜索
        search_results = self.browser.search(topic)
        for result in search_results[:10]:
            page_content = self.browser.read_page(result.url)
            sources.append({
                "url": result.url,
                "title": result.title,
                "content": page_content
            })

        # 3. 信息整合与分析
        analysis = self._synthesize_information(sources, topic)

        # 4. 生成报告
        report = self._generate_research_report(analysis)

        return ResearchResult(
            summary=analysis.summary,
            key_findings=analysis.findings,
            sources=sources,
            report=report
        )

    def _handle_document_creation(self, request: UserRequest) -> DocumentResult:
        """处理文档创作请求"""

        # 理解文档需求
        doc_spec = self._parse_document_requirements(request)

        # 收集参考材料
        references = []
        if request.has_attachments:
            for attachment in request.attachments:
                content = self.document_processor.read(attachment)
                references.append(content)

        # 生成文档大纲
        outline = self._generate_outline(doc_spec, references)

        # 逐段生成内容
        sections = []
        for section in outline.sections:
            content = self._write_section(section, references, sections)
            sections.append({
                "heading": section.heading,
                "content": content
            })

        # 格式化和导出
        document = self._format_document(sections, doc_spec.format)

        return DocumentResult(
            document=document,
            outline=outline,
            word_count=sum(len(s["content"]) for s in sections)
        )

class CodeAgentChallenges:
    """代码Agent面临的关键挑战与解决方案"""

    @staticmethod  # @staticmethod无需实例即可调用
    def challenge_1_context_window():
        """
        挑战1: 大型代码库的上下文限制

        解决方案:
        1. 智能代码检索
        2. 分层摘要
        3. 增量索引
        """

        # 智能检索示例
        def intelligent_retrieval(project, query):
            # 第一层：文件级检索
            relevant_files = file_level_search(project, query)

            # 第二层：函数/类级检索
            code_chunks = []
            for file in relevant_files:
                chunks = chunk_code(file)
                scored_chunks = rank_by_relevance(chunks, query)
                code_chunks.extend(scored_chunks[:3])

            # 第三层：关系扩展
            extended_context = expand_by_relations(code_chunks)

            return extended_context

    @staticmethod
    def challenge_2_code_execution_safety():
        """
        挑战2: 代码执行安全性

        解决方案:
        1. 沙箱环境
        2. 资源限制
        3. 权限控制
        """

        class SecureExecutor:
            def __init__(self):
                self.sandbox = DockerSandbox()
                self.resource_limits = {
                    "cpu": "1 core",
                    "memory": "512MB",
                    "timeout": 30,
                    "network": False
                }

            def execute(self, code: str) -> ExecutionResult:
                # 代码静态分析
                if self._contains_dangerous_operations(code):
                    return ExecutionResult.error("Dangerous operations detected")

                # 在沙箱中执行
                return self.sandbox.run(code, limits=self.resource_limits)

    @staticmethod
    def challenge_3_long_horizon_planning():
        """
        挑战3: 长程规划与执行

        解决方案:
        1. 分层规划
        2. 里程碑检查
        3. 动态调整
        """

        class HierarchicalPlanner:
            def plan(self, goal: str) -> Plan:
                # 高层规划
                high_level = self._create_high_level_plan(goal)

                # 逐层细化
                detailed_plan = Plan()
                for phase in high_level.phases:
                    sub_tasks = self._decompose(phase)
                    detailed_plan.add_phase(phase, sub_tasks)

                return detailed_plan

            def execute_with_checkpoints(self, plan: Plan):
                for phase in plan.phases:
                    for task in phase.tasks:
                        result = self.execute_task(task)

                        # 检查点验证
                        if not self._validate_checkpoint(result):
                            self._handle_deviation(phase, result)

"""
Claude Skills 概念实现

Skill = 可复用的Agent工作流模板
每个Skill包含：触发条件、执行步骤、所需工具、上下文模板

⚠️ 注意：Skills 是 Anthropic 产品层面的特性，以下代码展示其核心设计理念，
非官方API。实际使用请参考 Anthropic 官方文档。
"""

from dataclasses import dataclass, field
from datetime import datetime
import json
from typing import Callable

@dataclass  # @dataclass自动生成__init__等方法
class Skill:
    """Agent技能定义"""
    name: str                         # 技能名称
    description: str                  # 技能描述（用于匹配）
    trigger_patterns: list[str]       # 触发条件（关键词/意图描述）
    steps: list[dict]                 # 执行步骤模板
    required_tools: list[str]         # 所需工具
    context_template: str             # 上下文模板（含占位符）
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    success_count: int = 0            # 成功执行次数
    avg_tokens: int = 0               # 平均Token消耗

@dataclass
class SkillExecutionResult:
    """技能执行结果"""
    skill_name: str
    success: bool
    output: str
    tokens_used: int
    steps_completed: int

class SkillRegistry:
    """技能注册与管理中心"""

    def __init__(self):
        self.skills: dict[str, Skill] = {}

    def register(self, skill: Skill):
        """注册一个技能"""
        self.skills[skill.name] = skill
        print(f"💾 技能已注册: {skill.name}")

    def match(self, user_request: str) -> Skill | None:
        """根据用户请求匹配最合适的技能

        简化实现：关键词匹配。实际产品会使用embedding相似度。
        """
        best_match = None
        best_score = 0

        for skill in self.skills.values():
            score = sum(
                1 for pattern in skill.trigger_patterns
                if pattern.lower() in user_request.lower()
            )
            if score > best_score:
                best_score = score
                best_match = skill

        return best_match if best_score > 0 else None

    def save(self, filepath: str):
        """持久化保存所有技能"""
        data = {}
        for name, skill in self.skills.items():
            data[name] = {
                "name": skill.name,
                "description": skill.description,
                "trigger_patterns": skill.trigger_patterns,
                "steps": skill.steps,
                "required_tools": skill.required_tools,
                "context_template": skill.context_template,
                "success_count": skill.success_count,
            }
        with open(filepath, "w", encoding="utf-8") as f:  # with自动管理文件关闭
            json.dump(data, f, ensure_ascii=False, indent=2)

class SkillableAgent:
    """支持Skills的Agent"""

    def __init__(self, name: str, skill_registry: SkillRegistry):
        self.name = name
        self.skills = skill_registry

    def run(self, user_request: str) -> str:
        """执行请求：优先匹配已有技能，否则从头推理"""

        # 1. 尝试匹配已有技能
        matched_skill = self.skills.match(user_request)

        if matched_skill:
            print(f"⚡ 匹配到技能: {matched_skill.name} (已成功{matched_skill.success_count}次)")
            return self._execute_skill(matched_skill, user_request)

        # 2. 无匹配 → 从头推理
        print(f"🧠 无匹配技能，从头推理...")
        result = self._reason_from_scratch(user_request)

        # 3. 成功后，询问是否保存为技能（可自动化）
        print(f"💡 提示: 此工作流可保存为技能以供复用")
        return result

    def _execute_skill(self, skill: Skill, user_request: str) -> str:
        """按照技能模板执行"""
        output_parts = []

        for i, step in enumerate(skill.steps):  # enumerate同时获取索引和元素
            action = step["action"]
            # 将模板中的占位符替换为实际输入
            if "{user_input}" in action:
                action = action.replace("{user_input}", user_request)

            print(f"  Step {i+1}: {step.get('description', action)}")
            # 实际执行（简化：这里直接输出步骤描述）
            output_parts.append(f"[{step.get('description', '')}] 完成")

        skill.success_count += 1
        return "\n".join(output_parts)

    def _reason_from_scratch(self, user_request: str) -> str:
        """从头推理（常规Agent流程）"""
        return f"[从头推理完成] {user_request}"

    def learn_skill(self, name: str, description: str, trigger_patterns: list[str],
                    steps: list[dict], required_tools: list[str] = None):
        """从成功执行中学习一个新技能"""
        skill = Skill(
            name=name,
            description=description,
            trigger_patterns=trigger_patterns,
            steps=steps,
            required_tools=required_tools or [],
            context_template="",
        )
        self.skills.register(skill)

# === 使用示例 ===

registry = SkillRegistry()

# 预定义一些技能
registry.register(Skill(
    name="代码审查",
    description="对代码进行全面的安全性、性能、可读性审查",
    trigger_patterns=["代码审查", "review", "检查代码", "代码质量"],
    steps=[
        {"description": "读取目标代码文件", "action": "read_file({user_input})"},
        {"description": "检查安全漏洞", "action": "analyze_security"},
        {"description": "检查性能问题", "action": "analyze_performance"},
        {"description": "检查代码风格", "action": "check_style"},
        {"description": "生成审查报告", "action": "generate_report"},
    ],
    required_tools=["read_file", "code_analyzer", "report_generator"],
))

registry.register(Skill(
    name="周报生成",
    description="根据本周工作记录自动生成周报",
    trigger_patterns=["周报", "weekly report", "本周总结"],
    steps=[
        {"description": "收集本周Git提交记录", "action": "git_log_this_week"},
        {"description": "收集本周会议记录", "action": "get_meeting_notes"},
        {"description": "收集本周任务完成情况", "action": "get_task_status"},
        {"description": "汇总并生成周报", "action": "generate_weekly_report"},
    ],
    required_tools=["git", "calendar", "task_tracker", "document_writer"],
))

# 使用Agent
agent = SkillableAgent("Claude", registry)

# 匹配到已有技能 → 直接执行
print("--- 请求1: 有匹配技能 ---")
result = agent.run("请帮我做一下这个PR的代码审查")
print(f"结果: {result}\n")

# 无匹配技能 → 从头推理
print("--- 请求2: 无匹配技能 ---")
result = agent.run("帮我分析这个数据集的异常值")
print(f"结果: {result}\n")

# 从成功执行中学习新技能
agent.learn_skill(
    name="异常值分析",
    description="分析数据集中的异常值并生成报告",
    trigger_patterns=["异常值", "outlier", "数据异常"],
    steps=[
        {"description": "加载数据集", "action": "load_dataset({user_input})"},
        {"description": "统计描述分析", "action": "descriptive_stats"},
        {"description": "异常值检测(IQR+Z-score)", "action": "detect_outliers"},
        {"description": "可视化异常分布", "action": "plot_outliers"},
        {"description": "生成分析报告", "action": "generate_report"},
    ],
)

# 再次请求 → 匹配到刚学习的技能
print("--- 请求3: 匹配到新学习的技能 ---")
result = agent.run("这批数据有很多异常值，帮我分析一下")
print(f"结果: {result}")

class OpenAIOperator:
    """
    OpenAI Operator架构

    核心能力：
    1. 视觉感知屏幕内容
    2. 理解GUI元素
    3. 执行鼠标/键盘操作
    4. 多步骤任务执行
    """

    def __init__(self):
        self.vision_model = GPT4Vision()
        self.action_model = ActionPredictionModel()
        self.browser = BrowserAutomation()
        self.os_interface = OSInterface()

    def execute_task(self, task_description: str) -> TaskResult:
        """
        执行需要操作计算机界面的任务

        示例任务：
        - "在Amazon上搜索蓝牙耳机并按评分排序"
        - "在Gmail中查找上周的会议邀请并添加到日历"
        """

        max_steps = 50
        step = 0
        state_history = []

        while step < max_steps:
            # 1. 截取当前屏幕
            screenshot = self.os_interface.capture_screen()

            # 2. 视觉理解
            ui_elements = self.vision_model.analyze_ui(screenshot)

            # 3. 决策下一步行动
            action = self.action_model.predict(
                task=task_description,
                current_state=screenshot,
                ui_elements=ui_elements,
                history=state_history
            )

            # 4. 执行行动
            if action.type == "click":
                self.os_interface.click(action.target)
            elif action.type == "type":
                self.os_interface.type(action.text)
            elif action.type == "scroll":
                self.os_interface.scroll(action.direction, action.amount)
            elif action.type == "key":
                self.os_interface.press_key(action.key)
            elif action.type == "complete":
                return TaskResult.success(action.result)

            # 5. 记录状态
            state_history.append({
                "step": step,
                "screenshot": screenshot,
                "action": action,
                "ui_elements": ui_elements
            })

            step += 1
            time.sleep(0.5)  # 等待界面响应

        return TaskResult.incomplete(state_history)

class ComputerUseAgent:
    """
    通用计算机使用Agent

    能够操作：
    - 浏览器
    - 桌面应用
    - 文件系统
    - 命令行
    """

    def __init__(self):
        self.tools = {
            "screenshot": ScreenshotTool(),
            "mouse": MouseControlTool(),
            "keyboard": KeyboardTool(),
            "shell": ShellTool(),
            "browser": BrowserTool()
        }

    def run(self, instruction: str) -> ExecutionTrace:
        """执行需要计算机操作的指令"""

        trace = ExecutionTrace()

        while not self._is_complete():
            # 感知环境
            observation = self._observe()
            trace.add_observation(observation)

            # 推理下一步
            thought = self._reason(instruction, trace)
            trace.add_thought(thought)

            # 选择行动
            action = self._select_action(thought)
            trace.add_action(action)

            # 执行
            result = self._execute(action)
            trace.add_result(result)

        return trace

    def _observe(self) -> Observation:
        """观察当前计算机状态"""
        screenshot = self.tools["screenshot"].capture()

        # 使用视觉模型分析
        analysis = self.vision_model.describe(
            screenshot,
            prompt="Describe the current computer state, including open applications,
                   visible UI elements, and any relevant text content."
        )

        return Observation(
            screenshot=screenshot,
            description=analysis.description,
            ui_elements=analysis.elements
        )

class ComputerUseImplementation:
    """Computer Use Agent的技术实现细节"""

    @staticmethod
    def visual_grounding():
        """
        视觉定位：将自然语言指令映射到屏幕坐标
        """

        class VisualGroundingModel:
            """
            基于视觉-语言模型的UI元素定位
            """

            def __init__(self):
                self.model = CLIP()  # 或类似的视觉-语言模型

            def locate_element(self, screenshot: Image, description: str) -> BoundingBox:
                """
                在截图中定位描述的元素

                Args:
                    screenshot: 屏幕截图
                    description: 元素描述，如"搜索按钮"

                Returns:
                    BoundingBox: 元素位置 (x1, y1, x2, y2)
                """
                # 方法1: 使用OCR检测可点击文本
                text_regions = self.ocr.detect(screenshot)

                # 方法2: 使用图标检测模型
                icon_regions = self.icon_detector.detect(screenshot)

                # 方法3: 使用视觉-语言匹配
                all_regions = text_regions + icon_regions

                # 计算与描述的相似度
                best_match = None
                best_score = 0

                for region in all_regions:
                    # 裁剪区域
                    crop = screenshot.crop(region.bbox)

                    # 计算视觉-文本相似度
                    score = self.model.similarity(crop, description)

                    if score > best_score:
                        best_score = score
                        best_match = region

                return best_match.bbox if best_match else None

    @staticmethod
    def action_space_design():
        """
        动作空间设计
        """

        # 定义Agent可以执行的动作
        ACTION_SPACE = {
            # 鼠标操作
            "mouse_move": {
                "description": "移动鼠标到指定坐标",
                "parameters": {"x": int, "y": int}
            },
            "mouse_click": {
                "description": "在指定位置点击",
                "parameters": {"x": int, "y": int, "button": ["left", "right"]}
            },
            "mouse_drag": {
                "description": "拖拽操作",
                "parameters": {"start": (int, int), "end": (int, int)}
            },
            "mouse_scroll": {
                "description": "滚动",
                "parameters": {"direction": ["up", "down"], "amount": int}
            },

            # 键盘操作
            "key_press": {
                "description": "按下按键",
                "parameters": {"key": str}
            },
            "type_text": {
                "description": "输入文本",
                "parameters": {"text": str}
            },
            "hotkey": {
                "description": "组合键",
                "parameters": {"keys": List[str]}
            },

            # 系统操作
            "screenshot": {
                "description": "截取屏幕",
                "parameters": {}
            },
            "wait": {
                "description": "等待界面响应",
                "parameters": {"seconds": float}
            },

            # 任务控制
            "complete": {
                "description": "标记任务完成",
                "parameters": {"result": str}
            },
            "fail": {
                "description": "标记任务失败",
                "parameters": {"reason": str}
            }
        }

        return ACTION_SPACE

    @staticmethod
    def safety_mechanisms():
        """
        Computer Use的安全机制
        """

        class SafetyGuardrails:
            def __init__(self):
                self.restricted_actions = [
                    "delete_system_files",
                    "modify_system_settings",
                    "access_sensitive_data"
                ]

                self.confirmation_required = [
                    "make_purchase",
                    "send_email",
                    "delete_files",
                    "modify_permissions"
                ]

            def check_action(self, action: Action) -> SafetyResult:
                """检查行动是否安全"""

                # 检查是否在禁止列表
                if action.type in self.restricted_actions:
                    return SafetyResult.reject("Action not allowed")

                # 检查是否需要确认
                if action.type in self.confirmation_required:
                    return SafetyResult.require_confirmation(
                        f"This action will {action.description}. Proceed?"
                    )

                # 检查异常模式
                if self._detect_suspicious_pattern(action):
                    return SafetyResult.require_review("Suspicious pattern detected")

                return SafetyResult.allow()

            def _detect_suspicious_pattern(self, action: Action) -> bool:
                """检测可疑操作模式"""
                # 实现各种安全检查逻辑
                pass

class BrowserAutomation:
    """浏览器自动化实现"""

    def __init__(self):
        self.driver = PlaywrightDriver()
        self.page = None

    async def navigate(self, url: str):  # async def定义协程函数
        """导航到指定URL"""
        self.page = await self.driver.new_page()  # await等待异步操作完成
        await self.page.goto(url)

    async def search(self, query: str, engine: str = "google"):
        """在搜索引擎中搜索"""
        if engine == "google":
            await self.navigate("https://www.google.com")
            await self.page.fill('input[name="q"]', query)
            await self.page.press('input[name="q"]', "Enter")

        # 等待结果加载
        await self.page.wait_for_selector("#search")

    async def fill_form(self, form_data: Dict[str, str]):
        """填写表单"""
        for field, value in form_data.items():
            # 尝试多种选择器策略
            selectors = [
                f'input[name="{field}"]',
                f'input[id="{field}"]',
                f'input[placeholder*="{field}"]',
                f'textarea[name="{field}"]'
            ]

            for selector in selectors:
                try:
                    await self.page.fill(selector, value)
                    break
                except:
                    continue

    async def extract_data(self, extraction_rules: List[Rule]) -> List[Dict]:
        """根据规则提取网页数据"""
        results = []

        for rule in extraction_rules:
            elements = await self.page.query_selector_all(rule.selector)

            for element in elements:
                item = {}
                for field, field_rule in rule.fields.items():
                    value = await element.eval_on_selector(
                        field_rule.selector,
                        "el => el.textContent"
                    )
                    item[field] = value.strip() if value else None  # 链式调用：strip去除空白

                results.append(item)

        return results

    async def monitor_changes(self, selector: str, callback):
        """监控页面元素变化"""
        await self.page.evaluate(f"""
            new MutationObserver((mutations) => {{
                window.__pageChanges = window.__pageChanges || [];
                window.__pageChanges.push(mutations);
            }}).observe(
                document.querySelector('{selector}'),
                {{ childList: true, subtree: true }}
            );
        """)

class AgentCoreArchitecture:
    """Agent核心架构实现"""

    class PerceptionModule:
        """感知模块"""

        def __init__(self):
            self.text_processor = TextProcessor()
            self.vision_processor = VisionProcessor()
            self.audio_processor = AudioProcessor()

        def process(self, input_data: Union[str, Image, Audio]) -> Perception:
            """统一感知处理入口"""
            if isinstance(input_data, str):  # isinstance检查类型
                return self._process_text(input_data)
            elif isinstance(input_data, Image):
                return self._process_image(input_data)
            elif isinstance(input_data, Audio):
                return self._process_audio(input_data)
            else:
                raise ValueError(f"Unsupported input type: {type(input_data)}")

        def _process_text(self, text: str) -> TextPerception:
            """文本感知处理"""
            # 意图识别
            intent = self.text_processor.classify_intent(text)

            # 实体提取
            entities = self.text_processor.extract_entities(text)

            # 情感分析
            sentiment = self.text_processor.analyze_sentiment(text)

            return TextPerception(
                raw_text=text,
                intent=intent,
                entities=entities,
                sentiment=sentiment
            )

    class MemoryModule:
        """记忆模块"""

        def __init__(self, vector_store: VectorStore):
            self.working_memory = WorkingMemory(capacity=7)  # 米勒定律
            self.episodic_memory = EpisodicMemory(vector_store)
            self.semantic_memory = SemanticMemory(vector_store)
            self.procedural_memory = ProceduralMemory()

        def store(self, experience: Experience):
            """存储经验"""
            # 工作记忆
            self.working_memory.add(experience)

            # 情节记忆（长期）
            if experience.is_significant:
                self.episodic_memory.store(experience)

            # 提取知识到语义记忆
            knowledge = self._extract_knowledge(experience)
            self.semantic_memory.store(knowledge)

            # 更新程序性记忆（技能）
            if experience.type == "skill_execution":
                self.procedural_memory.update(experience)

        def retrieve(self, query: str, context: Context) -> RetrievedInfo:
            """检索相关信息"""
            # 多路检索
            working_results = self.working_memory.search(query)
            episodic_results = self.episodic_memory.search(query, top_k=5)
            semantic_results = self.semantic_memory.search(query, top_k=5)
            procedural_results = self.procedural_memory.match(context)

            # 融合排序
            fused = self._fusion_rank(
                working_results,
                episodic_results,
                semantic_results,
                procedural_results
            )

            return fused

    class PlanningModule:
        """规划模块"""

        def __init__(self, llm):
            self.llm = llm
            self.plan_library = PlanLibrary()

        def create_plan(self, goal: Goal, context: Context) -> Plan:
            """创建执行计划"""

            # 尝试从计划库匹配
            similar_plan = self.plan_library.find_similar(goal)
            if similar_plan and similar_plan.success_rate > 0.8:
                return self._adapt_plan(similar_plan, context)

            # 否则生成新计划
            return self._generate_plan(goal, context)

        def _generate_plan(self, goal: Goal, context: Context) -> Plan:
            """使用LLM生成计划"""

            prompt = f"""Create a step-by-step plan to achieve the following goal:

Goal: {goal.description}

Context:
{context.to_prompt()}

Available tools: {context.available_tools}

Generate a detailed plan with:
1. Sequential steps
2. Each step's expected outcome
3. Dependencies between steps
4. Potential failure points and alternatives

Output as structured JSON."""

            plan_json = self.llm.generate(prompt, format="json")
            return Plan.from_json(plan_json)

        def replan(self, current_plan: Plan, failure: Failure) -> Plan:
            """失败后的重规划"""

            prompt = f"""The current plan failed. Create a revised plan.

Original Plan: {current_plan.to_json()}
Failure: {failure.description}
Current State: {failure.state}

Provide a new plan that addresses the failure."""

            new_plan_json = self.llm.generate(prompt, format="json")
            return Plan.from_json(new_plan_json)

    class ToolModule:
        """工具模块"""

        def __init__(self):
            self.tools: Dict[str, Tool] = {}
            self.tool_descriptions = []

        def register(self, tool: Tool):
            """注册工具"""
            self.tools[tool.name] = tool
            self.tool_descriptions.append(tool.get_description())

        def select_and_execute(self, intent: Intent, context: Context) -> ToolResult:
            """选择并执行合适的工具"""

            # 工具选择
            tool_name = self._select_tool(intent, context)
            tool = self.tools[tool_name]

            # 参数提取
            parameters = self._extract_parameters(intent, tool)

            # 执行
            try:
                result = tool.execute(**parameters)  # **parameters将字典解包为关键字参数，实现动态参数传递
                return ToolResult.success(result)
            except Exception as e:
                return ToolResult.failure(str(e))

        def _select_tool(self, intent: Intent, context: Context) -> str:
            """基于意图选择工具"""

            prompt = f"""Given the user intent and available tools, select the most appropriate tool.

Intent: {intent.description}
Available Tools:
{self._format_tool_descriptions()}

Select the best tool and explain why."""

            selection = self.llm.generate(prompt)
            return self._parse_tool_selection(selection)