AI Agent Harness Engineering 与RPA的区别：自动化工具的下一代进化方向

副标题：从规则驱动到智能决策，探索企业自动化的未来演进路径

---

摘要/引言

在当今快速发展的数字化时代，企业对效率提升和成本优化的追求从未停止。自动化技术作为实现这一目标的关键手段，已经经历了从传统IT自动化到机器人流程自动化(RPA)的演进。然而，随着业务环境的日益复杂和动态变化，传统RPA在处理非结构化数据、适应流程变化以及做出智能决策等方面面临着显著局限。

本文将深入探讨AI Agent Harness Engineering（智能体工程）作为自动化工具的下一代进化方向，如何通过结合大语言模型、规划能力和工具使用，突破传统RPA的限制。我们将详细分析两者在架构、工作原理、适用场景和能力边界上的本质区别，并通过实际案例和代码示例，展示如何设计和实现基于AI Agent的自动化解决方案。

读完本文，您将：

• 清晰理解RPA与AI Agent的核心概念与技术差异
• 掌握AI Agent系统的设计原则和实现方法
• 了解如何在实际业务场景中选择合适的自动化方案
• 获得构建简单AI Agent系统的实践经验

让我们一起踏上这场自动化技术演进之旅，探索如何 harness（驾驭）AI Agent的力量，为企业创造更大的价值。

---

目标读者与前置知识

目标读者：

• 企业技术决策者和数字化转型负责人
• 有一定RPA实施经验的自动化工程师
• 对AI应用和自动化技术感兴趣的软件开发者
• 希望了解前沿自动化技术的产品经理

前置知识：

• 基础的编程知识（Python优先）
• 对API调用和Web服务有基本了解
• 对机器学习和大语言模型概念有初步认识
• （可选）有RPA工具使用经验将有助于理解对比内容

---

文章目录

1. 引言与基础
2. 问题背景与动机
3. 核心概念与理论基础
   • RPA的定义、架构与工作原理
   • AI Agent Harness Engineering的定义、架构与工作原理
4. RPA与AI Agent的深度对比
5. AI Agent的核心技术组件解析
6. 环境准备
7. 分步实现：构建你的第一个AI Agent
8. 关键代码解析与深度剖析
9. 结果展示与验证
10. 性能优化与最佳实践
11. 常见问题与解决方案
12. 未来展望与扩展方向
13. 总结
14. 参考资料
15. 附录

---

问题背景与动机

自动化的演进历程

企业自动化的需求由来已久。从早期的批处理脚本到宏命令，再到后来的IT自动化工具，企业一直在寻求减少人工操作、提高效率的方法。然而，这些早期的自动化方案往往局限于特定系统或技术栈，缺乏通用性和灵活性。

进入21世纪第二个十年，机器人流程自动化(RPA)开始崭露头角。RPA通过模拟人类用户与计算机系统的交互方式，能够跨多个应用系统执行重复性任务，无需对现有系统进行修改。这一特性使得RPA迅速成为企业数字化转型的热门技术，被广泛应用于财务、人力资源、客户服务等领域。

传统RPA的局限性

尽管RPA取得了显著成功，但随着应用场景的不断拓展，其内在局限性也日益凸显：

1. 规则驱动的刚性：传统RPA严格基于预定义规则运行，任何超出规则范围的异常情况都可能导致流程失败。
2. 处理非结构化数据的困难：RPA主要针对结构化数据和标准化流程设计，处理文本、图像等非结构化数据的能力有限。
3. 适应变化的能力差：当业务流程或系统界面发生变化时，RPA机器人往往需要重新配置甚至重新开发，维护成本高。
4. 缺乏决策能力：传统RPA只能执行预设指令，无法基于上下文进行推理或做出复杂决策。
5. 无法处理开放性任务：对于目标模糊或需要探索性工作的任务，RPA几乎无能为力。

新一代自动化需求的兴起

与此同时，企业面临的业务环境正在发生深刻变化：

• 数据类型多样化：企业需要处理的非结构化数据（文本、图像、语音等）比例越来越高。
• 流程动态化：市场竞争加剧要求企业业务流程能够快速调整和优化。
• 决策复杂化：许多业务场景需要综合多方面信息，做出基于上下文的智能决策。
• 任务开放化：从简单的执行型任务向复杂的探索型、创新型任务扩展。

这些变化催生了对新一代自动化工具的迫切需求。企业需要的不仅仅是"按指令执行"的工具，更是能够"理解任务、制定计划、自主执行、适应变化"的智能助手。

AI Agent的崛起

正是在这样的背景下，AI Agent（智能体）作为下一代自动化技术开始受到广泛关注。AI Agent结合了大语言模型(LLM)的强大理解和推理能力、规划与决策能力，以及工具使用能力，能够处理更加复杂、动态和开放的任务。

与传统RPA相比，AI Agent具有更高的灵活性、适应性和智能性，能够在没有明确规则的情况下自主完成任务。这使得AI Agent成为应对当前企业自动化挑战的理想选择，也代表了自动化技术的未来发展方向。

---

核心概念与理论基础

RPA的定义、架构与工作原理

什么是RPA？

机器人流程自动化(Robotic Process Automation, RPA)是一种技术，允许企业通过配置软件"机器人"来捕获和解释现有应用程序，以便处理交易、操纵数据、触发响应并与其他数字系统通信。

RPA机器人通过模拟人类用户与计算机系统的交互方式工作，它们可以像人一样点击按钮、输入文本、读取数据，并在不同系统之间传递信息。关键的是，RPA不需要对现有系统进行修改或集成，而是"坐落在"现有系统之上。

RPA的核心架构

典型的RPA系统通常包含以下几个核心组件：

1. 开发环境（Studio/Designer）：用于创建和编辑自动化流程的可视化开发工具。
2. 机器人（Bot）：执行自动化流程的软件组件，可以是有人值守（Attended）或无人值守（Unattended）。
3. 控制中心（Orchestrator/Controller）：用于管理、调度和监控机器人的中央控制台。
4. 存储库（Repository）：用于存储流程定义、配置和相关资产。

RPA的工作原理

RPA的工作流程通常遵循以下步骤：

1. 流程分析与设计：首先分析目标业务流程，识别可自动化的步骤，设计流程图。
2. 流程开发：在开发环境中使用录制或拖拽方式创建自动化脚本。
3. 测试与调试：在测试环境中验证自动化流程的正确性，处理异常情况。
4. 部署与执行：将流程部署到生产环境，由机器人按照预定计划或触发条件执行。
5. 监控与维护：监控机器人运行状态，收集运行数据，根据需要调整流程。

RPA的技术特点

• 基于规则：严格按照预定义规则执行，逻辑清晰但缺乏灵活性。
• 界面交互：通过识别UI元素（按钮、文本框等）与系统交互。
• 结构化数据处理：擅长处理表格、数据库等结构化数据。
• 精确执行：在规则范围内执行精度高，减少人为错误。
• 可审计：所有操作都可记录和审计，满足合规要求。

AI Agent Harness Engineering的定义、架构与工作原理

什么是AI Agent？

AI Agent（智能体）是一个能够感知环境、做出决策并采取行动的自主系统。在本文的语境中，我们主要讨论基于大语言模型(LLM)的AI Agent，这类Agent利用LLM作为核心"大脑"，结合规划、记忆和工具使用能力，能够自主完成复杂任务。

AI Agent Harness Engineering则是一门专注于设计、构建和部署AI Agent系统的工程学科，它涉及如何有效"驾驭"(Harness)AI的能力，使其能够可靠、高效地完成实际任务。

AI Agent的核心架构

一个典型的AI Agent系统通常包含以下核心组件：

1. 大语言模型(LLM)核心：作为Agent的"大脑"，负责理解任务、生成推理和决策。
2. 感知模块：负责从环境中收集信息，可能包括文本处理、图像识别等。
3. 规划模块：负责将复杂任务分解为可管理的子任务，制定执行计划。
4. 记忆模块：存储和检索任务相关信息，包括短期记忆和长期记忆。
5. 工具使用模块：允许Agent访问和使用外部工具（API、数据库、其他软件等）。
6. 执行模块：负责执行计划的具体操作，并与环境交互。
7. 反思与优化模块：评估执行结果，总结经验教训，优化未来表现。

AI Agent的工作原理

AI Agent的工作流程通常遵循以下循环：

1. 任务理解：接收并理解用户的目标或任务需求。
2. 环境感知：收集与任务相关的环境信息和上下文。
3. 规划制定：基于任务目标和当前状态，制定执行计划。
4. 行动执行：根据计划执行具体行动，可能包括使用工具。
5. 结果观察：观察行动结果，获取反馈信息。
6. 反思调整：评估结果与目标的差距，调整计划或策略。
7. 循环迭代：重复上述步骤直到达成目标或确定无法完成。

AI Agent的技术特点

• 基于目标：以完成目标为导向，而非严格遵循预设步骤。
• 上下文理解：能够理解自然语言和复杂上下文，处理非结构化信息。
• 自主决策：可以根据情况自主做出决策，适应环境变化。
• 学习能力：能够从经验中学习，优化未来表现。
• 工具扩展：可以通过集成外部工具扩展能力边界。

---

RPA与AI Agent的深度对比

概念核心属性维度对比

为了更清晰地理解RPA与AI Agent的区别，我们从多个核心属性维度进行对比：

属性维度	RPA	AI Agent
驱动方式	规则驱动	目标驱动 + 推理
智能水平	低（仅执行预设指令）	高（理解、推理、决策）
适应性	差（流程变化需重新配置）	强（可自主适应变化）
数据处理	主要处理结构化数据	可处理结构化和非结构化数据
决策能力	无（仅按规则执行）	有（基于上下文推理决策）
任务类型	重复性、标准化任务	复杂、动态、开放性任务
开发方式	流程录制/拖拽式开发	提示工程 + 工具集成
错误处理	预设异常处理分支	自主识别并尝试解决问题
学习能力	无（不会从经验中学习）	有（可从反馈中优化）
人机交互	预设交互点	自然语言交互，灵活协作
维护成本	高（流程变化需频繁维护）	相对较低（适应能力强）
实施周期	中等（依赖流程标准化程度）	视复杂度而定，框架搭建后扩展快
可扩展性	有限（每个新流程需单独开发）	强（可通过添加工具和知识扩展能力）

概念联系的ER实体关系图

交互关系图

---

AI Agent的核心技术组件解析

大语言模型(LLM)核心

大语言模型是AI Agent的"大脑"，负责理解任务、生成推理和做出决策。当前主流的LLM包括GPT-4、Claude、Llama等，它们具有以下关键能力：

• 自然语言理解：能够理解复杂的自然语言指令和上下文。
• 推理能力：可以进行逻辑推理、因果分析和问题解决。
• 知识应用：利用训练过程中获得的广泛知识解决问题。
• 语言生成：生成流畅、连贯的自然语言响应。

在AI Agent中，LLM的作用不仅仅是生成文本，更重要的是作为推理引擎，将任务目标转化为可执行的计划和行动。

规划模块

规划模块负责将复杂任务分解为可管理的子任务，并制定执行计划。有效的规划对于AI Agent成功完成复杂任务至关重要。常见的规划方法包括：

• 思维链(Chain-of-Thought)：引导LLM逐步思考，将复杂问题分解为中间步骤。
• 思维树(Tree-of-Thought)：探索多个推理路径，评估不同方案的可行性。
• 规划与执行(Plan-and-Execute)：先制定高层计划，再逐步执行并调整。

数学上，任务规划可以表示为状态空间搜索问题：

$$P=(S,A,T,s_0,G)$$

其中：

• $S$ 是状态集合
• $A$ 是行动集合
• $T:S\times A\to S$ 是状态转移函数
• $s_0$ 是初始状态
• $G\subseteq S$ 是目标状态集合

Agent的目标是找到一个行动序列，使系统从初始状态转移到目标状态。

记忆模块

记忆模块允许AI Agent存储和检索任务相关信息，对于维持上下文一致性和学习能力至关重要。记忆通常分为以下几个层次：

• 短期记忆(Working Memory)：存储当前任务的上下文和中间结果，通常通过LLM的上下文窗口实现。
• 长期记忆(Long-term Memory)：存储历史交互和经验，可以通过外部数据库或向量存储实现。
• 程序记忆(Procedural Memory)：存储完成特定任务的技能和流程。

向量数据库是实现长期记忆的常用技术，它可以将文本转换为向量表示，并通过相似度搜索检索相关信息：

$$similarity(a,b)=\frac{a\cdot b}{\Vert a\Vert \Vert b\Vert }$$

其中 $a$ 和 $b$ 是两个向量，相似度通常用余弦相似度计算。

工具使用模块

工具使用能力是AI Agent扩展其能力边界的关键。通过集成外部工具，Agent可以访问实时数据、执行特定计算、与其他系统交互等。常见的工具类型包括：

• API客户端：访问外部服务（如搜索引擎、数据库等）。
• 计算工具：执行数学计算、数据分析等。
• 文件操作工具：读取、写入和处理文件。
• 代码解释器：执行代码并获取结果。

工具集成的关键在于提供清晰的工具描述和使用规范，使LLM能够理解何时以及如何使用每个工具。ReAct模式是一种常用的工具使用框架，它结合了推理(Reasoning)和行动(Acting)：

思考：我需要先了解X信息才能继续
行动：使用工具Y查询X信息
观察：工具Y返回了结果Z
思考：基于Z，我现在应该...

执行模块

执行模块负责将LLM的决策转化为实际行动，并与环境交互。根据应用场景的不同，执行模块可能涉及：

• API调用：调用外部服务执行特定功能。
• 代码执行：运行生成的代码完成计算或数据处理。
• UI自动化：模拟用户交互操作软件界面（类似RPA）。
• 物理设备控制：控制机器人或其他物理设备。

执行模块需要处理各种可能的错误和异常情况，并将结果反馈给Agent的核心部分。

反思与优化模块

反思与优化模块使AI Agent能够从经验中学习，不断提升其性能。这包括：

• 结果评估：比较实际结果与预期目标，分析差距原因。
• 策略调整：根据评估结果调整未来的决策策略。
• 知识更新：将新获得的经验和知识添加到记忆系统中。

自我反思(self-reflection)是一种有效的技术，它引导Agent回顾其推理过程和行动，识别可能的错误或改进空间：

回顾：我刚才采取了行动X，得到了结果Y
分析：这是否是最优的选择？有没有更好的方法？
学习：下次遇到类似情况，我应该...

---

环境准备

在开始构建我们的AI Agent之前，我们需要准备适当的开发环境。以下是我们将使用的技术栈和环境配置指南。

所需软件与库

组件	版本要求	用途
Python	3.9+	主要开发语言
OpenAI API	最新版	访问GPT模型
LangChain	最新版	AI Agent开发框架
ChromaDB	最新版	向量数据库，用于记忆
python-dotenv	最新版	环境变量管理
FastAPI	最新版	API服务（可选）

环境配置步骤

1. 创建虚拟环境

   python -m venv ai-agent-env
   source ai-agent-env/bin/activate  # Linux/Mac
   ai-agent-env\Scripts\activate     # Windows
   

2. 安装依赖包

   pip install openai langchain chromadb python-dotenv fastapi uvicorn
   

3. 创建项目结构

   ai-agent-project/
   ├── .env
   ├── requirements.txt
   ├── main.py
   ├── agent/
   │   ├── __init__.py
   │   ├── core.py
   │   ├── memory.py
   │   ├── tools.py
   │   └── planner.py
   ├── utils/
   │   ├── __init__.py
   │   └── helpers.py
   └── data/
       └── knowledge_base/
   

4. 配置API密钥

   在.env文件中添加您的API密钥：

   OPENAI_API_KEY=your_openai_api_key_here
   

5. 生成requirements.txt

   pip freeze > requirements.txt
   

完成以上步骤后，我们的开发环境就准备好了。接下来，我们将逐步构建AI Agent系统。

---

分步实现：构建你的第一个AI Agent

在这一部分，我们将从零开始构建一个实用的AI Agent系统。我们的Agent将能够执行研究任务、搜索信息、分析数据并生成报告。

步骤1：设置基础架构和配置

首先，让我们创建基础的配置和工具函数：

# utils/helpers.py
import os
from dotenv import load_dotenv

def load_configuration():
    """加载环境变量和配置"""
    load_dotenv()
    config = {
        "openai_api_key": os.getenv("OPENAI_API_KEY"),
        "model_name": os.getenv("MODEL_NAME", "gpt-4"),
        "temperature": float(os.getenv("TEMPERATURE", "0.7")),
        "max_tokens": int(os.getenv("MAX_TOKENS", "2000")),
        "chroma_db_path": os.getenv("CHROMA_DB_PATH", "./data/chroma_db"),
    }
    return config

步骤2：实现记忆模块

接下来，我们创建一个基于向量数据库的记忆模块：

# agent/memory.py
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.schema import Document
from typing import List, Dict, Any
import uuid

class MemoryModule:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.embeddings = OpenAIEmbeddings(openai_api_key=config["openai_api_key"])
        self.vector_store = Chroma(
            persist_directory=config["chroma_db_path"],
            embedding_function=self.embeddings
        )
        self.short_term_memory = []  # 简单的短期记忆实现
        
    def add_to_long_term_memory(self, content: str, metadata: Dict[str, Any] = None):
        """添加内容到长期记忆"""
        doc = Document(
            page_content=content,
            metadata=metadata or {"source": "agent_memory", "timestamp": str(uuid.uuid4())}
        )
        self.vector_store.add_documents([doc])
        self.vector_store.persist()
        
    def query_long_term_memory(self, query: str, top_k: int = 5) -> List[Document]:
        """从长期记忆中查询相关内容"""
        docs = self.vector_store.similarity_search(query, k=top_k)
        return docs
        
    def add_to_short_term_memory(self, item: str):
        """添加内容到短期记忆"""
        self.short_term_memory.append(item)
        # 限制短期记忆大小
        if len(self.short_term_memory) > 10:
            self.short_term_memory.pop(0)
            
    def get_short_term_memory(self) -> List[str]:
        """获取短期记忆内容"""
        return self.short_term_memory
        
    def clear_short_term_memory(self):
        """清空短期记忆"""
        self.short_term_memory = []

步骤3：创建工具集

现在，我们为Agent创建一组实用工具。我们将实现网络搜索、文件操作和数据处理等工具：

# agent/tools.py
from typing import Dict, Any, List
import requests
from bs4 import BeautifulSoup
import json
import os

class Toolkit:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.tools = {
            "web_search": self.web_search,
            "read_file": self.read_file,
            "write_file": self.write_file,
            "analyze_text": self.analyze_text,
            "calculate": self.calculate
        }
        
    def web_search(self, query: str, num_results: int = 5) -> str:
        """模拟网络搜索工具（实际项目中可集成真实搜索API）"""
        # 这里我们用模拟数据代替真实搜索
        search_results = [
            {
                "title": f"关于{query}的重要信息",
                "url": "https://example.com/info1",
                "snippet": f"{query}是一个重要的主题，涉及多个方面..."
            },
            {
                "title": f"{query}的最新研究",
                "url": "https://example.com/research",
                "snippet": f"最新研究表明，{query}在现代技术中发挥着关键作用..."
            },
            {
                "title": f"{query}的应用案例",
                "url": "https://example.com/case-studies",
                "snippet": f"许多企业已经成功应用{query}来解决实际问题..."
            }
        ]
        
        # 格式化搜索结果
        formatted_results = "搜索结果:\n"
        for i, result in enumerate(search_results[:num_results], 1):
            formatted_results += f"{i}. {result['title']}\n"
            formatted_results += f"   URL: {result['url']}\n"
            formatted_results += f"   摘要: {result['snippet']}\n\n"
            
        return formatted_results
        
    def read_file(self, file_path: str) -> str:
        """读取文件内容"""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            return f"文件内容:\n{content}"
        except Exception as e:
            return f"读取文件错误: {str(e)}"
            
    def write_file(self, file_path: str, content: str) -> str:
        """写入内容到文件"""
        try:
            # 确保目录存在
            directory = os.path.dirname(file_path)
            if directory and not os.path.exists(directory):
                os.makedirs(directory)
                
            with open(file_path, 'w', encoding='utf-8') as f:
                f.write(content)
            return f"成功写入文件: {file_path}"
        except Exception as e:
            return f"写入文件错误: {str(e)}"
            
    def analyze_text(self, text: str) -> str:
        """简单的文本分析工具"""
        word_count = len(text.split())
        sentence_count = text.count('.') + text.count('!') + text.count('?')
        paragraph_count = text.count('\n\n') + 1
        
        analysis = f"""文本分析结果:
- 总字数: {len(text)}
- 词数: {word_count}
- 句子数: {sentence_count}
- 段落数: {paragraph_count}
- 平均句长: {word_count / max(sentence_count, 1):.1f} 词"""
        
        return analysis
        
    def calculate(self, expression: str) -> str:
        """安全的数学计算工具"""
        # 注意：在生产环境中应使用更安全的表达式解析方法
        try:
            # 简单的安全检查
            allowed_chars = set("0123456789+-*/(). ")
            if not set(expression).issubset(allowed_chars):
                return "错误：表达式包含不允许的字符"
                
            result = eval(expression)
            return f"计算结果: {expression} = {result}"
        except Exception as e:
            return f"计算错误: {str(e)}"
            
    def execute_tool(self, tool_name: str, params: Dict[str, Any]) -> str:
        """执行指定工具"""
        if tool_name not in self.tools:
            return f"错误：未知工具 '{tool_name}'"
            
        try:
            return self.tools[tool_name](**params)
        except Exception as e:
            return f"执行工具 '{tool_name}' 时出错: {str(e)}"
            
    def get_tool_descriptions(self) -> str:
        """获取所有工具的描述"""
        descriptions = """可用工具:
1. web_search(query, num_results=5) - 搜索网络信息
2. read_file(file_path) - 读取文件内容
3. write_file(file_path, content) - 写入内容到文件
4. analyze_text(text) - 分析文本基本属性
5. calculate(expression) - 执行数学计算"""
        return descriptions

步骤4：实现规划模块

接下来，我们创建一个规划模块，帮助Agent分解任务并制定执行计划：

# agent/planner.py
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import HumanMessage, SystemMessage
from typing import Dict, Any, List
import json

class PlanningModule:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.llm = ChatOpenAI(
            model_name=config["model_name"],
            temperature=config["temperature"],
            max_tokens=config["max_tokens"],
            openai_api_key=config["openai_api_key"]
        )
        
    def create_plan(self, goal: str, context: str = "", available_tools: str = "") -> List[Dict[str, Any]]:
        """根据目标创建执行计划"""
        system_prompt = """你是一位专业的任务规划专家。你的工作是将复杂目标分解为可执行的步骤序列。

请遵循以下指导原则:
1. 分析目标，理解需要完成什么
2. 考虑可用的工具和资源
3. 将目标分解为清晰、有序的步骤
4. 每个步骤应该是具体且可执行的
5. 考虑潜在的依赖关系和前置条件
6. 包含评估和验证步骤以确保质量

请以JSON格式返回计划，格式如下:
{
  "steps": [
    {
      "id": 1,
      "description": "步骤描述",
      "tool": "使用的工具名称或'none'",
      "expected_output": "预期输出描述",
      "depends_on": []  // 依赖的步骤ID列表
    }
  ]
}"""

        user_prompt = f"""目标: {goal}

上下文信息:
{context if context else "无额外上下文"}

可用工具:
{available_tools if available_tools else "无特殊工具可用"}

请根据上述信息，创建一个详细的执行计划。"""

        messages = [
            SystemMessage(content=system_prompt),
            HumanMessage(content=user_prompt)
        ]
        
        try:
            response = self.llm(messages)
            # 尝试解析JSON响应
            plan_text = response.content
            # 提取JSON部分（有时候LLM会添加额外文本）
            json_start = plan_text.find('{')
            json_end = plan_text.rfind('}') + 1
            if json_start != -1 and json_end > json_start:
                plan_json = json.loads(plan_text[json_start:json_end])
                return plan_json.get("steps", [])
            else:
                return [{"id": 1, "description": plan_text, "tool": "none", "expected_output": "执行建议", "depends_on": []}]
        except Exception as e:
            return [{"id": 1, "description": f"无法创建计划，错误: {str(e)}", "tool": "none", "expected_output": "错误处理", "depends_on": []}]
            
    def refine_plan(self, current_plan: List[Dict[str, Any]], feedback: str, completed_steps: List[int]) -> List[Dict[str, Any]]:
        """根据反馈优化计划"""
        # 将当前计划转换为文本
        plan_text = json.dumps(current_plan, indent=2, ensure_ascii=False)
        completed_text = json.dumps(completed_steps, ensure_ascii=False)
        
        system_prompt = """你是一位专业的任务规划专家。你的工作是根据执行反馈优化现有计划。

请遵循以下指导原则:
1. 分析已完成的步骤和当前反馈
2. 识别计划中的问题或改进点
3. 调整后续步骤以解决问题或利用新发现
4. 保持计划的整体目标不变
5. 确保调整后的步骤仍然是可执行的

请以JSON格式返回优化后的计划，格式与原始计划相同。"""

        user_prompt = f"""当前计划:
{plan_text}

已完成步骤:
{completed_text}

反馈信息:
{feedback}

请根据上述信息，优化后续执行计划。"""

        messages = [
            SystemMessage(content=system_prompt),
            HumanMessage(content=user_prompt)
        ]
        
        try:
            response = self.llm(messages)
            plan_text = response.content
            json_start = plan_text.find('{')
            json_end = plan_text.rfind('}') + 1
            if json_start != -1 and json_end > json_start:
                plan_json = json.loads(plan_text[json_start:json_end])
                return plan_json.get("steps", current_plan)
            else:
                return current_plan
        except Exception as e:
            print(f"优化计划时出错: {str(e)}")
            return current_plan

步骤5：构建核心Agent

现在，我们将所有组件整合在一起，构建核心的Agent类：

# agent/core.py
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import HumanMessage, SystemMessage
from typing import Dict, Any, List
import json

from .memory import MemoryModule
from .tools import Toolkit
from .planner import PlanningModule

class ResearchAgent:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.llm = ChatOpenAI(
            model_name=config["model_name"],
            temperature=config["temperature"],
            max_tokens=config["max_tokens"],
            openai_api_key=config["openai_api_key"]
        )
        
        # 初始化各个模块
        self.memory = MemoryModule(config)
        self.tools = Toolkit(config)
        self.planner = PlanningModule(config)
        
        # 执行状态
        self.current_plan = []
        self.completed_steps = []
        self.execution_history = []
        
    def execute_task(self, goal: str) -> Dict[str, Any]:
        """执行给定的任务目标"""
        print(f"开始执行任务: {goal}")
        
        # 重置状态
        self.current_plan = []
        self.completed_steps = []
        self.execution_history = []
        self.memory.clear_short_term_memory()
        
        # 添加任务到记忆
        self.memory.add_to_short_term_memory(f"任务目标: {goal}")
        
        try:
            # 1. 创建初始计划
            print("正在创建执行计划...")
            context = "\n".join(self.memory.get_short_term_memory())
            available_tools = self.tools.get_tool_descriptions()
            self.current_plan = self.planner.create_plan(goal, context, available_tools)
            
            if not self.current_plan:
                return {"success": False, "result": "无法创建执行计划"}
                
            print(f"计划创建完成，包含 {len(self.current_plan)} 个步骤")
            self.memory.add_to_long_term_memory(
                f"任务计划: {json.dumps(self.current_plan, ensure_ascii=False)}",
                {"type": "plan", "goal": goal}
            )
            
            # 2. 执行计划
            results = self._execute_plan()
            
            # 3. 生成最终报告
            print("正在生成最终报告...")
            final_report = self._generate_final_report(goal, results)
            
            # 保存完整执行历史
            self.memory.add_to_long_term_memory(
                f"执行历史: {json.dumps(self.execution_history, ensure_ascii=False)}",
                {"type": "execution_history", "goal": goal}
            )
            
            return {
                "success": True,
                "goal": goal,
                "report": final_report,
                "execution_details": self.execution_history
            }
            
        except Exception as e:
            error_msg = f"执行任务时出错: {str(e)}"
            print(error_msg)
            return {
                "success": False,
                "goal": goal,
                "error": error_msg,
                "execution_details": self.execution_history
            }
            
    def _execute_plan(self) -> List[Dict[str, Any]]:
        """执行当前计划"""
        results = []
        
        for step in self.current_plan:
            step_id = step["id"]
            
            # 检查依赖
            dependencies = step.get("depends_on", [])
            if not all(dep in self.completed_steps for dep in dependencies):
                print(f"跳过步骤 {step_id}，依赖未满足: {dependencies}")
                continue
                
            print(f"\n执行步骤 {step_id}: {step['description']}")
            
            try:
                # 执行步骤
                step_result = self._execute_step(step)
                
                # 记录结果
                result_record = {
                    "step_id": step_id,
                    "description": step["description"],
                    "success": step_result["success"],
                    "output": step_result["output"],
                    "reflection": step_result.get("reflection", "")
                }
                results.append(result_record)
                self.execution_history.append(result_record)
                
                # 添加到记忆
                self.memory.add_to_short_term_memory(
                    f"步骤 {step_id} 结果: {step_result['output']}"
                )
                self.memory.add_to_long_term_memory(
                    f"步骤 {step_id} 执行详情: {json.dumps(result_record, ensure_ascii=False)}",
                    {"type": "step_result", "step_id": step_id}
                )
                
                if step_result["success"]:
                    self.completed_steps.append(step_id)
                    print(f"步骤 {step_id} 完成: {step.get('expected_output', '无预期输出描述')}")
                else:
                    print(f"步骤 {step_id} 失败: {step_result['output']}")
                    
                    # 尝试优化计划
                    print("尝试优化计划...")
                    feedback = f"步骤 {step_id} 失败: {step_result['output']}"
                    self.current_plan = self.planner.refine_plan(
                        self.current_plan, feedback, self.completed_steps
                    )
                    
            except Exception as e:
                error_msg = f"执行步骤 {step_id} 时出错: {str(e)}"
                print(error_msg)
                
                # 记录错误
                error_record = {
                    "step_id": step_id,
                    "description": step["description"],
                    "success": False,
                    "output": error_msg
                }
                results.append(error_record)
                self.execution_history.append(error_record)
                
        return results
        
    def _execute_step(self, step: Dict[str, Any]) -> Dict[str, Any]:
        """执行单个步骤"""
        tool_name = step.get("tool", "none")
        
        if tool_name == "none" or not tool_name:
            # 不需要工具，直接思考/生成文本
            return self._reasoning_step(step)
        else:
            # 使用工具
            return self._tool_step(step, tool_name)
            
    def _reasoning_step(self, step: Dict[str, Any]) -> Dict[str, Any]:
        """执行推理步骤"""
        system_prompt = """你是一位专业的研究助手。请根据提供的信息和当前步骤，进行必要的思考和分析。"""
        
        context = "\n".join(self.memory.get_short_term_memory())
        user_prompt = f"""当前步骤: {step['description']}

上下文信息:
{context}

请根据上述信息，执行当前步骤并提供结果。"""

        messages = [
            SystemMessage(content=system_prompt),
            HumanMessage(content=user_prompt)
        ]
        
        try:
            response = self.llm(messages)
            return {
                "success": True,
                "output": response.content
            }
        except Exception as e:
            return {
                "success": False,
                "output": f"推理步骤出错: {str(e)}"
            }
            
    def _tool_step(self, step: Dict[str, Any], tool_name: str) -> Dict[str, Any]:
        """执行工具步骤"""
        # 首先使用LLM生成工具参数
        system_prompt = """你是一位专业的工具使用专家。请根据步骤描述，生成适当的工具参数。

请以JSON格式返回结果，例如:
{"param1": "value1", "param2": "value2"}"""

        context = "\n".join(self.memory.get_short_term_memory())
        user_prompt = f"""步骤描述: {step['description']}
工具名称: {tool_name}

上下文信息:
{context}

请生成执行此步骤所需的工具参数。"""

        messages = [
            SystemMessage(content=system_prompt),
            HumanMessage(content=user_prompt)
        ]
        
        try:
            response = self.llm(messages)
            # 解析工具参数
            param_text = response.content
            json_start = param_text.find('{')
            json_end = param_text.rfind('}') + 1
            if json_start != -1 and json_end > json_start:
                params = json.loads(param_text[json_start:json_end])
            else:
                # 如果无法解析JSON，尝试使用整个文本作为第一个参数
                params = {"query": param_text} if tool_name == "web_search" else {"text": param_text}
                
            # 执行工具
            tool_output = self.tools.execute_tool(tool_name, params)
            
            # 反思结果
            reflection = self._reflect_on_tool_output(step, tool_output)
            
            return {
                "success": True,
                "output": tool_output,
                "reflection": reflection
            }
        except Exception as e:
            return {
                "success": False,
                "output": f"工具步骤出错: {str(e)}"
            }
            
    def _reflect_on_tool_output(self, step: Dict[str, Any], tool_output: str) -> str:
        """反思工具输出"""
        system_prompt = """你是一位善于反思的研究助手。请评估工具输出是否满足步骤需求，并提供有价值的见解。"""

        user_prompt = f"""步骤: {step['description']}
预期: {step.get('expected_output', '无明确预期')}

实际工具输出:
{tool_output}

请评估上述输出是否满足需求，是否需要进一步行动，以及你从中学到了什么。"""

        messages = [
            SystemMessage(content=system_prompt),
            HumanMessage(content=user_prompt)
        ]
        
        try:
            response = self.llm(messages)
            return response.content
        except Exception as e:
            return f"反思过程出错: {str(e)}"
            
    def _generate_final_report(self, goal: str, results: List[Dict[str, Any]]) -> str:
        """生成最终报告"""
        system_prompt = """你是一位专业的报告撰写人。请根据任务执行情况，生成一份结构清晰、内容全面的最终报告。

报告应包含:
1. 任务概述
2. 执行摘要
3. 主要发现
4. 结论与建议
5. 附录(可选)"""

        context = "\n".join(self.memory.get_short_term_memory())
        results_summary = json.dumps(results, indent=2, ensure_ascii=False)
        
        user_prompt = f"""任务目标: {goal}

上下文信息:
{context}

执行结果摘要:
{results_summary}

请根据上述信息，生成一份专业的最终报告。"""

        messages = [
            SystemMessage(content=system_prompt),
            HumanMessage(content=user_prompt)
        ]
        
        try:
            response = self.llm(messages)
            return response