发布于 由admin

解析 ‘Meta-Cognition in Agents’：让 Agent 定期运行一个‘自我审查’节点，检查之前的决策是否有误

各位听众，下午好。

今天，我们将深入探讨一个引人入胜且至关重要的主题：Agent 中的元认知，特别是如何让 Agent 具备“自我审查”的能力，从而检查并纠正其过去的决策。在人工智能领域，我们已经见证了 Agent 在特定任务中展现出卓越的性能。然而，真正的智能不仅仅是执行任务，更重要的是理解、反思和改进自身。这正是元认知能力的核心价值。

什么是元认知？以及它在 Agent 中的重要性

元认知（Meta-Cognition）一词源自心理学，指的是“关于认知的认知”，即个体对自己认知过程的理解、监控和调节。简单来说，就是“思考你的思考”。当一个人学习、解决问题或做出决策时，如果他能同时反思这个过程本身，例如“我是否理解了这个问题？”、“我目前的解决策略有效吗？”、“我犯了什么错误？”，那么他就具备了元认知能力。

在 Agent 的语境中，元认知意味着一个 Agent 不仅仅是感知环境、规划行动、执行任务，它还能：

  1. 监控自身的内部状态和外部行为。
  2. 评估其决策的质量和结果。
  3. 诊断失败的原因或次优行为。
  4. 调整其内部知识、策略或学习过程，以提高未来的表现。

为什么元认知在 Agent 中如此重要？

  • 适应性与鲁棒性： 真实世界是动态、不确定且充满变化的。Agent 需要能够适应新的情况，从错误中学习，而不是简单地重复失败。元认知提供了这种适应性的基础。
  • 自主学习与成长： 具备元认知能力的 Agent 可以更自主地进行学习和改进，减少对外部反馈或人工干预的依赖。
  • 错误纠正与安全性： 识别和纠正错误对于部署在关键应用中的 Agent 至关重要，能够避免潜在的风险和损失。
  • 可解释性与信任： 当 Agent 能够解释其决策过程以及为何进行修正时，这有助于提高人类对其行为的理解和信任。

今天，我们将聚焦于元认知的一个关键实践：让 Agent 定期运行一个“自我审查”节点，检查之前的决策是否有误。这如同人类在项目结束后进行复盘，或在考试后分析错题，旨在通过反思来优化未来的表现。

传统 Agent 决策流程的局限性

在深入探讨元认知 Agent 之前，让我们回顾一下一个典型的、不具备元认知能力的 Agent 是如何运作的。一个基础的 Agent 通常遵循一个感知-规划-行动（Sense-Plan-Act）循环：

  1. 感知 (Sense): Agent 通过传感器（或API）获取当前环境的状态信息。
  2. 规划 (Plan): 基于感知到的信息、其内部目标和现有的知识模型，Agent 制定一个行动计划或选择一个最佳行动。
  3. 行动 (Act): Agent 执行选定的行动，改变环境状态。

这个循环不断重复。

示例：一个简单的机器人 Agent

考虑一个在仓库中搬运物品的机器人 Agent。

  • 感知: 摄像头识别周围环境（障碍物、目标物品、目的地），内部传感器提供电池电量、当前位置。
  • 规划: 根据目标（将物品X从A搬到B），规划一条路径，决定抓取动作、移动速度。
  • 行动: 移动到物品X旁，抓取物品，移动到目的地B，放下物品。

这种模式在环境稳定、规则明确、目标单一的情况下运作良好。然而，一旦环境变得复杂、动态或不确定，其局限性便会显现：

  • 无法从错误中学习： 如果机器人选择了一条低效的路径，或者在某个地方卡住了，传统的 Agent 仅仅会尝试再次执行其当前的规划，或者在规划失败时简单地重新规划，而不会深入分析“为什么”会失败。它缺乏一种机制来识别其决策过程中的缺陷，并将其转化为经验教训。
  • 重复犯错： 如果导致错误的根本原因（例如，对某个区域的地图信息不准确，或对抓取力度的估计有偏差）没有被识别和纠正，Agent 会在类似的情况下重复犯同样的错误。
  • 缺乏适应性： 当环境规则发生变化（例如，仓库中新增了移动障碍物），Agent 可能无法有效地调整其策略，除非通过外部的人工重新编程或大量重新训练。
  • 次优决策： 即使决策没有导致完全的失败，也可能不是最优的。传统 Agent 缺乏自我评估的能力来识别这些次优的路径或行动，从而错失提升效率的机会。

这些局限性促使我们思考：Agent 如何才能像人类一样，不仅能做，还能思考自己是如何做的，以及如何做得更好？这正是引入元认知“自我审查”节点的意义所在。

元认知：引入“自我审查”节点

元认知 Agent 的核心思想是，它不仅是一个“执行者”，更是一个“反思者”和“学习者”。它不再满足于仅仅完成任务，而是会周期性地停下来，回顾其过去的决策、行动和结果，进行深刻的自我分析。这个过程我们称之为“自我审查”或“反思”。

“自我审查”节点的功能主要包括：

  1. 回顾历史决策： 访问 Agent 的记忆模块，检索过去一段时间内的决策记录、环境观测、Agent 内部状态以及采取的行动。
  2. 评估结果： 比较实际发生的行动结果与 Agent 在决策时所预期的结果。判断目标是否达成，以及达成质量如何（例如，是否高效、是否符合约束条件）。
  3. 识别错误与次优： 基于评估结果，诊断导致失败或次优表现的根本原因。这可能涉及：
    • 决策错误： Agent 在特定情境下选择了错误的行动。
    • 知识缺陷： Agent 内部世界模型不准确或不完整。
    • 推理偏差： Agent 的规划逻辑存在缺陷。
    • 执行失败： 行动本身由于外部因素或自身能力限制未能成功执行。
  4. 生成改进建议： 根据识别出的问题，提出具体的改进措施，例如更新内部规则、调整模型参数、修改规划策略，甚至建议获取新的信息。

这就像一个工程师在项目完成后，不仅要提交代码，还要撰写事后分析报告（post-mortem），详细记录遇到的问题、解决方案，以及未来的改进方向。通过这种方式，Agent 能够将过去的经验转化为宝贵的知识，用于指导未来的决策，从而实现真正的“从错误中学习，不断进步”。

设计一个元认知 Agent 架构

为了实现上述的元认知能力，我们需要对传统的 Agent 架构进行扩展。一个新的架构将包含以下关键组件：

  1. 感知器 (Perceiver):

    • 功能: 从环境中收集原始数据和信息。
    • 输出: 格式化的环境观测数据，例如传感器读数、API响应、用户输入等。

  2. 决策器 (Planner/Decision Maker):

    • 功能: 基于感知到的环境状态、Agent 目标、内部知识和当前策略，生成一个或一系列行动计划。
    • 输入: 环境观测、Agent 目标、内部知识。
    • 输出: 待执行的行动。

  3. 执行器 (Actuator):

    • 功能: 将决策器生成的抽象行动转化为具体的物理或数字操作，并作用于环境。
    • 输入: Agent 决策的行动。
    • 输出: 环境状态的改变。

  4. 记忆模块 (Memory Module): （新增）

    • 功能: 存储 Agent 的所有相关信息，包括环境观测、Agent 内部状态、采取的行动、决策过程（为什么选择该行动）、行动结果以及其他关键事件。这是自我审查的基础。
    • 存储内容: 决策记录（DecisionRecord）、环境历史、任务目标等。

  5. 自我审查模块 (Self-Critique Module): （新增）

    • 功能: 定期或在特定条件下被触发，从记忆模块中检索历史决策，评估其有效性，诊断潜在的错误或次优之处，并生成改进建议。
    • 输入: 历史决策记录和相关环境信息。
    • 输出: 错误诊断报告、改进建议。

  6. 知识更新模块 (Knowledge Update Module): （新增）

    • 功能: 接收来自自我审查模块的改进建议，并据此更新 Agent 的内部知识、决策规则、策略模型或系统提示（对于基于 LLM 的 Agent）。
    • 输入: 改进建议。
    • 输出: 更新后的 Agent 内部知识和行为策略。

Agent 核心循环（扩展后）：

  1. 感知: Perceiver 收集环境信息。
  2. 记录: 将当前环境观测和 Agent 内部状态发送给 Memory Module。
  3. 决策: Decision Maker 基于当前信息和内部知识制定行动。
  4. 记录: 将决策过程和预期结果发送给 Memory Module。
  5. 执行: Actuator 执行行动。
  6. 记录: 将实际行动结果发送给 Memory Module。
  7. 循环检查（自我审查触发）： 在特定时机（例如，每N个决策后，或当任务失败时），Self-Critique Module 被触发。
  8. 审查: Self-Critique Module 从 Memory Module 获取历史记录，进行分析，生成改进建议。
  9. 更新: Knowledge Update Module 根据建议修改 Decision Maker 的内部知识或策略。
  10. 继续循环。

通过这种扩展架构，Agent 形成了一个完整的学习闭环，使其能够持续地从经验中学习和成长。

记忆模块的实现：构建决策历史与环境状态存储

记忆模块是元认知 Agent 的基石。没有详细且结构化的记忆，自我审查就无从谈起。我们需要存储足够的信息，以便 Agent 在未来能够重现决策时的上下文，并评估其结果。

需要存储的内容：

  1. 时间戳 (Timestamp): 记录事件发生的时间，有助于排序和追溯。
  2. 环境观测 (Environmental Observation): Agent 在做出决策时所感知到的环境状态的详细快照。
  3. Agent 目标 (Agent Goal): Agent 在该时间点试图达成的具体目标。
  4. Agent 内部状态 (Agent Internal State): Agent 自身的内部变量、信念、效用函数参数等。
  5. 采取的行动 (Action Taken): Agent 实际执行的行动。
  6. 预期结果 (Expected Outcome): Agent 在选择该行动时所预期的结果。
  7. 实际结果 (Actual Outcome): 行动执行后环境的实际变化或Agent自身状态的变化。
  8. 决策理由 (Decision Rationale): Agent 做出该决策的思考过程、依据的规则或推理链。这对于基于 LLM 的 Agent 尤为重要。

数据结构选择：

  • 列表 (List of Objects): 最简单的方式，将每个决策记录作为一个对象存储在列表中。适用于小规模、短期的记忆。
  • 关系型数据库 (Relational Database): 适用于需要复杂查询、持久化存储和结构化数据的场景。可以为不同类型的记录创建表。
  • 知识图谱 (Knowledge Graph): 对于需要存储复杂关系和语义信息的场景，知识图谱能提供更丰富的上下文。
  • 向量数据库 (Vector Database): 结合 LLM，可以将决策记录编码为向量，方便进行语义相似性搜索和检索。

在这里，我们为了简洁和演示，将使用一个 Python 类来模拟一个简单的记忆模块，存储一系列 DecisionRecord 对象。

import datetime
import uuid
from typing import Any, Dict, List, Optional

# 定义一个决策记录的结构
class DecisionRecord:
    """
    存储Agent每一次决策的详细信息，用于后续的自我审查。
    """
    def __init__(self,
                 timestamp: datetime.datetime,
                 decision_id: str,
                 agent_goal: str,
                 environment_observation: Dict[str, Any],
                 agent_internal_state: Dict[str, Any],
                 action_taken: Dict[str, Any],
                 expected_outcome: Dict[str, Any],
                 actual_outcome: Optional[Dict[str, Any]] = None,
                 decision_rationale: Optional[str] = None,
                 critique_report: Optional[Dict[str, Any]] = None):
        self.timestamp = timestamp
        self.decision_id = decision_id  # 唯一标识符
        self.agent_goal = agent_goal
        self.environment_observation = environment_observation
        self.agent_internal_state = agent_internal_state
        self.action_taken = action_taken
        self.expected_outcome = expected_outcome
        self.actual_outcome = actual_outcome
        self.decision_rationale = decision_rationale
        self.critique_report = critique_report # 存储审查结果，方便追溯

    def __repr__(self):
        return (f"DecisionRecord(ID={self.decision_id[:8]}, Goal='{self.agent_goal}', "
                f"Action={self.action_taken}, Outcome={self.actual_outcome})")

    def to_dict(self):
        """将记录转换为字典，便于存储或发送给LLM。"""
        return {
            "timestamp": self.timestamp.isoformat(),
            "decision_id": self.decision_id,
            "agent_goal": self.agent_goal,
            "environment_observation": self.environment_observation,
            "agent_internal_state": self.agent_internal_state,
            "action_taken": self.action_taken,
            "expected_outcome": self.expected_outcome,
            "actual_outcome": self.actual_outcome,
            "decision_rationale": self.decision_rationale,
            "critique_report": self.critique_report
        }

# 记忆模块类
class MemoryModule:
    """
    负责存储和检索Agent的决策历史。
    """
    def __init__(self):
        self.decision_history: List[DecisionRecord] = []

    def add_record(self, record: DecisionRecord):
        """
        添加一个新的决策记录。
        """
        self.decision_history.append(record)
        print(f"Memory: Added new decision record {record.decision_id[:8]}.")

    def get_records(self, count: Optional[int] = None, since: Optional[datetime.datetime] = None) -> List[DecisionRecord]:
        """
        检索决策记录。
        Args:
            count: 要检索的最近记录的数量。
            since: 从某个时间点开始检索记录。
        Returns:
            符合条件的DecisionRecord列表。
        """
        records = self.decision_history
        if since:
            records = [r for r in records if r.timestamp >= since]
        if count:
            records = records[-count:] # 获取最新的N条记录
        return records

    def update_record_critique(self, decision_id: str, critique_report: Dict[str, Any]):
        """
        更新特定决策记录的审查报告。
        """
        for record in self.decision_history:
            if record.decision_id == decision_id:
                record.critique_report = critique_report
                print(f"Memory: Updated critique for decision {decision_id[:8]}.")
                return
        print(f"Memory: Decision {decision_id[:8]} not found for critique update.")

# 示例用法
if __name__ == "__main__":
    memory = MemoryModule()

    # 模拟Agent做出一个决策并记录
    current_time = datetime.datetime.now()
    decision1 = DecisionRecord(
        timestamp=current_time,
        decision_id=str(uuid.uuid4()),
        agent_goal="Find optimal path",
        environment_observation={"location": "A", "traffic": "low", "weather": "sunny"},
        agent_internal_state={"energy": 80, "knowledge_version": "1.0"},
        action_taken={"action_type": "move", "destination": "B", "path_id": "path_1"},
        expected_outcome={"arrival_time": "T+10min", "energy_cost": 5},
        decision_rationale="Path_1 is shortest and traffic is low."
    )
    memory.add_record(decision1)

    # 模拟一段时间后，Agent完成了行动并记录实际结果
    import time
    time.sleep(0.1)
    decision1.actual_outcome = {"arrival_time": "T+15min", "energy_cost": 7, "obstacle_encountered": True}
    print(f"Memory: Updated actual outcome for {decision1.decision_id[:8]}.")

    # 模拟Agent做出第二个决策
    time.sleep(0.1)
    current_time = datetime.datetime.now()
    decision2 = DecisionRecord(
        timestamp=current_time,
        decision_id=str(uuid.uuid4()),
        agent_goal="Deliver package",
        environment_observation={"location": "B", "package_status": "ready"},
        agent_internal_state={"energy": 73, "knowledge_version": "1.0"},
        action_taken={"action_type": "deliver", "package_id": "PKG123", "recipient": "C"},
        expected_outcome={"delivery_status": "completed", "energy_cost": 2},
        decision_rationale="Recipient C is nearby."
    )
    memory.add_record(decision2)
    time.sleep(0.1)
    decision2.actual_outcome = {"delivery_status": "completed", "energy_cost": 2}
    print(f"Memory: Updated actual outcome for {decision2.decision_id[:8]}.")

    # 检索最近的记录
    print("nRetrieving last 2 records:")
    recent_records = memory.get_records(count=2)
    for r in recent_records:
        print(r)

    # 假设自我审查模块对 decision1 进行了审查并生成了报告
    critique_report_for_d1 = {
        "status": "failed",
        "reason": "Path_1 had an unexpected obstacle, leading to longer travel time and higher energy cost.",
        "suggestion": "Update map with obstacle information, consider alternative paths even if slightly longer initially."
    }
    memory.update_record_critique(decision1.decision_id, critique_report_for_d1)

    print("nDecision 1 after critique update:")
    print(decision1.to_dict())

这段代码展示了一个基础的记忆模块如何存储决策记录，并允许我们追溯和更新这些记录。这些记录将是自我审查模块进行分析的原始数据。

自我审查模块的核心逻辑与挑战

自我审查模块是元认知 Agent 的“大脑”，负责对过去的决策进行分析和反思。

审查触发机制 (Trigger Mechanism):

  • 定期触发 (Periodic): 每隔一段时间（例如，每小时、每天）或每 N 个决策后运行一次。这确保了 Agent 能够持续地进行反思。
  • 事件驱动触发 (Event-driven): 当特定事件发生时触发，例如：
    • 任务失败： Agent 未能达成其目标。
    • 异常行为： Agent 表现出与预期不符或效率低下的行为。
    • 性能下降： 任务完成时间变长，资源消耗增加等。
    • 外部反馈： 收到人类用户或环境的负面反馈。
  • 需求驱动触发 (Demand-driven): Agent 内部检测到知识空白或决策不确定性高时主动触发。

审查对象 (Scope of Critique):

  • 单一决策 (Single Decision): 分析某个特定的、最近的或失败的决策。
  • 决策序列 (Sequence of Decisions): 分析一系列相关的决策，以理解其间的因果关系和整体策略的有效性。
  • 特定类型决策 (Specific Decision Types): 针对某一类重复出现的决策模式进行审查。

审查方法 (Critique Methods):

这里我们将介绍几种主要的方法，并特别强调基于 LLM 的方法。

  1. 规则匹配 (Rule-based Critique):

    • 原理: 预定义一系列的错误模式和触发条件。当决策记录符合这些模式时，就标记为错误。
    • 优点: 简单、高效，对于已知且明确的错误非常有效。
    • 缺点: 无法发现未知错误或复杂、隐含的模式。需要人工维护规则。
    • 示例: "如果实际能耗超过预期能耗20%且任务完成时间延长，则标记为次优。"

  2. 模拟与反事实推理 (Simulation & Counterfactual Reasoning):

    • 原理: Agent 内部维护一个环境模型。当审查一个决策时，Agent 可以在其内部模型中“回放”该决策，甚至尝试其他可能的行动（反事实），然后比较这些行动的预期结果与实际结果。
    • 优点: 能够发现潜在的更好选择，评估不同策略的优劣。
    • 缺点: 依赖于准确的环境模型，模拟成本高。
    • 示例: "如果当时选择了路径A而不是路径B，结果会更好吗？" Agent 在模拟器中运行两种情况并比较。

  3. 基于模型的评估 (Model-based Evaluation):

    • 原理: Agent 拥有一个评估模型（例如，一个预测任务成功率、资源消耗或奖励值的模型）。审查时，将决策输入评估模型，得到一个分数，与预期分数或最佳分数进行比较。
    • 优点: 可以量化决策质量，发现次优决策。
    • 缺点: 评估模型本身需要训练和维护。
    • 示例: 使用一个预训练的Q-network来评估某个状态-动作对的Q值，看是否与期望值有较大偏差。

  4. 外部反馈融合 (External Feedback Integration):

    • 原理: 将人类用户或外部系统提供的反馈信息整合到审查过程中。
    • 优点: 引入了外部视角，可以纠正 Agent 自身无法识别的错误。
    • 缺点: 外部反馈可能不及时、不完整或存在偏差。
    • 示例: 用户手动标记某个Agent行动为“错误”或“低效”。

  5. 启发式评估 (Heuristic Evaluation):

    • 原理: 使用一些简单的经验法则或启发式规则来快速判断决策的质量。
    • 优点: 快速、低开销。
    • 缺点: 粗略，可能漏掉细微的错误或次优情况。
    • 示例: "如果目标未能在预设时间内完成，则视为失败决策。"

  6. 基于大型语言模型 (LLM) 的反思 (LLM-based Reflection):

    • 原理: 利用 LLM 强大的文本理解、推理和生成能力，将决策记录和相关上下文作为 Prompt 输入，让 LLM 分析决策的合理性、识别错误并提出改进建议。
    • 优点:
      • 上下文理解: 能够理解复杂的决策理由和环境描述。
      • 抽象推理: 能够识别高层次的策略缺陷，而不仅仅是表面错误。
      • 自然语言解释: 以人类可读的方式输出错误诊断和改进建议。
      • 泛化能力: 能够处理未曾明确编码的错误模式。
    • 缺点:
      • 计算成本: 调用 LLM 成本较高，延迟较大。
      • 幻觉与偏差: LLM 可能生成不准确或不相关的建议。
      • Prompt 工程: 需要精心设计 Prompt 才能获得高质量的输出。
      • 缺乏真实世界交互: LLM 本身不具备感知和行动能力，其分析基于提供的文本信息。

识别错误的类型:

  • 错误决策 (Incorrect Decision): Agent 选择了导致任务失败或严重负面后果的行动。
  • 次优决策 (Suboptimal Decision): Agent 选择了能完成任务但效率低下、资源消耗过多的行动，存在更好的替代方案。
  • 遗漏机会 (Missed Opportunity): Agent 在特定情境下没有识别或利用到一个可以带来显著收益的机会。
  • 知识模型不准确 (Inaccurate Knowledge Model): Agent 对环境、自身能力或任务规则的理解存在偏差。
  • 推理链缺陷 (Reasoning Flaw): Agent 在从观察到行动的推理过程中存在逻辑错误。

我们将主要关注基于 LLM 的自我审查方法，因为它在处理复杂、非结构化信息和生成可解释建议方面具有显著优势。

import datetime
import json
from typing import List, Dict, Any, Optional

# 假设我们有一个LLM客户端接口
class MockLLMClient:
    def generate_response(self, prompt: str, temperature: float = 0.7) -> str:
        """
        模拟LLM的响应。
        在实际应用中，这里会调用OpenAI, Anthropic或其他LLM提供商的API。
        """
        print(f"n--- MockLLM Input Prompt ---n{prompt}n--- End MockLLM Input ---")
        # 这是一个非常简化的模拟，实际LLM会根据prompt内容进行复杂推理
        if "任务失败" in prompt or "次优" in prompt:
            return json.dumps({
                "status": "critiqued",
                "diagnosis": "The agent chose a suboptimal path due to outdated traffic data. The expected arrival time was missed, and energy consumption was higher.",
                "root_cause": "Outdated traffic data in the agent's internal knowledge base.",
                "suggestion": "Prioritize real-time traffic data integration. Consider dynamic path re-planning if conditions change significantly during travel.",
                "confidence": "high"
            })
        else:
            return json.dumps({
                "status": "successful",
                "diagnosis": "The agent successfully delivered the package as planned, on time and within energy budget.",
                "root_cause": "Accurate route planning and efficient execution.",
                "suggestion": "Continue to monitor for efficiency improvements, no immediate changes needed.",
                "confidence": "high"
            })

class SelfCritiqueModule:
    """
    负责Agent的自我审查，诊断过去决策中的错误或次优之处。
    """
    def __init__(self, memory: MemoryModule, llm_client: MockLLMClient, critique_interval: int = 5):
        self.memory = memory
        self.llm_client = llm_client
        self.critique_interval = critique_interval # 每多少个决策触发一次审查
        self._decision_count_since_last_critique = 0

    def should_critique(self) -> bool:
        """
        判断是否应该触发自我审查。
        """
        self._decision_count_since_last_critique += 1
        if self._decision_count_since_last_critique >= self.critique_interval:
            self._decision_count_since_last_critique = 0 # 重置计数器
            return True
        return False

    def _generate_critique_prompt(self, record: DecisionRecord) -> str:
        """
        根据决策记录生成用于LLM的Prompt。
        """
        prompt = f"""
        你是一名高度智能的AI Agent审查专家。你的任务是分析一个AI Agent的决策记录，并提供一份详细的审查报告。
        请仔细评估Agent的决策是否有效、是否达到预期目标，以及是否存在任何错误、次优之处或改进空间。

        请根据以下JSON格式的决策记录进行分析：
        {json.dumps(record.to_dict(), indent=2, ensure_ascii=False)}

        请特别关注以下几点：
        1.  **目标达成情况:** Agent是否成功实现了 'agent_goal'？
        2.  **预期与实际结果对比:** 'expected_outcome' 和 'actual_outcome' 之间是否存在显著差异？如果有，请分析原因。
        3.  **决策合理性:** 基于 'environment_observation' 和 'agent_internal_state'，Agent 的 'action_taken' 和 'decision_rationale' 是否合理？
        4.  **错误/次优识别:** 是否存在任何明显的错误、低效行为或可以改进的地方？
        5.  **根本原因分析:** 如果有错误或次优，请尝试分析其根本原因（例如，知识不准确、推理缺陷、环境变化等）。
        6.  **改进建议:** 提出具体的、可操作的建议，以帮助Agent在未来做得更好。

        请以以下JSON格式返回你的审查报告。确保所有字段都存在，并且内容清晰、具体。
        {{
            "status": "string (e.g., 'successful', 'failed', 'suboptimal')",
            "diagnosis": "string (详细描述Agent决策的表现和问题)",
            "root_cause": "string (导致问题发生的根本原因)",
            "suggestion": "string (具体的改进措施)",
            "confidence": "string (e.g., 'high', 'medium', 'low')"
        }}
        """
        return prompt

    def perform_critique(self) -> Optional[Dict[str, Any]]:
        """
        执行自我审查过程，获取最近的决策记录并进行分析。
        返回一个包含所有审查结果和建议的列表，或None如果无记录可审查。
        """
        records_to_critique = self.memory.get_records(count=self.critique_interval)
        if not records_to_critique:
            print("SelfCritique: No new records to critique.")
            return None

        print(f"nSelfCritique: Starting critique for {len(records_to_critique)} records.")
        all_critique_results = {}
        for record in records_to_critique:
            prompt = self._generate_critique_prompt(record)
            try:
                llm_response_str = self.llm_client.generate_response(prompt)
                critique_report = json.loads(llm_response_str)
                # 将审查报告存储回记忆模块，与原始决策关联
                self.memory.update_record_critique(record.decision_id, critique_report)
                all_critique_results[record.decision_id] = critique_report
                print(f"SelfCritique: Critiqued decision {record.decision_id[:8]} -> Status: {critique_report.get('status')}")
            except json.JSONDecodeError as e:
                print(f"SelfCritique Error: Failed to parse LLM response for {record.decision_id[:8]}: {e}")
            except Exception as e:
                print(f"SelfCritique Error: An unexpected error occurred during LLM call for {record.decision_id[:8]}: {e}")

        return all_critique_results

# 示例用法 (结合MemoryModule)
if __name__ == "__main__":
    memory = MemoryModule()
    llm_client = MockLLMClient()
    critique_module = SelfCritiqueModule(memory, llm_client, critique_interval=2) # 每2个决策审查一次

    # 模拟Agent的第一次决策
    decision1 = DecisionRecord(
        timestamp=datetime.datetime.now(),
        decision_id=str(uuid.uuid4()),
        agent_goal="Navigate to destination X",
        environment_observation={"current_pos": "A", "route_options": ["path1", "path2"], "traffic_data": {"path1": "low", "path2": "medium"}},
        agent_internal_state={"energy": 90, "map_version": "2.0"},
        action_taken={"action_type": "move", "selected_route": "path1"},
        expected_outcome={"arrival_time": "T+10min", "energy_cost": 5},
        decision_rationale="Path1 is faster according to current traffic."
    )
    memory.add_record(decision1)
    decision1.actual_outcome = {"arrival_time": "T+15min", "energy_cost": 8, "unexpected_congestion": True} # 模拟实际结果与预期不符

    # 模拟Agent的第二次决策
    import time
    time.sleep(0.1)
    decision2 = DecisionRecord(
        timestamp=datetime.datetime.now(),
        decision_id=str(uuid.uuid4()),
        agent_goal="Pick up item Y",
        environment_observation={"item_pos": "B", "robot_pos": "A", "item_status": "available"},
        agent_internal_state={"energy": 82, "gripper_strength_setting": "medium"},
        action_taken={"action_type": "grab", "item_id": "itemY"},
        expected_outcome={"item_grabbed": True, "energy_cost": 2},
        decision_rationale="Item Y is within reach, medium strength is sufficient."
    )
    memory.add_record(decision2)
    decision2.actual_outcome = {"item_grabbed": True, "energy_cost": 2} # 模拟成功

    # 此时，应该触发审查
    if critique_module.should_critique():
        print("n--- Triggering Self-Critique ---")
        critique_results = critique_module.perform_critique()
        if critique_results:
            print("nAggregated Critique Results:")
            for dec_id, report in critique_results.items():
                print(f"  Decision {dec_id[:8]}: {report.get('status')} - {report.get('diagnosis')}")

    # 模拟Agent的第三次决策 (不会立即触发审查)
    time.sleep(0.1)
    decision3 = DecisionRecord(
        timestamp=datetime.datetime.now(),
        decision_id=str(uuid.uuid4()),
        agent_goal="Report status",
        environment_observation={"system_health": "good"},
        agent_internal_state={"report_interval": "1hr"},
        action_taken={"action_type": "send_report"},
        expected_outcome={"report_sent": True},
        decision_rationale="Standard operational procedure."
    )
    memory.add_record(decision3)
    decision3.actual_outcome = {"report_sent": True}

    if critique_module.should_critique():
        print("n--- Triggering Self-Critique (after 3rd decision, now total 3) ---")
        critique_results = critique_module.perform_critique()
        if critique_results:
            print("nAggregated Critique Results:")
            for dec_id, report in critique_results.items():
                print(f"  Decision {dec_id[:8]}: {report.get('status')} - {report.get('diagnosis')}")
    else:
        print("nSelf-Critique not triggered yet for 3rd decision.")

在上述代码中，SelfCritiqueModule 定义了审查的触发机制 (should_critique) 和如何生成 LLM Prompt (_generate_critique_prompt)。perform_critique 方法将从记忆中获取最新的决策记录，并调用模拟的 LLM 客户端进行分析。LLM 的响应（这里是模拟的 JSON 字符串）会被解析并存储回原始的 DecisionRecord 中，以便于追溯。

挑战：

  • Prompt 工程的艺术： 编写高质量的 Prompt 至关重要。Prompt 必须清晰、具体，引导 LLM 给出结构化的、有用的回答。模糊的 Prompt 会导致模糊的分析。
  • LLM 的局限性： LLM 可能“幻觉”出不存在的问题或提出不切实际的建议。需要通过后续的验证或人类监督来筛选。
  • 成本与延迟： 频繁调用大型 LLM 会带来显著的计算成本和延迟。需要权衡审查的频率和深度。
  • 可验证性： LLM 给出的建议有时难以直接验证其有效性。
  • 上下文窗口限制： LLM 的上下文窗口有限，一次性分析大量历史决策可能需要复杂的摘要或分批处理策略。

深入探讨基于 LLM 的自我审查

鉴于 LLM 在理解复杂文本和生成有意义解释方面的独特优势，我们进一步详细讨论如何利用它进行自我审查。

LLM 的优势在于：

  • 理解非结构化数据： 能够处理自然语言描述的环境观测、决策理由和行动结果，而无需预先进行严格的结构化。
  • 高层次推理： 能够识别出不仅仅是简单的规则匹配错误，而是更深层次的策略缺陷、逻辑漏洞或知识偏差。
  • 生成解释和建议： 以自然语言形式输出详细的诊断报告，包括根本原因分析和具体的改进建议，这对于人类理解和后续的知识更新非常有价值。
  • 泛化能力： 即使面对从未见过的特定错误模式，LLM 也能基于其广泛的知识进行合理的推断。

Prompt Engineering 的关键：

高质量的 Prompt 是从 LLM 获得高质量审查结果的关键。一个有效的 Prompt 应该包含以下要素：

  1. 明确的角色设定 (Role Assignment): 让 LLM 扮演一个专业的、公正的审查员角色。
    • 示例: "你是一名经验丰富的AI Agent行为分析师和优化专家。"
  2. 清晰的任务定义 (Clear Task Definition): 明确告知 LLM 需要做什么。
    • 示例: "你的任务是审查以下Agent的决策记录，识别问题，分析原因，并提供改进建议。"
  3. 提供详尽的上下文 (Provide Context): 将 Agent 的决策记录（包括环境、目标、行动、预期和实际结果、决策理由）完整地提供给 LLM。
    • 示例: 以 JSON 格式嵌入 DecisionRecord 的所有字段。
  4. 指定分析维度 (Specify Analysis Dimensions): 引导 LLM 从哪些方面进行分析。
    • 示例: 目标达成、预期与实际差异、决策合理性、错误类型、根本原因等。
  5. 明确的输出格式 (Explicit Output Format): 要求 LLM 以结构化的格式（如 JSON）返回结果，这便于后续的程序处理。
    • 示例: 提供一个 JSON 模板，并要求严格遵守。
  6. 强调具体性和可操作性 (Emphasize Specificity & Actionability): 鼓励 LLM 提出具体的、可执行的建议，而不是泛泛而谈。
    • 示例: "请确保改进建议是具体的，并且可以指导Agent的后续行为调整。"

示例 Prompt 优化：

在之前的 _generate_critique_prompt 基础上，我们可以进一步优化，使其更具指导性。

# ... (MemoryModule, DecisionRecord, MockLLMClient classes remain the same) ...

class SelfCritiqueModule:
    # ... (init and should_critique methods remain the same) ...

    def _generate_critique_prompt(self, record: DecisionRecord) -> str:
        """
        根据决策记录生成用于LLM的Prompt。
        优化后的Prompt，更强调结构化分析和具体建议。
        """
        prompt = f"""
        你是一名经验丰富的AI Agent行为分析师和优化专家。
        你的任务是深入审查一份AI Agent的单一决策记录，并生成一份结构化的专业审查报告。
        请你扮演一个严格而公正的审查员，旨在帮助Agent识别其决策过程中的潜在缺陷并提供具体改进方案。

        以下是Agent的决策细节，请以JSON格式呈现：
        ```json
        {json.dumps(record.to_dict(), indent=2, ensure_ascii=False)}
    请按照以下步骤进行详细分析：

    **步骤1：总结决策关键信息**
    简要概述Agent的目标、采取的行动及其主要背景。

    **步骤2：评估决策结果**
    比较“expected_outcome”（预期结果）与“actual_outcome”（实际结果）。
    -   目标是否完全达成？
    -   实际结果是否与预期存在显著差异？请量化差异（如果可能）。
    -   是否存在意外的积极或消极后果？

    **步骤3：分析决策合理性**
    基于“environment_observation”（环境观测）、“agent_internal_state”（Agent内部状态）和“decision_rationale”（决策理由），
    -   Agent选择的“action_taken”（采取的行动）在当时是否是最优的？是否存在更好的替代方案？
    -   “decision_rationale”是否充分且逻辑严谨？是否存在信息缺失、误解或推理偏差？

    **步骤4：诊断问题类型与根本原因**
    如果发现问题（如任务失败、次优决策、资源浪费、效率低下等），请明确指出问题类型，并深入分析其根本原因。
    根本原因可能包括但不限于：
    -   **知识缺陷:** Agent对环境模型、规则或自身能力的认知不准确或不完整。
    -   **感知错误:** Agent对环境信息的感知有偏差或遗漏。
    -   **规划/推理缺陷:** Agent的决策逻辑或规划算法存在漏洞。
    -   **执行偏差:** 实际执行与计划有出入。
    -   **环境突变:** 外部环境发生了Agent未预料到的重大变化。

    **步骤5：提出具体的改进建议**
    根据根本原因分析，提出至少一条具体的、可操作的改进建议。这些建议应能直接指导Agent在未来类似情况下如何调整其知识、策略或行为。
    建议可以是：
    -   更新Agent的知识库（例如，添加新规则、修正世界模型）。
    -   调整Agent的决策策略或参数。
    -   建议Agent寻求更多信息或进行额外的环境探索。
    -   优化Agent的Prompt（如果Agent本身基于LLM）。

    请严格以以下JSON格式返回你的完整审查报告。请确保所有字段都存在，并且内容具体、详细、避免模糊表述。
    {{
        "decision_id": "{record.decision_id}",
        "summary": "string (对决策的简要总结)",
        "evaluation_outcome": {{
            "goal_achieved": "boolean",
            "outcome_comparison": "string (预期与实际结果的详细对比)",
            "unexpected_impact": "string (意外的正面或负面影响)"
        }},
        "rationality_analysis": {{
            "is_optimal": "boolean",
            "alternative_actions": "string (是否存在更好的替代行动)",
            "rationale_critique": "string (对决策理由的详细审查)"
        }},
        "problem_diagnosis": {{
            "problem_type": "string (e.g., 'Failure', 'Suboptimal', 'Inefficiency', 'KnowledgeGap')",
            "root_cause": "string (根本原因的详细描述)"
        }},
        "improvement_suggestions": [
            {{
                "type": "string (e.g., 'KnowledgeUpdate', 'StrategyAdjustment', 'ParameterTuning', 'DataCollection')",
                "description": "string (具体可操作的建议)",
                "priority": "string (e.g., 'high', 'medium', 'low')"
            }}
        ],
        "confidence_score": "float (0.0-1.0，表示审查报告的置信度)"
    }}
    """
    return prompt

# ... (perform_critique method remains the same) ...

示例用法 (进一步验证优化后的Prompt)

if name == "main":

… (MemoryModule, MockLLMClient, SelfCritiqueModule initialization as before) …

memory = MemoryModule()
llm_client = MockLLMClient()
critique_module = SelfCritiqueModule(memory, llm_client, critique_interval=1) # 简化为每1个决策审查一次

decision_example = DecisionRecord(
    timestamp=datetime.datetime.now(),
    decision_id=str(uuid.uuid4()),
    agent_goal="Optimize energy consumption for data processing",
    environment_observation={
        "data_volume_gb": 1000,
        "current_server_load_percent": 70,
        "available_servers": ["server_A", "server_B", "server_C"],
        "server_specs": {
            "server_A": {"power_kw": 0.5, "max_load": 80, "cost_per_hour": 1.0},
            "server_B": {"power_kw": 0.3, "max_load": 60, "cost_per_hour": 0.7},
            "server_C": {"power_kw": 0.8, "max_load": 100, "cost_per_hour": 1.5}
        }
    },
    agent_internal_state={
        "energy_optimization_strategy": "cost_priority",
        "current_energy_budget_usd": 50
    },
    action_taken={
        "action_type": "allocate_processing",
        "server_to_use": "server_A",
        "estimated_processing_time_min": 60
    },
    expected_outcome={
        "total_cost_usd": 1.0,
        "processing_completed": True,
        "final_server_load_A": 75
    },
    decision_rationale="Server A balances power efficiency and cost, suitable for current load and budget."
)
memory.add_record(decision_example)
# 模拟实际结果，假设服务器A超载，处理时间延长，导致成本增加
decision_example.actual_outcome = {
    "total_cost_usd": 1.5,
    "processing_completed": True,
    "final_server_load_A": 95, # 实际超载
    "actual_processing_time_min": 90, # 处理时间延长
    "warning_message": "Server A overloaded during processing"
}

if critique_module.should_critique():
    print("n--- Triggering Self-Critique with optimized prompt ---")
    critique_results = critique_module.perform_critique()
    if critique_results:
        print("nOptimized Critique Results (for decision_example):")
        for dec_id, report in critique_results.items():
            print(f"  Decision {dec_id[:8]}: {report.get('problem_diagnosis', {}).get('problem_type')} - {report.get('problem_diagnosis', {}).get('root_cause')}")
            for suggestion in report.get('improvement_suggestions', []):
                print(f"    Suggestion ({suggestion.get('type')}): {suggestion.get('description')}")

通过优化 Prompt，我们可以更有效地引导 LLM 进行深度分析，并以期望的结构化格式返回结果，这对于后续的知识更新模块至关重要。

### 知识更新与 Agent 改进

自我审查模块生成了诊断报告和改进建议，但这些信息本身并不能改变 Agent 的行为。知识更新模块的职责就是将这些抽象的建议转化为 Agent 实际可执行的改变。

**审查结果的输出形式：**

*   **错误报告 (Error Report):** 详细描述问题、根本原因和影响。
*   **改进建议 (Improvement Suggestions):** 具体、可操作的建议。

**如何将建议融入 Agent 的未来行为：**

1.  **规则库更新 (Rule-base Update):**
    *   **适用场景:** 基于规则的决策系统，或当建议是添加、修改、删除某个判断条件或行动规则时。
    *   **示例:** 如果审查发现“Agent 在交通拥堵时选择了最短路径导致延误”，建议可能是“添加规则：当交通拥堵超过阈值时，优先考虑避开拥堵区域的长路径，而非最短路径”。
    *   **实现:** 直接修改 Agent 内部存储的规则集（例如，一个 Python 字典、配置文件或一个简单的专家系统）。

2.  **内部模型调整 (Internal Model Adjustment):**
    *   **适用场景:** Agent 依赖于内部世界模型、预测模型或效用函数来做出决策。
    *   **示例:** 如果审查发现“Agent 低估了某个操作的能量消耗”，建议可能是“调整能量消耗预测模型中的相关参数”。
    *   **实现:** 修改模型参数、重新训练部分模型，或更新模型所依赖的数据。

3.  **策略学习 (Policy Learning):**
    *   **适用场景:** 强化学习 Agent，其行为由一个策略（Policy）决定。
    *   **示例:** 如果审查发现“Agent 在某个状态下选择了次优动作”，建议可能是“在那个状态下，降低该动作的Q值，或提高其他动作的Q值”。
    *   **实现:** 可以在 Agent 的学习循环中引入一个“元学习”步骤，根据审查结果微调策略，或者生成新的训练数据来引导策略学习过程。

4.  **Prompt 优化 (Prompt Optimization):**
    *   **适用场景:** 基于 LLM 的 Agent。
    *   **示例:** 如果审查发现“Agent 经常忘记在规划时考虑用户偏好”，建议可能是“修改 Agent 的系统 Prompt，在规划指引中明确强调‘务必将用户偏好作为首要考虑因素’”。
    *   **实现:** 修改 Agent 的主要 Prompt 或为其提供更多的 Few-shot 示例。

5.  **数据收集/探索策略调整 (Data Collection/Exploration Strategy Adjustment):**
    *   **适用场景:** Agent 缺乏必要信息导致决策失误。
    *   **示例:** 如果审查发现“Agent 对某个区域的实时交通信息不足”，建议可能是“调整感知模块，增加对该区域交通数据的查询频率和来源”。
    *   **实现:** 修改感知模块的配置或规划模块的探索策略。

**实现示例：**

我们来创建一个 `KnowledgeUpdateModule`，它将根据审查报告中的建议，尝试更新 Agent 的内部规则。

```python
import json
from typing import Dict, Any, List, Optional

# 假设Agent的决策器有一个简单的规则库
class AgentDecisionMaker:
    def __init__(self):
        self.rules: Dict[str, Any] = {
            "path_selection_strategy": "shortest_path", # 默认策略
            "energy_threshold_for_recharge": 20,
            "traffic_avoidance_threshold": {"level": "medium", "max_delay_minutes": 5}
        }
        self.knowledge_base: Dict[str, Any] = {
            "known_obstacles": ["obstacle_A_at_path1"],
            "traffic_data_source_priority": ["realtime", "historical"]
        }
        print(f"DecisionMaker: Initialized with rules: {self.rules}")

    def make_decision(self, observation: Dict[str, Any], goal: str) -> Dict[str, Any]:
        """
        模拟Agent的决策逻辑，基于其内部规则。
        """
        # 简化决策逻辑
        if goal == "Navigate to destination X":
            traffic_level = observation.get("traffic_data", {}).get("path1", "low")
            if self.rules["path_selection_strategy"] == "shortest_path" and traffic_level == "low":
                return {"action_type": "move", "selected_route": "path1", "rationale": "Shortest path, low traffic."}
            elif self.rules["path_selection_strategy"] == "avoid_traffic" and traffic_level == "medium":
                return {"action_type": "move", "selected_route": "path2", "rationale": "Avoiding medium traffic on path1."}
            else:
                return {"action_type": "move", "selected_route": "path1", "rationale": "Default path."}
        elif goal == "Optimize energy consumption for data processing":
            if self.rules["energy_optimization_strategy"] == "cost_priority": # 假设这条规则可以被更新
                return {"action_type": "allocate_processing", "server_to_use": "server_B", "rationale": "Server B has lower cost."}
            else:
                return {"action_type": "allocate_processing", "server_to_use": "server_A", "rationale": "Default server A."}
        return {"action_type": "no_action", "rationale": "No specific goal match."}

    def update_rule(self, rule_name: str, new_value: Any):
        """
        更新Agent的决策规则。
        """
        if rule_name in self.rules:
            self.rules[rule_name] = new_value
            print(f"DecisionMaker: Rule '{rule_name}' updated to '{new_value}'.")
            return True
        else:
            print(f"DecisionMaker: Rule '{rule_name}' not found.")
            return False

    def update_knowledge(self, knowledge_key: str, new_value: Any):
        """
        更新Agent的知识库。
        """
        self.knowledge_base[knowledge_key] = new_value
        print(f"DecisionMaker: Knowledge '{knowledge_key}' updated to '{new_value}'.")
        return True

class KnowledgeUpdateModule:
    """
    负责根据自我审查模块的建议更新Agent的知识和决策策略。
    """
    def __init__(self, decision_maker: AgentDecisionMaker):
        self.decision_maker = decision_maker

    def apply_suggestions(self, critique_report: Dict[str, Any]):
        """
        解析审查报告，并尝试应用其中的改进建议。
        """
        suggestions: List[Dict[str, Any]] = critique_report.get("improvement_suggestions", [])
        if not suggestions:
            print("KnowledgeUpdate: No suggestions to apply.")
            return

        print(f"nKnowledgeUpdate: Applying {len(suggestions)} suggestions...")
        for suggestion in suggestions:
            s_type = suggestion.get("type")
            description = suggestion.get("description")

            if s_type == "KnowledgeUpdate":
                # 示例：假设建议描述中包含要更新的知识和值
                if "map with obstacle information" in description:
                    self.decision_maker.update_knowledge("known_obstacles", ["obstacle_A_at_path1", "new_obstacle_X_on_path1"])
                elif "real-time traffic data integration" in description:
                    self.decision_maker.update_knowledge("traffic_data_source_priority", ["realtime", "sensor_fusion", "historical"])
                else:
                    print(f"KnowledgeUpdate: Unhandled KnowledgeUpdate suggestion: {description}")

            elif s_type == "StrategyAdjustment":
                if "dynamic path re-planning" in description:
                    self.decision_maker.update_rule("path_selection_strategy", "avoid_traffic") # 简化为切换策略
                elif "prioritize real-time traffic data integration" in description:
                    self.decision_maker.update_rule("traffic_data_source_priority", ["realtime", "historical"])
                elif "consider server B for lower cost" in description: # 对应LLM的建议
                    self.decision_maker.rules["energy_optimization_strategy"] = "cost_priority" # 直接修改规则
                    print(f"DecisionMaker: Rule 'energy_optimization_strategy' updated to 'cost_priority'.")
                else:
                    print(f"KnowledgeUpdate: Unhandled StrategyAdjustment suggestion: {description}")

            elif s_type == "ParameterTuning":
                # 更复杂的实现会解析描述来调整具体数值
                print(f"KnowledgeUpdate: ParameterTuning suggestion: {description} (requires specific parsing).")

            else:
                print(f"KnowledgeUpdate: Unrecognized suggestion type: {s_type}")

# 示例用法
if __name__ == "__main__":
    # 初始化模块
    memory = MemoryModule()
    llm_client = MockLLMClient() # 使用MockLLMClient
    decision_maker = AgentDecisionMaker()
    critique_module = SelfCritiqueModule(memory, llm_client, critique_interval=1)
    knowledge_update_module = KnowledgeUpdateModule(decision_maker)

    print("n--- Initial Agent State ---")
    print(f"Decision Maker Rules: {decision_maker.rules}")
    print(f"Decision Maker Knowledge: {decision_maker.knowledge_base}")

    # 模拟一个导致问题的决策
    decision_bad = DecisionRecord(
        timestamp=datetime.datetime.now(),
        decision_id=str(uuid.uuid4()),
        agent_goal="Optimize energy consumption for data processing",
        environment_observation={
            "data_volume_gb": 1000,
            "current_server_load_percent": 70,
            "available_servers": ["server_A", "server_B", "server_C"],
            "server_specs": {
                "server_A": {"power_kw": 0.5, "max_load": 80, "cost_per_hour": 1.0},
                "server_B": {"power_kw": 0.3, "max_load": 60, "cost_per_hour": 0.7},
                "server_C": {"power_kw": 0.8, "max_load": 100, "cost_per_hour": 1.5}
            }
        },
        agent_internal_state={
            "energy_optimization_strategy": "default_strategy", # 初始策略
            "current_energy_budget_usd": 50
        },
        action_taken={
            "action_type": "allocate_processing",
            "server_to_use": "server_A",
            "estimated_processing_time_min": 60
        },
        expected_outcome={
            "total_cost_usd": 1.0,
            "processing_completed": True,
            "final_server_load_A": 75
        },
        decision_rationale="Server A is the default."
    )
    memory.add_record(decision_bad)
    # 模拟实际结果，假设服务器A超载，处理时间延长，导致成本增加
    decision_bad.actual_outcome = {
        "total_cost_usd": 1.5,
        "processing_completed": True,
        "final_server_load_A": 95,
        "actual_processing_time_min": 90,
        "warning_message": "Server A overloaded during processing"
    }

    # 触发自我审查
    if critique_module.should_critique():
        print("n--- Triggering Self-Critique ---")
        critique_results_map = critique_module.perform_critique() # 返回字典
        if critique_results_map:
            for dec_id, report in critique_results_map.items():
                # 将审查报告应用到知识更新模块
                knowledge_update_module.apply_suggestions(report)

    print("n--- Agent State After Update ---")
    print(f"Decision Maker Rules: {decision_maker.rules}")
    print(f"Decision Maker Knowledge: {decision_maker.knowledge_base}")

    # 再次模拟决策，看是否有所改进
    print("n--- Agent making decision again ---")
    new_observation = {
            "data_volume_gb": 1000,
            "current_server_load_percent": 70,
            "available_servers": ["server_A", "server_B", "server_C"],
            "server_specs": {
                "server_A": {"power_kw": 0.5, "max_load": 80, "cost_per_hour": 1.0},
                "server_B": {"power_kw": 0.3, "max_load": 60, "cost_per_hour": 0.7},
                "server_C": {"power_kw": 0.8, "max_load": 100, "cost_per_hour": 1.5}
            }
        }
    new_decision = decision_maker.make_decision(new_observation, "Optimize energy consumption for data processing")
    print(f"New decision made: {new_decision}")
    # 预期现在会选择 server_B 因为策略已更新为 cost_priority
    assert new_decision["server_to_use"] == "server_B"

这个 KnowledgeUpdateModule 相对简单，它通过字符串匹配来解析 LLM 提供的建议，并调用 AgentDecisionMaker 的方法来更新规则或知识。在实际应用中，这部分会更加复杂，可能需要更精细的语义解析，甚至一个专门的“知识表示”层来管理 Agent 的信念和规则。

一个完整的元认知 Agent 示例流程

现在，让我们将所有组件整合起来，构建一个端到端的元认知 Agent。我们将模拟一个简单的场景：一个资源调度 Agent，负责将计算任务分配给不同的服务器，目标是最小化成本。

场景设定：
Agent 需要将一个计算任务分配给 server_A 或 server_B。

  • server_A: 成本较高，但处理能力较强（假设）。
  • server_B: 成本较低，但处理能力较弱（假设）。
    初始策略是默认使用 server_A。

期望的元认知流程：

  1. Agent 首次运行，将任务分配给 server_A。
  2. 由于某些原因（例如，server_A 实际负载过高，导致成本意外增加），该决策被标记为次优。
  3. 自我审查模块被触发，分析该决策，并建议“优先考虑成本更低的 server_B”。
  4. 知识更新模块根据建议，修改 Agent 的策略。
  5. Agent 再次遇到类似任务时，会基于新的策略选择 server_B。
import datetime
import uuid
import json
import time
from typing import Any, Dict, List, Optional

# --- 1. DecisionRecord and MemoryModule (from previous sections) ---
class DecisionRecord:
    def __init__(self,
                 timestamp: datetime.datetime,
                 decision_id: str,
                 agent_goal: str,
                 environment_observation: Dict[str, Any],
                 agent_internal_state: Dict[str, Any],
                 action_taken: Dict[str, Any],
                 expected_outcome: Dict[str, Any],
                 actual_outcome: Optional[Dict[str, Any]] = None,
                 decision_rationale: Optional[str] = None,
                 critique_report: Optional[Dict[str, Any]] = None):
        self.timestamp = timestamp
        self.decision_id = decision_id
        self.agent_goal = agent_goal
        self.environment_observation = environment_observation
        self.agent_internal_state = agent_internal_state
        self.action_taken = action_taken
        self.expected_outcome = expected_outcome
        self.actual_outcome = actual_outcome
        self.decision_rationale = decision_rationale
        self.critique_report = critique_report

    def __repr__(self):
        return (f"DecisionRecord(ID={self.decision_id[:8]}, Goal='{self.agent_goal}', "
                f"Action={self.action_taken}, Outcome={self.actual_outcome})")

    def to_dict(self):
        return {
            "timestamp": self.timestamp.isoformat(),
            "decision_id": self.decision_id,
            "agent_goal": self.agent_goal,
            "environment_observation": self.environment_observation,
            "agent_internal_state": self.agent_internal_state,
            "action_taken": self.action_taken,
            "expected_outcome": self.expected_outcome,
            "actual_outcome": self.actual_outcome,
            "decision_rationale": self.decision_rationale,
            "critique_report": self.critique_report
        }

class MemoryModule:
    def __init__(self):
        self.decision_history: List[DecisionRecord] = []

    def add_record(self, record: DecisionRecord):
        self.decision_history.append(record)
        print(f"Memory: Added new decision record {record.decision_id[:8]}.")

    def get_records(self, count: Optional[int] = None, since: Optional[datetime.datetime] = None) -> List[DecisionRecord]:
        records = self.decision_history
        if since:
            records = [r for r in records if r.timestamp >= since]
        if count:
            records = records[-count:]
        return records

    def update_record_critique(self, decision_id: str, critique_report: Dict[str, Any]):
        for record in self.decision_history:
            if record.decision_id == decision_id:
                record.critique_report = critique_report
                print(f"Memory: Updated critique for decision {decision_id[:8]}.")
                return
        print(f"Memory: Decision {decision_id[:8]} not found for critique update.")

# --- 2. MockLLMClient (from previous sections) ---
class MockLLMClient:
    def generate_response(self, prompt: str, temperature: float = 0.7) -> str:
        # print(f"n--- MockLLM Input Prompt ---n{prompt}n--- End MockLLM Input ---")
        # 针对资源调度Agent的模拟响应
        if "server_A overloaded" in prompt or "total_cost_usd" in prompt and "1.5" in prompt:
            return json.dumps({
                "decision_id": "mock_id", # This will be replaced by actual ID later
                "summary": "Agent allocated task to Server A, but it overloaded, leading to higher cost and processing time.",
                "evaluation_outcome": {
                    "goal_achieved": True,
                    "outcome_comparison": "Actual cost ($1.5) was higher than expected ($1.0), and actual processing time (90min) was longer than expected (60min). Server A reached 95% load, indicating overload.",
                    "unexpected_impact": "Unexpected cost increase and time delay due to server overload."
                },
                "rationality_analysis": {
                    "is_optimal": False,
                    "alternative_actions": "Agent could have chosen Server B, which has lower cost and might have handled the load better if Server A was near capacity.",
                    "rationale_critique": "The rationale 'Server A is the default' is insufficient. Agent should consider real-time server load and cost-efficiency more rigorously."
                },
                "problem_diagnosis": {
                    "problem_type": "Suboptimal",
                    "root_cause": "Lack of dynamic server load consideration and insufficient cost-optimization strategy in default decision-making."
                },
                "improvement_suggestions": [
                    {
                        "type": "StrategyAdjustment",
                        "description": "Adjust resource allocation strategy to prioritize lower-cost servers (e.g., Server B) if their capacity is sufficient, especially when Server A is already under moderate to high load.",
                        "priority": "high"
                    },
                    {
                        "type": "KnowledgeUpdate",
                        "description": "Integrate real-time server load metrics more prominently into decision parameters.",
                        "priority": "medium"
                    }
                ],
                "confidence_score": 0.95
            })
        else:
            return json.dumps({
                "decision_id": "mock_id",
                "summary": "Agent successfully allocated task and processed it within expected parameters.",
                "evaluation_outcome": {
                    "goal_achieved": True,
                    "outcome_comparison": "Actual outcome matched expected outcome.",
                    "unexpected_impact": "None"
                },
                "rationality_analysis": {
                    "is_optimal": True,
                    "alternative_actions": "No significantly better alternatives identified for this specific instance.",
                    "rationale_critique": "Rationale was sound based on available information."
                },
                "problem_diagnosis": {
                    "problem_type": "None",
                    "root_cause": "N/A"
                },
                "improvement_suggestions": [
                    {
                        "type": "NoChange",
                        "description": "Continue current strategy. Monitor for future optimizations.",
                        "priority": "low"
                    }
                ],
                "confidence_score": 0.99
            })

# --- 3. SelfCritiqueModule (from previous sections) ---
class SelfCritiqueModule:
    def __init__(self, memory: MemoryModule, llm_client: MockLLMClient, critique_interval: int = 1):
        self.memory = memory
        self.llm_client = llm_client
        self.critique_interval = critique_interval
        self._decision_count_since_last_critique = 0

    def should_critique(self) -> bool:
        self._decision_count_since_last_critique += 1
        if self._decision_count_since_last_critique >= self.critique_interval:
            self._decision_count_since_last_critique = 0
            return True
        return False

    def _generate_critique_prompt(self, record: DecisionRecord) -> str:
        prompt = f"""
        你是一名经验丰富的AI Agent行为分析师和优化专家。
        你的任务是深入审查一份AI Agent的单一决策记录，并生成一份结构化的专业审查报告。
        请你扮演一个严格而公正的审查员，旨在帮助Agent识别其决策过程中的潜在缺陷并提供具体改进方案。

        以下是Agent的决策细节，请以JSON格式呈现：
        ```json
        {json.dumps(record.to_dict(), indent=2, ensure_ascii=False)}
    请按照以下步骤进行详细分析：

    **步骤1：总结决策关键信息**
    简要概述Agent的目标、采取的行动及其主要背景。

    **步骤2：评估决策结果**
    比较“expected_outcome”（预期结果）与“actual_outcome”（实际结果）。
    -   目标是否完全达成？
    -   实际结果是否与预期存在显著差异？请量化差异（如果可能）。
    -   是否存在意外的积极或消极后果？

    **步骤3：分析决策合理性**
    基于“environment_observation”（环境观测）、“agent_internal_state”（Agent内部状态）和“decision_rationale”（决策理由），
    -   Agent选择的“action_taken”（采取的行动）在当时是否是最优的？是否存在更好的替代方案？
    -   “decision_rationale”是否充分且逻辑严谨？是否存在信息缺失、误解或推理偏差？

    **步骤4：诊断问题类型与根本原因**
    如果发现问题（如任务失败、次优决策、资源浪费、效率低下等），请明确指出问题类型，并深入分析其根本原因。
    根本原因可能包括但不限于：
    -   **知识缺陷:** Agent对环境模型、规则或自身能力的认知不准确或不完整。
    -   **感知错误:** Agent对环境信息的感知有偏差或遗漏。
    -   **规划/推理缺陷:** Agent的决策逻辑或规划算法存在漏洞。
    -   **执行偏差:** 实际执行与计划有出入。
    -   **环境突变:** 外部环境发生了Agent未预料到的重大变化。

    **步骤5：提出具体的改进建议**
    根据根本原因分析，提出至少一条具体的、可操作的改进建议。这些建议应能直接指导Agent在未来类似情况下如何调整其知识、策略或行为。
    建议可以是：
    -   更新Agent的知识库（例如，添加新规则、修正世界模型）。
    -   调整Agent的决策策略或参数。
    -   建议Agent寻求更多信息或进行额外的环境探索。
    -   优化Agent的Prompt（如果Agent本身基于LLM）。

    请严格以以下JSON格式返回你的完整审查报告。请确保所有字段都存在，并且内容具体、详细、避免模糊表述。
    {{
        "decision_id": "{record.decision_id}",
        "summary": "string",
        "evaluation_outcome": {{
            "goal_achieved": "boolean",
            "outcome_comparison": "string",
            "unexpected_impact": "string"
        }},
        "rationality_analysis": {{
            "is_optimal": "boolean",
            "alternative_actions": "string",
            "rationale_critique": "string"
        }},
        "problem_diagnosis": {{
            "problem_type": "string",
            "root_cause": "string"
        }},
        "improvement_suggestions": [
            {{
                "type": "string",
                "description": "string",
                "priority": "string"
            }}
        ],
        "confidence_score": "float"
    }}
    """
    return prompt

def perform_critique(self) -> Dict[str, Any]:
    records_to_critique = self.memory.get_records(count=self.critique_interval)
    if not records_to_critique:
        print("SelfCritique: No new records to critique.")
        return {}

    print(f"nSelfCritique: Starting critique for {len(records_to_critique)} records.")
    all_critique_results = {}
    for record in records_to_critique:
        prompt = self._generate_critique_prompt(record)
        try:
            llm_response_str = self.llm_client.generate_response(prompt)
            critique_report = json.loads(llm_response_str)
            critique_report["decision_id"] = record.decision_id # 确保ID正确
            self.memory.update_record_critique(record.decision_id, critique_report)
            all_critique_results[record.decision_id] = critique_report
            print(f"SelfCritique: Critiqued decision {record.decision_id[:8]} -> Status: {critique_report.get('problem_diagnosis',{}).get('problem_type', 'N/A')}")
        except json.JSONDecodeError as e:
            print(f"SelfCritique Error: Failed to parse LLM response for {record.decision_id[:8]}: {e}")
            print(f"Raw LLM response: {llm_response_str}")
        except Exception as e:
            print(f"SelfCritique Error: An unexpected error occurred during LLM call for {record.decision_id[:8]}: {e}")
    return all_critique_results

— 4. AgentDecisionMaker and KnowledgeUpdateModule (from previous sections, slightly adapted) —

class AgentDecisionMaker:
def init(self):
self.rules: Dict[str, Any] = {
"resource_allocation_strategy": "default_server_A", # 初始策略
"server_A_load_threshold_for_reconsideration": 85 # 阈值
}
self.knowledge_base: Dict[str, Any] = {
"server_A_cost_per_task": 1.0,
"server_B_cost_per_task": 0.7,
"server_A_capacity_max_load": 90,
"server_B_capacity_max_load": 70
}
print(f"DecisionMaker: Initialized with rules: {self.rules}")

def make_decision(self, observation: Dict[str, Any], goal: str) -> Dict[str, Any]:
    task_type = observation.get("task_type")
    current_load_A = observation.get("server_A_current_load", 0)
    current_load_B = observation.get("server_B_current_load", 0)

    if goal == "Allocate task to minimize cost":
        if self.rules["resource_allocation_strategy"] == "default_server_A":
            if current_load_A < self.rules["server_A_load_threshold_for_reconsideration"]:
                return {"action_type": "allocate_task", "server_to_use": "server_A", "expected_cost": self.knowledge_base["server_A_cost_per_task"], "rationale": "Default strategy, Server A available."}
            else: # Fallback if default server is too loaded
                return {"action_type": "allocate_task", "server_to_use": "server_B", "expected_cost": self.knowledge_base["server_B_cost_per_task"], "rationale": "Server A overloaded, fallback to Server B."}
        elif self.rules["resource_allocation_strategy"] == "cost_priority_B_first":
            if current_load_B < self.knowledge_base["server_B_capacity_max_load"]: # 优先考虑B
                return {"action_type": "allocate_task", "server_to_use": "server_B", "expected_cost": self.knowledge_base["server_B_cost_per_task"], "rationale": "Cost priority strategy, Server B available and cheaper."}
            elif current_load_A < self.knowledge_base["server_A_capacity_max_load"]: # B不可用时考虑A
                return {"action_type": "allocate_task", "server_to_use": "server_A", "expected_cost": self.knowledge_base["server_A_cost_per_task"], "rationale": "Cost priority strategy, Server B

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注