各位同仁，欢迎来到今天的技术讲座。今天我们将深入探讨一个在构建高智能、自适应Agent系统中日益关键的概念：Reflection Node。在当今快速发展的AI领域，我们不仅追求Agent能够高效地执行任务，更渴望它们具备自我学习、自我修正的能力。而“反思节点”正是实现这一宏伟目标的核心机制之一。
设想一下，一个Agent完成了一项复杂任务，例如编写一段代码、分析一组数据，或者与用户进行了一次深入的对话。任务结束后，我们通常会评估其结果。但仅仅知道结果是成功还是失败，对于Agent的长期成长是远远不够的。我们需要Agent能够理解为什么成功，为什么失败，并从中提炼出可操作的经验，指导未来的行为。这正是Reflection Node的职责所在。
1. Agent工作流中的反思：为何如此关键？
在传统的软件工程中，我们构建的系统往往是确定性的：给定输入，经过一系列预设逻辑，产生输出。然而，现代Agent系统，尤其是基于大型语言模型（LLMs）的Agent，运行在一个高度不确定、动态变化的环境中。它们需要处理开放域的问题，适应未知的输入，并在复杂的多步骤推理中做出决策。
一个典型的Agent工作流可能包含以下核心组件：
- Planner (规划器): 根据任务目标和当前环境，制定一系列行动步骤。
- Executor (执行器): 按照规划器的指令执行具体动作，例如调用工具、查询知识库、生成文本等。
- Observer (观察器): 监控执行器的动作和环境反馈，收集信息。
- Memory (记忆库): 存储Agent的经验、知识、过去的对话历史和反思结果。
在没有Reflection Node的情况下，Agent的工作流通常是线性的或循环的。每次迭代，Agent会根据新的观察和记忆重新规划并执行。但这种方式缺乏一个关键的环节：对整个执行过程和结果进行深度的批判性分析。Agent可能会重复同样的错误，或者无法识别并利用成功的模式。
Reflection Node的引入，打破了这种局限性。它为Agent提供了一个暂停、回顾、分析并提炼经验的机会。这不仅仅是简单的日志记录，而是一种高级认知功能，旨在：
- 识别成功与失败的根源： 找出导致任务成功或失败的特定步骤、决策或外部因素。
- 提炼可复用的知识： 将具体的经验转化为通用的规则、策略或模式，存储到记忆库中。
- 优化未来规划： 根据反思结果，调整规划器的策略，避免重复犯错，或更有效地利用成功经验。
- 提高韧性和适应性： 使Agent在面对新挑战时能够更快地学习和适应。
简而言之，Reflection Node赋予了Agent自我学习和自我进化的能力，是构建真正智能、自主Agent系统的必不可少的一环。
2. 反思节点的定义与核心职能
Reflection Node，顾名思义，是Agent工作流中专门负责对过去行为、观察和结果进行批判性分析，并从中学习的组件。它通常在Agent完成一个任务、一个子任务，或者一个迭代周期之后被激活。
其核心职能可以概括为以下几点：
| 职能类别 | 具体描述 引言：理解Agent的局限性与反思的必要性
各位同仁，亲爱的编程专家们，大家好！
我们生活在一个智能系统高速发展的时代，而其中最令人着迷的便是Agent系统。这些Agent，在某种程度上，模拟着人类的决策与行动，它们能够感知环境、规划行动、执行任务，甚至与我们进行复杂的交互。然而，当我们深入观察这些Agent的工作机制时，会发现一个普遍存在的挑战：如何让Agent不仅仅是执行者，更是学习者？
想象一个场景：你构建了一个Agent，它的任务是自动生成和优化Python代码。在一次运行中，Agent接收了一个需求，规划了一系列步骤（例如，分析需求、选择库、编写骨架代码、实现逻辑、添加测试），并最终输出了一段代码。如果这段代码通过了所有测试，我们自然会认为任务成功。但如果失败了呢？或者说，即使成功了，这段代码是否足够优雅、高效、符合最佳实践？
传统的Agent工作流，在任务结束后，往往就戛然而止了。Agent交付了它的产物，然后等待下一个指令。它可能积累了大量的“成功”或“失败”的经验，但如果没有一个机制去深入剖析这些经验，去理解为什么会成功、为什么会失败，那么下一次面对类似的问题时，它很可能会重复同样的错误，或者无法复用已有的成功模式。
这正是人类学习与成长的核心：我们不仅仅是行动者，更是反思者。我们会回顾过去的决策，分析其后果，总结经验教训，并据此调整未来的行为。这种能力，对于Agent而言，同样至关重要。它能让Agent从简单的“执行机器”进化为“自适应学习系统”。
而今天我们讲座的主题——Reflection Node，正是为Agent赋予这种“反思能力”的关键组件。它在Agent工作流即将结束时，提供了一个关键的暂停点，让Agent能够回顾整个任务的执行过程，评估其得失，并从中提炼出宝贵的经验，以指导其未来的行动。
2. 反思节点的本质：自我审视与经验提炼
2.1 什么是Reflection Node？
从架构角度看，Reflection Node是Agent工作流中的一个特殊阶段或模块。它并非直接参与任务的执行，而是在任务执行完毕或一个重要的阶段性目标达成后被激活。它的核心职责是接收并分析Agent在整个任务周期内生成的所有相关信息（包括原始任务目标、Agent的决策路径、执行结果、环境反馈等），然后对这些信息进行批判性评估，从而生成有价值的“反思报告”或“学习成果”。
2.2 反思节点的核心工作原理
Reflection Node的工作流程可以抽象为以下几个核心阶段：
- 数据收集 (Data Collection): 汇集任务执行过程中产生的所有相关数据。
- 自我批判与分析 (Self-Critique & Analysis): 基于收集到的数据，Agent对自己的表现进行深入剖析。
- 知识生成与提炼 (Knowledge Generation & Refinement): 从分析中提取出可操作的经验教训和改进策略。
- 记忆更新与反馈 (Memory Update & Feedback): 将提炼出的知识存储到Agent的长期记忆中，并可能反馈给规划器或执行器，影响未来的行为。
这种机制使得Agent能够超越简单的指令遵循，进入一个自我迭代、自我优化的学习循环。它模仿了人类在完成一个项目后进行“复盘”的过程，是Agent实现真正自主性和智能的关键一步。
3. 反思节点的输入：Agent的“记忆”与“经历”
一个高效的Reflection Node必须拥有全面、结构化的输入数据，这些数据构成了Agent“反思”的基础。如同一个人要反思一次旅行，他需要知道旅行的初衷、路线、遇到的问题、拍下的照片和最终的感受。对于Agent而言，这些输入通常包括：
-
原始任务上下文 (Original Task Context):
- 初始目标/需求 (Initial Goal/Requirement): Agent最初被赋予的任务是什么？它被期望实现什么？
- 约束条件 (Constraints): 任务是否有时间、资源、技术栈或安全等方面的限制？
- 初始环境状态 (Initial Environment State): 任务开始时，Agent所处的环境情况。
-
执行轨迹 (Execution Trace):
- 行动序列 (Sequence of Actions): Agent在任务执行过程中采取的每一个具体行动（例如，调用了哪个工具、发出了什么API请求、生成了什么中间文本、修改了哪个文件）。
- 观察与反馈 (Observations & Feedback): 每次行动后，Agent从环境中获得的观察结果或工具返回的反馈。这包括成功消息、错误信息、数据查询结果等。
- 中间推理步骤 (Intermediate Reasoning Steps): 如果Agent使用了LLM进行多步推理，这些推理过程中的中间思考和决策。
- 耗时与资源消耗 (Time & Resource Consumption): 记录每个步骤的执行时间，以及可能消耗的计算或API资源。
-
最终结果与输出 (Final Output & Result):
- 任务最终产物 (Final Artifact): 例如，完成的代码文件、生成的报告、更新的数据库记录等。
- 任务状态 (Task Status): 任务是成功完成、部分完成还是彻底失败？
- 关键指标 (Key Metrics): 如果有预定义的评估指标（例如，代码测试通过率、响应时间、准确度），这些指标的值。
-
评估结果 (Evaluation Results):
- 外部评估 (External Evaluation): 如果有外部系统或人工对Agent的最终输出进行了评估，其评估报告或评分。
- 内部自评估 (Internal Self-Evaluation): Agent自身在完成任务后，根据预设标准对输出进行的初步检查。
将这些输入数据结构化，对于后续的分析至关重要。我们可以用JSON或Pydantic模型来表示这些信息，以便LLM能够更好地理解和处理。
例如，一个简单的执行轨迹的结构可能如下：
[
{
"step_id": 1,
"action_type": "plan",
"details": "Agent decided to break down the task into sub-tasks: code generation, testing, and documentation.",
"timestamp": "2023-10-27T10:00:05Z",
"tool_used": None,
"output": "Plan: Generate code -> Write tests -> Run tests -> Document code."
},
{
"step_id": 2,
"action_type": "tool_use",
"details": "Calling code_generator tool to generate initial Python function.",
"timestamp": "2023-10-27T10:01:10Z",
"tool_used": "code_generator",
"tool_input": {
"prompt": "Write a Python function to calculate the factorial of a number."
},
"output": {
"status": "success",
"code": "def factorial(n):n if n == 0: return 1n else: return n * factorial(n-1)"
}
},
{
"step_id": 3,
"action_type": "tool_use",
"details": "Calling test_runner tool to execute generated code with test cases.",
"timestamp": "2023-10-27T10:02:30Z",
"tool_used": "test_runner",
"tool_input": {
"code": "def factorial(n):n if n == 0: return 1n else: return n * factorial(n-1)",
"tests": [
{"input": 0, "expected_output": 1},
{"input": 1, "expected_output": 1},
{"input": 5, "expected_output": 120},
{"input": -1, "expected_output": None} # This test case expects an error or specific handling
]
},
"output": {
"status": "failure",
"test_results": [
{"input": 0, "actual_output": 1, "passed": True},
{"input": 1, "actual_output": 1, "passed": True},
{"input": 5, "actual_output": 120, "passed": True},
{"input": -1, "actual_output": "RecursionError", "passed": False, "error_message": "maximum recursion depth exceeded"}
],
"summary": "1 out of 4 tests failed."
}
}
# ... more steps
]4. 反思节点的处理流程：从观察到洞察
有了丰富的输入数据，Reflection Node便可以启动其核心处理逻辑。这个过程通常由一个强大的语言模型（LLM）驱动，它能够理解复杂的上下文、进行逻辑推理并生成连贯的文本。
4.1 自我批判与问题识别
这是反思的第一步，也是最关键的一步。LLM需要像一个严谨的审查员一样，审视Agent的每一步行动，并与任务目标、预期结果进行对比。
核心问题：
- 任务是否成功完成？ 如果失败了，失败在哪里？
- 哪些步骤是成功的？ 它们为何成功？能否推广这种成功模式？
- 哪些步骤是失败的？ 它们为何失败？是规划问题？执行错误？工具使用不当？还是对环境理解有误？
- 是否存在更优的路径？ Agent是否走了弯路？是否有冗余或低效的步骤？
- Agent是否充分利用了其工具和知识？ 是否有遗漏或误用？
LLM提示工程示例 (Critique Prompt):
critique_prompt_template = """
你是一个高度智能的Agent自我批判与反思系统。你的任务是分析Agent的过去行为轨迹、任务目标和最终结果，以识别成功之处、失败之处及其根本原因。
---
**原始任务目标:**
{task_context}
**Agent执行轨迹:**
{execution_trace}
**最终结果与评估:**
{final_result}
---
请根据以上信息，执行以下步骤：
1. **任务完成度评估:**
* Agent是否成功完成了原始任务目标？请清晰说明“是”或“否”，并简要说明理由。
* 如果任务失败或部分失败，请指出具体失败点。
2. **成功点分析:**
* 回顾执行轨迹，识别出Agent表现出色，或做出正确决策的关键步骤。
* 解释这些成功步骤背后的原因，以及Agent采取的策略或利用的工具为何有效。
3. **失败点及问题分析:**
* 回顾执行轨迹，识别出导致任务失败、效率低下或产生次优结果的关键步骤或决策。
* 对于每个识别出的问题点，深入分析其根本原因。考虑以下可能性：
* **规划失误:** 初始规划是否合理？是否遗漏了关键步骤或考虑不周？
* **执行错误:** Agent在调用工具或生成内容时是否出现了错误？是否误解了工具的用法或环境反馈？
* **工具选择/使用不当:** 是否选择了不合适的工具？或者工具的参数配置有误？
* **知识/信息不足:** Agent是否缺乏完成任务所需的关键知识或信息？
* **环境理解偏差:** Agent是否对环境的某些方面产生了误解？
* **LLM推理局限:** LLM在生成文本或逻辑推理时是否出现“幻觉”或逻辑谬误？
* 请提供具体的执行轨迹步骤ID和相关输出作为证据。
4. **效率与优化潜力:**
* Agent的执行路径是否存在冗余、重复或不必要的步骤？
* 是否有更简洁、高效的方式来达成目标？
* 在资源利用（如LLM调用次数、工具调用次数）方面是否存在优化空间？
请以结构化的JSON格式输出你的分析结果，例如：
{{
"task_completion": "成功/失败/部分成功",
"completion_reason": "简要说明",
"success_points": [
{{
"step_id": 2,
"description": "成功调用code_generator生成初始代码。",
"reason": "Agent正确理解了需求，并选择了合适的工具。"
}}
],
"failure_points": [
{{
"step_id": 3,
"description": "测试用例失败，未能处理负数输入。",
"root_cause": "规划阶段未充分考虑边缘情况，代码生成器也未被明确指示处理负数，导致测试失败。",
"type": "规划失误/代码缺陷"
}}
],
"optimization_potential": [
{{
"area": "规划",
"suggestion": "在规划阶段增加对边缘情况（如负数、大数）的考虑，并明确要求代码生成器处理这些情况。"
}}
]
}}
"""4.2 提炼经验教训与改进策略
在完成自我批判后，Reflection Node的下一步是将这些批判转化为可操作的经验教训和具体的改进策略。这不仅仅是指出问题，更是提出解决方案。
核心问题：
- 从这次经历中，Agent学到了什么？ 有哪些通用的原则或模式可以被提取出来？
- Agent应该如何改进？ 具体而言，下一次面对类似任务时，它应该怎么做？
- 记忆库中应该添加哪些新的知识？ 这些知识如何帮助Agent在未来做得更好？
LLM提示工程示例 (Lessons Learned & Improvement Prompt):
lessons_prompt_template = """
你是一个Agent学习与策略优化系统。你的任务是根据先前的Agent自我批判报告，提炼出可操作的经验教训，并制定具体的改进策略，以提升Agent在未来执行类似任务时的表现。
---
**Agent自我批判报告:**
{critique_report}
**原始任务目标:**
{task_context}
---
请根据以上批判报告和原始任务目标，执行以下步骤：
1. **总结核心经验教训 (Lessons Learned):**
* 从成功点和失败点中，提取出本次任务中最核心的3-5条经验教训。这些教训应该是通用的、可迁移的原则或模式，而不仅仅是针对本次任务的细节。
* 例如：“在生成代码前，应始终考虑并明确边缘测试用例。” 或 “对于复杂需求，分阶段验证比一次性生成完整方案更可靠。”
2. **制定具体改进策略 (Improvement Strategies):**
* 针对批判报告中识别出的每个问题点和优化潜力，提出1-3条具体的、可执行的改进策略。
* 这些策略应该明确指出Agent的哪个组件（规划器、执行器、工具选择、记忆使用等）需要如何调整。
* 例如：
* **对于规划器:** “在规划生成代码任务时，增加一个子步骤‘明确边缘测试用例’。”
* **对于执行器/工具调用:** “当调用code_generator时，除了需求描述，额外传入明确的测试用例和其预期行为，包括边缘情况。”
* **对于记忆库:** “将‘处理负数输入的阶乘函数实现’作为一个示例，并注明其错误处理方式。”
* **对于评估器:** “引入更严格的代码质量检查工具，如linter，作为自动评估的一部分。”
3. **记忆更新建议 (Memory Update Suggestions):**
* 建议将哪些关键信息、经验教训或最佳实践以何种形式（例如，新的知识片段、更新的策略规则）存储到Agent的长期记忆库中。
* 这些信息应该能够被Agent在未来的任务中检索和利用。
请以结构化的JSON格式输出你的结果，例如：
{{
"lessons_learned": [
"经验教训1：在需求分析阶段，必须主动识别并处理所有可能的边缘情况，特别是数值计算任务中的负数、零、大数等。",
"经验教训2：当发现工具（如code_generator）未能完全满足需求时，应考虑提供更详细的上下文或更具体的约束条件，而非直接接受其首次输出。",
"经验教训3：测试驱动开发（TDD）的理念对Agent同样适用，先定义全面的测试用例有助于指导代码生成和验证。"
],
"improvement_strategies": [
{{
"target_component": "Planner",
"strategy": "在接收到代码生成任务后，规划器应首先生成一组全面的测试用例（包括正常、边界、异常情况），然后再指导代码生成器基于这些测试用例进行开发。",
"priority": "High"
}},
{{
"target_component": "Executor/CodeGenerator Tool",
"strategy": "修改code_generator工具的调用接口，允许在生成代码时直接传入测试用例，并指示其代码必须通过这些测试。如果无法通过，则要求其进行自我修正。",
"priority": "High"
}},
{{
"target_component": "Memory",
"strategy": "将本次任务中‘处理负数输入的阶乘函数’的正确实现（例如，抛出ValueError）作为最佳实践存储，以便在未来类似数值处理任务中参考。",
"priority": "Medium"
}}
],
"memory_update_content": [
"知识片段: '阶乘函数处理负数最佳实践：对于负数输入，应抛出ValueError或返回特定错误码，而非进入无限递归。'",
"策略规则: '在规划任何数值计算类函数时，必须明确考虑并处理零、负数、浮点数等特殊输入。'"
]
}}
"""5. 架构集成与技术选型
5.1 反思节点在Agent架构中的位置
Reflection Node通常位于Agent工作流的末端，但其输出会循环反馈给Agent的其他核心组件。
+----------------+ +----------------+ +----------------+
| Task Manager |------>| Planner |------>| Executor |
+----------------+ +-------^--------+ +-------v--------+
| |
| v
+----------------+ +----------------+ +----------------+
| Memory |<------| Reflection Node|<------| Observer |
| (Knowledge Base)|----->| (LLM-powered) |<------| (Trace Collector)|
+----------------+ +----------------+ +----------------+
^ |
| |
+------------------------+
(Execution Trace, Results)- Planner (规划器): Reflection Node生成的改进策略会直接影响规划器如何制定未来的行动计划。例如，如果反思建议在特定任务前进行数据预处理，规划器就会将此步骤纳入其规划。
- Executor (执行器): 如果反思建议优化某个工具的使用方式，执行器在后续调用该工具时会采纳这些建议。
- Memory (记忆库): 这是Reflection Node最重要的输出去向。提炼出的经验教训、改进策略、最佳实践等都会以结构化或非结构化的形式（如知识图谱、向量嵌入、文本片段）存储在记忆库中，供Agent在未来的任务中检索和利用。
- Observer (观察器): 观察器在收集执行轨迹时，需要确保其收集的数据足够详细和结构化，以满足Reflection Node的需求。
5.2 技术栈选择
-
核心推理引擎：大型语言模型 (LLMs):
- GPT-4, Claude, Gemini等模型是Reflection Node的核心。它们强大的理解、推理和生成能力是实现深度反思的基础。
- 通过精心设计的Prompt Engineering，我们可以引导LLM进行自我批判、问题分析、经验总结和策略生成。
-
Agent编排框架:
- LangChain, LlamaIndex: 这些框架提供了构建Agent的模块化组件（LLM接口、工具集成、记忆管理等），可以方便地将Reflection Node集成到Agent工作流中。它们也提供了Prompt模板和链式调用等功能，简化了LLM的交互。
- 自定义框架: 对于更复杂的场景，可能需要根据具体需求构建定制化的Agent编排逻辑。
-
记忆存储:
- 向量数据库 (Vector Databases): 如Pinecone, Weaviate, Milvus等，用于存储反思后生成的知识片段的向量嵌入。Agent可以通过语义相似性检索相关经验。
- 关系型数据库/NoSQL数据库: 用于存储结构化的反思报告、任务日志、策略规则等。
- 文本文件/JSON文件: 简单场景下，也可以直接存储为文件。
-
工具调用:
- Reflection Node本身不直接调用外部工具进行任务执行，但它可能会分析Agent对工具的调用记录，并提出优化工具使用方式的建议。
- Agent执行过程中使用的工具（如代码解释器、文件系统操作、API客户端等）是Reflection Node分析的重要对象。
6. 实施Reflection Node：一个Python编程Agent的例子
我们来通过一个具体的Python编程Agent的例子，演示如何实现一个Reflection Node。这个Agent的目标是根据用户需求编写Python函数，并确保其正确性。
6.1 Agent工作流概览
- 接收任务: 用户提供一个Python函数的需求。
- 规划 (Plan): Agent根据需求制定编写代码和测试的计划。
- 执行 (Execute):
- 生成代码: 调用LLM或代码生成工具生成初步代码。
- 生成测试: 根据需求和代码生成测试用例。
- 运行测试: 在沙箱环境中执行代码和测试用例。
- 调试/迭代 (可选): 如果测试失败，Agent可能进入内部循环进行调试和代码修正。
- 反思 (Reflect): 无论任务成功或失败，Agent都会对整个过程进行反思。
- 记忆更新 (Memory Update): 将反思结果存储到记忆库。
6.2 核心组件的Python实现
我们将使用一个简化的LLM接口和Tool接口来模拟实际的LLM调用和工具执行。
import json
import uuid
import datetime
from typing import List, Dict, Any, Optional
# --- 模拟LLM接口 ---
class MockLLM:
def __init__(self, model_name="gpt-4"):
self.model_name = model_name
def generate(self, prompt: str, temperature: float = 0.7, max_tokens: int = 1024) -> str:
# 这是一个模拟的LLM生成函数，实际中会调用OpenAI/Anthropic等API
print(f"n--- LLM Input for {self.model_name} ---n{prompt}n--- End LLM Input ---")
# 在实际应用中，这里会根据prompt内容调用真实的LLM API
# 为了演示，我们返回一个占位符或预设的响应
if "critique_prompt_template" in prompt:
return json.dumps({
"task_completion": "部分成功",
"completion_reason": "核心功能实现，但未能处理边缘情况（负数输入）。",
"success_points": [
{
"step_id": 2,
"description": "成功调用code_generator生成初始代码。",
"reason": "Agent正确理解了需求，并选择了合适的工具。"
}
],
"failure_points": [
{
"step_id": 3,
"description": "测试用例失败：未能处理负数输入。",
"root_cause": "规划阶段未充分考虑边缘情况，代码生成器也未被明确指示处理负数，导致测试失败。",
"type": "规划失误/代码缺陷"
}
],
"optimization_potential": [
{
"area": "规划",
"suggestion": "在规划阶段增加对边缘情况（如负数、大数）的考虑，并明确要求代码生成器处理这些情况。"
}
]
}, indent=2, ensure_ascii=False)
elif "lessons_prompt_template" in prompt:
return json.dumps({
"lessons_learned": [
"经验教训1：在需求分析阶段，必须主动识别并处理所有可能的边缘情况，特别是数值计算任务中的负数、零、大数等。",
"经验教训2：当发现工具（如code_generator）未能完全满足需求时，应考虑提供更详细的上下文或更具体的约束条件，而非直接接受其首次输出。",
"经验教训3：测试驱动开发（TDD）的理念对Agent同样适用，先定义全面的测试用例有助于指导代码生成和验证。"
],
"improvement_strategies": [
{
"target_component": "Planner",
"strategy": "在接收到代码生成任务后，规划器应首先生成一组全面的测试用例（包括正常、边界、异常情况），然后再指导代码生成器基于这些测试用例进行开发。",
"priority": "High"
},
{
"target_component": "Executor/CodeGenerator Tool",
"strategy": "修改code_generator工具的调用接口，允许在生成代码时直接传入测试用例，并指示其代码必须通过这些测试。如果无法通过，则要求其进行自我修正。",
"priority": "High"
}
],
"memory_update_content": [
"知识片段: '阶乘函数处理负数最佳实践：对于负数输入，应抛出ValueError或返回特定错误码，而非进入无限递归。'",
"策略规则: '在规划任何数值计算类函数时，必须明确考虑并处理零、负数、浮点数等特殊输入。'"
]
}, indent=2, ensure_ascii=False)
return "模拟LLM生成结果"
# --- 模拟工具接口 ---
class Tool:
def __init__(self, name: str, description: str):
self.name = name
self.description = description
def run(self, **kwargs) -> Any:
raise NotImplementedError
class CodeGeneratorTool(Tool):
def __init__(self):
super().__init__("code_generator", "根据需求生成Python代码。")
def run(self, prompt: str) -> Dict[str, Any]:
print(f"Executing CodeGeneratorTool with prompt: {prompt}")
# 模拟代码生成，这里假设生成了一个有缺陷的阶乘函数
if "factorial" in prompt.lower():
return {
"status": "success",
"code": "def factorial(n):n if n == 0: return 1n else: return n * factorial(n-1)"
}
return {"status": "success", "code": f"def generated_function():n # {prompt}n pass"}
class TestRunnerTool(Tool):
def __init__(self):
super().__init__("test_runner", "在沙箱环境中运行Python代码和测试用例。")
def run(self, code: str, tests: List[Dict[str, Any]]) -> Dict[str, Any]:
print(f"Executing TestRunnerTool with code:n{code}nand tests: {tests}")
results = []
passed_count = 0
for test in tests:
input_val = test.get("input")
expected_output = test.get("expected_output")
# 模拟执行代码并捕获错误
try:
# 这是一个非常简化的模拟执行，实际中需要更安全的沙箱环境
exec_globals = {}
exec(code, exec_globals)
func = exec_globals[list(exec_globals.keys())[0]] # 假设第一个定义的是目标函数
actual_output = func(input_val)
passed = (actual_output == expected_output)
if passed:
passed_count += 1
results.append({"input": input_val, "expected_output": expected_output, "actual_output": actual_output, "passed": passed})
except RecursionError:
results.append({"input": input_val, "expected_output": expected_output, "actual_output": "RecursionError", "passed": False, "error_message": "maximum recursion depth exceeded"})
except Exception as e:
results.append({"input": input_val, "expected_output": expected_output, "actual_output": str(e), "passed": False, "error_message": str(e)})
summary = f"{passed_count} out of {len(tests)} tests passed."
return {"status": "success" if passed_count == len(tests) else "failure", "test_results": results, "summary": summary}
# --- 记忆库接口 ---
class Memory:
def __init__(self):
self.reflections: List[Dict[str, Any]] = []
self.knowledge_base: List[str] = []
self.strategy_rules: List[str] = []
def add_reflection(self, reflection_data: Dict[str, Any]):
self.reflections.append(reflection_data)
print(f"Added reflection to memory. Total reflections: {len(self.reflections)}")
def update_knowledge_base(self, new_knowledge: List[str]):
self.knowledge_base.extend(new_knowledge)
print(f"Updated knowledge base. Total entries: {len(self.knowledge_base)}")
def update_strategy_rules(self, new_rules: List[str]):
self.strategy_rules.extend(new_rules)
print(f"Updated strategy rules. Total rules: {len(self.strategy_rules)}")
def retrieve_relevant_info(self, query: str, top_k: int = 1) -> List[str]:
# 实际中这里会用向量搜索，这里简化为关键词匹配
relevant_info = [
item for item in self.knowledge_base + self.strategy_rules
if query.lower() in item.lower()
]
return relevant_info[:top_k]
# --- Reflection Node 实现 ---
class ReflectionNode:
def __init__(self, llm: MockLLM, memory: Memory):
self.llm = llm
self.memory = memory
self.critique_prompt_template = critique_prompt_template # 从前面定义导入
self.lessons_prompt_template = lessons_prompt_template # 从前面定义导入
def reflect(self,
task_context: Dict[str, Any],
execution_trace: List[Dict[str, Any]],
final_result: Dict[str, Any]) -> Dict[str, Any]:
print("n--- ACTIVATE REFLECTION NODE ---")
# 1. 生成批判报告
critique_prompt = self.critique_prompt_template.format(
task_context=json.dumps(task_context, indent=2, ensure_ascii=False),
execution_trace=json.dumps(execution_trace, indent=2, ensure_ascii=False),
final_result=json.dumps(final_result, indent=2, ensure_ascii=False)
)
critique_raw = self.llm.generate(critique_prompt)
critique_report = json.loads(critique_raw)
print("n--- CRITIQUE REPORT GENERATED ---")
print(json.dumps(critique_report, indent=2, ensure_ascii=False))
# 2. 提炼经验教训和改进策略
lessons_prompt = self.lessons_prompt_template.format(
critique_report=json.dumps(critique_report, indent=2, ensure_ascii=False),
task_context=json.dumps(task_context, indent=2, ensure_ascii=False)
)
lessons_raw = self.llm.generate(lessons_prompt)
lessons_and_strategies = json.loads(lessons_raw)
print("n--- LESSONS LEARNED & STRATEGIES GENERATED ---")
print(json.dumps(lessons_and_strategies, indent=2, ensure_ascii=False))
# 3. 更新记忆库
self.memory.add_reflection({
"reflection_id": str(uuid.uuid4()),
"timestamp": datetime.datetime.now().isoformat(),
"task_context": task_context,
"execution_trace": execution_trace,
"final_result": final_result,
"critique_report": critique_report,
"lessons_and_strategies": lessons_and_strategies
})
self.memory.update_knowledge_base(lessons_and_strategies.get("memory_update_content", []))
# 实际中，strategy_rules的更新可能需要更复杂的逻辑，例如替换旧规则或合并
self.memory.update_strategy_rules([
s["strategy"] for s in lessons_and_strategies.get("improvement_strategies", [])
])
print("n--- MEMORY UPDATED ---")
return {
"critique": critique_report,
"lessons": lessons_and_strategies
}
# --- Agent 工作流 ---
class CodeAgent:
def __init__(self, llm: MockLLM, memory: Memory, tools: List[Tool]):
self.llm = llm
self.memory = memory
self.tools = {tool.name: tool for tool in tools}
self.execution_trace: List[Dict[str, Any]] = []
self.task_context: Dict[str, Any] = {}
def _record_step(self, action_type: str, details: str, tool_used: Optional[str] = None,
tool_input: Optional[Dict[str, Any]] = None, output: Optional[Any] = None):
self.execution_trace.append({
"step_id": len(self.execution_trace) + 1,
"timestamp": datetime.datetime.now().isoformat(),
"action_type": action_type,
"details": details,
"tool_used": tool_used,
"tool_input": tool_input,
"output": output
})
def run_task(self, requirement: str) -> Dict[str, Any]:
self.task_context = {
"initial_requirement": requirement,
"constraints": "Python 3.x, ensure robustness for edge cases."
}
self.execution_trace = []
final_output = {"status": "failure", "message": "Task not completed."}
self._record_step("start_task", f"Starting task: {requirement}")
# 1. 规划
plan_prompt = f"根据需求 '{requirement}'，制定一个编写Python函数并测试的计划。考虑如何处理边缘情况。历史经验：{self.memory.retrieve_relevant_info('规划')}。"
plan_response = self.llm.generate(plan_prompt, max_tokens=200)
plan = json.loads(plan_response) if plan_response.strip().startswith('{') else {"plan": plan_response}
self._record_step("plan", "Agent formulated a plan.", output=plan)
print(f"n--- Agent Plan ---n{plan.get('plan', plan_response)}n--- End Plan ---n")
# 2. 执行：生成代码
code_gen_prompt = f"根据需求 '{requirement}' 编写一个Python函数。请确保代码的健壮性并考虑边缘情况。历史经验：{self.memory.retrieve_relevant_info('代码生成')}。"
code_generator = self.tools["code_generator"]
code_output = code_generator.run(prompt=code_gen_prompt)
generated_code = code_output.get("code")
self._record_step("tool_use", "Generated initial code.", "code_generator", {"prompt": code_gen_prompt}, code_output)
print(f"n--- Generated Code ---n{generated_code}n--- End Code ---n")
if not generated_code:
final_output["message"] = "Failed to generate code."
self._record_step("end_task", "Task failed due to no code generation.", output=final_output)
return final_output
# 3. 执行：生成并运行测试
test_cases = [
{"input": 0, "expected_output": 1},
{"input": 1, "expected_output": 1},
{"input": 5, "expected_output": 120},
{"input": -1, "expected_output": None} # 期望这里会有错误处理，而不是递归
]
test_runner = self.tools["test_runner"]
test_results = test_runner.run(code=generated_code, tests=test_cases)
self._record_step("tool_use", "Ran tests on generated code.", "test_runner", {"code": generated_code, "tests": test_cases}, test_results)
print(f"n--- Test Results ---n{json.dumps(test_results, indent=2, ensure_ascii=False)}n--- End Test Results ---n")
if test_results.get("status") == "success":
final_output = {"status": "success", "message": "Code generated and passed all tests.", "code": generated_code, "test_results": test_results}
else:
final_output = {"status": "failure", "message": "Code generated but failed some tests.", "code": generated_code, "test_results": test_results}
self._record_step("end_task", "Task finished.", output=final_output)
return final_output
# --- 主程序入口 ---
if __name__ == "__main__":
llm = MockLLM()
memory = Memory()
# 初始化工具
code_generator_tool = CodeGeneratorTool()
test_runner_tool = TestRunnerTool()
tools_list = [code_generator_tool, test_runner_tool]
# 初始化Agent
agent = CodeAgent(llm, memory, tools_list)
# 初始化Reflection Node
reflection_node = ReflectionNode(llm, memory)
# 运行第一个任务
print("--- Running Task 1: Generate Factorial Function ---")
task_requirement_1 = "编写一个Python函数 `factorial(n)`，计算给定非负整数n的阶乘。"
task_result_1 = agent.run_task(task_requirement_1)
print("n##### Task 1 Completed. Initiating Reflection. #####")
reflection_output_1 = reflection_node.reflect(
task_context=agent.task_context,
execution_trace=agent.execution_trace,
final_result=task_result_1
)
print("##### Reflection for Task 1 Finished. #####")
print(f"nMemory after Task 1 Reflection:nKnowledge Base: {memory.knowledge_base}nStrategy Rules: {memory.strategy_rules}")
# 运行第二个任务 (假设 Agent 应该从第一次反思中学习并改进)
# 模拟LLM在规划和执行时会考虑记忆中的策略
print("nn--- Running Task 2: Generate Power Function ---")
task_requirement_2 = "编写一个Python函数 `power(base, exp)`，计算 `base` 的 `exp` 次幂。确保处理好 `exp` 为负数和零的情况。"
# 为了演示，这里我们手动注入一些“学习”后的行为，
# 实际中LLM在生成plan和code_gen_prompt时会引用memory.retrieve_relevant_info
# MockLLM的generate方法也会被修改，以体现这种学习后的行为
# 假设学习后，Agent的规划会更细致
# 假设学习后，Agent的code_generator会更注意边界条件
# 这里我们不修改MockLLM的行为，只是再次运行，看记忆库的变化
task_result_2 = agent.run_task(task_requirement_2)
print("n##### Task 2 Completed. Initiating Reflection. #####")
reflection_output_2 = reflection_node.reflect(
task_context=agent.task_context,
execution_trace=agent.execution_trace,
final_result=task_result_2
)
print("##### Reflection for Task 2 Finished. #####")
print(f"nMemory after Task 2 Reflection:nKnowledge Base: {memory.knowledge_base}nStrategy Rules: {memory.strategy_rules}")
# 验证记忆库内容
print("n--- Current Memory Content ---")
print(f"Reflections Count: {len(memory.reflections)}")
print(f"Knowledge Base Entries: {len(memory.knowledge_base)}")
for item in memory.knowledge_base:
print(f"- {item}")
print(f"Strategy Rules Entries: {len(memory.strategy_rules)}")
for rule in memory.strategy_rules:
print(f"- {rule}")代码解析：
MockLLM: 模拟了一个LLM的行为，它会根据输入的prompt返回预设的JSON字符串，代表LLM对批判和经验教训的生成。在真实场景中，这里会调用OpenAI或其他LLM API。Tools (CodeGeneratorTool,TestRunnerTool): 模拟了Agent可以调用的外部工具。CodeGeneratorTool生成代码（这里故意生成了一个对负数处理不当的阶乘函数）。TestRunnerTool执行代码并返回测试结果。Memory: 存储Agent的反思日志 (reflections)、提炼出的知识 (knowledge_base) 和改进策略 (strategy_rules)。retrieve_relevant_info方法模拟了从记忆中检索相关信息，供Agent在规划和执行阶段参考。ReflectionNode:- 接收
task_context,execution_trace,final_result作为输入。 - 使用
critique_prompt_template和lessons_prompt_template调用LLM进行两阶段的推理。 - 将LLM生成的批判报告和经验教训解析为Python对象。
- 将这些结果存储到
Memory中，更新知识库和策略规则。
- 接收
CodeAgent:- 代表了Agent的核心工作流。
_record_step方法用于记录Agent的每一步行动及其输出，构建execution_trace。run_task方法模拟了Agent从规划、代码生成到测试的整个流程。在规划和代码生成阶段，它会尝试从memory中检索相关信息。- 在
run_task结束后，main函数显式调用reflection_node.reflect()，将任务结果和执行轨迹传递给它进行反思。
通过这个例子，我们可以看到，即使第一次任务由于代码缺陷而失败，Reflection Node也能：
- 识别问题： 发现阶乘函数未能处理负数输入。
- 分析原因： 归因于规划阶段缺乏对边缘情况的考虑，以及代码生成器未被明确指示处理这些情况。
- 提出策略： 建议规划器在未来任务中增加生成全面测试用例的步骤，并要求代码生成工具在生成代码时考虑这些测试用例。
- 更新记忆： 将“阶乘函数处理负数最佳实践”等知识片段和新的策略规则存储起来。
在第二次任务中，理论上Agent应该能从这些更新的记忆中获益，从而改进其规划和执行，生成更健壮的代码。尽管在这个简化的MockLLM中，我们没有完全实现Agent基于记忆进行动态调整的复杂逻辑，但其核心思想已通过 memory.retrieve_relevant_info 和 memory.update_strategy_rules 得到了体现。
7. 高级反思技术与最佳实践
7.1 高级反思技术
-
元反思 (Meta-Reflection):
- Agent不仅反思其任务执行，还反思其反思过程本身。例如，它可能会问：“这次反思是否足够深入？我是否遗漏了关键信息？我的反思提示是否可以改进？”
- 这有助于优化Reflection Node自身的效率和质量。
-
多Agent协作反思:
- 在多Agent系统中，Agent可以共同反思一个协作任务。每个Agent从自己的视角贡献观察和批判，然后共同提炼出共享的经验教训和改进策略。
- 这需要Agent之间有良好的通信和协同反思机制。
-
时间序列反思:
- Agent不仅仅反思单次任务，还可以在多个任务或长期运行中进行时间序列的反思，识别出随时间变化的趋势、重复出现的问题模式或持续有效的策略。
- 这有助于Agent进行更宏观的策略调整和长期学习。
-
引导式反思 (Guided Reflection):
- 除了开放式的批判，可以向LLM提供特定的反思框架、问题列表或评估维度，以确保反思的全面性和深度。
- 例如，在代码生成任务中，可以要求Agent从“代码正确性”、“代码可读性”、“性能”、“安全性”等多个维度进行反思。
7.2 挑战与最佳实践
| 挑战 | 最佳实践 “`python
import json
import uuid
import datetime
from typing import List, Dict, Any, Optional
— LLM 接口抽象 —
class LLMInterface:
def generate(self, prompt: str, temperature: float = 0.7, max_tokens: int = 1024) -> str:
"""
模拟或调用实际的LLM API来生成文本。
"""
raise NotImplementedError
— 模拟LLM实现 —
class MockLLM(LLMInterface):
"""
一个用于演示的模拟LLM，根据prompt内容返回预设的JSON字符串。
在真实场景中，这里会调用OpenAI/Anthropic等API。
"""
def init(self, model_name="gpt-4"):
self.model_name = model_name
def generate(self, prompt: str, temperature: float = 0.7, max_tokens: int = 1024) -> str:
print(f"n--- LLM Input for {self.model_name} ---n{prompt[:500]}...n--- End LLM Input ---") # 只显示前500字符
# 模拟LLM根据Prompt内容返回不同的结构化结果
if "critique_prompt_template" in prompt and "factorial" in prompt:
return json.dumps({
"task_completion": "部分成功",
"completion_reason": "核心功能实现，但未能处理边缘情况（负数输入）。",
"success_points": [
{
"step_id": 2,
"description": "成功调用code_generator生成初始代码。",
"reason": "Agent正确理解了需求，并选择了合适的工具。"
}
],
"failure_points": [
{
"step_id": 3,
"description": "测试用例失败：未能处理负数输入。",
"root_cause": "规划阶段未充分考虑边缘情况，代码生成器也未被明确指示处理负数，导致测试失败。",
"type": "规划失误/代码缺陷"
}
],
"optimization_potential": [
{
"area": "规划",
"suggestion": "在规划阶段增加对边缘情况（如负数、大数）的考虑，并明确要求代码生成器处理这些情况。"
}
]
}, indent=2, ensure_ascii=False)
elif "lessons_prompt_template" in prompt and "factorial" in prompt:
return json.dumps({
"lessons_learned": [
"经验教训1：在需求分析阶段，必须主动识别并处理所有可能的边缘情况，特别是数值计算任务中的负数、零、大数等。",
"经验教训2：当发现工具（如code_generator）未能完全满足需求时，应考虑提供更详细的上下文或更具体的约束条件，而非直接接受其首次输出。",
"经验教训3：测试驱动开发（TDD）的理念对Agent同样适用，先定义全面的测试用例有助于指导代码生成和验证。"
],
"improvement_strategies": [
{
"target_component": "Planner",
"strategy": "在接收到代码生成任务后，规划器应首先生成一组全面的测试用例（包括正常、边界、异常情况），然后再指导代码生成器基于这些测试用例进行开发。",
"priority": "High"
},
{
"target_component": "Executor/CodeGenerator Tool",
"strategy": "修改code_generator工具的调用接口，允许在生成代码时直接传入测试用例，并指示其代码必须通过这些测试。如果无法通过，则要求其进行自我修正。",
"priority": "High"
}
],
"memory_update_content": [
"知识片段: '阶乘函数处理负数最佳实践：对于负数输入，应抛出ValueError或返回特定错误码，而非进入无限递归。'",
"策略规则: '在规划任何数值计算类函数时，必须明确考虑并处理零、负数、浮点数等特殊输入。'"
]
}, indent=2, ensure_ascii=False)
elif "power" in prompt: # 模拟第二次任务的反射
return json.dumps({
"task_completion": "成功",
"completion_reason": "代码正确处理了正负指数及零指数情况。",
"success_points": [
{
"step_id": 2,
"description": "代码生成器根据需求生成了正确的幂函数，并考虑了边界情况。",
"reason": "Agent在规划阶段可能参考了处理边缘情况的策略，并将其传递给了代码生成工具。"
},
{
"step_id": 3,
"description": "所有测试用例均通过。",
"reason": "测试用例设计全面，代码实现健壮。"
}
],
"failure_points": [],
"optimization_potential": []
}, indent=2, ensure_ascii=False)
return "{"plan": "模拟LLM生成结果"}" # 默认返回一个简单的JSON— 模拟工具接口 —
class Tool:
def init(self,