当前位置：首页 > news >正文

39、AI Agent系统开发：智能代理的完整构建体系

news 2025/9/20 17:06:47

核心学习目标：掌握基于大模型的智能代理系统设计，实现工具调用和任务规划功能，建立智能助手的完整开发能力，理解Agent系统的技术架构。

AI Agent系统代表了人工智能向自主智能体发展的重要里程碑，它不仅具备理解和生成能力，更能够主动感知环境、制定计划、调用工具、执行任务并从经验中学习。通过构建完整的感知-规划-执行循环，设计安全可控的工具调用机制，建立有效的记忆和学习系统，我们能够创建真正实用的智能助手，为复杂任务的自动化处理提供强大的技术支撑。

一、Agent架构系统设计：智能代理的核心框架

1.1 感知-规划-执行循环的设计原理

> 环境感知的多维度信息获取

用户意图的深层理解不仅仅是简单的语言理解，更需要结合上下文、用户历史、当前环境状态等多方面信息来准确把握用户的真实需求。Agent需要能够识别显性需求（用户直接表达的）和隐性需求（用户未直接说明但可以推断的），处理模糊指令、补全缺失信息，并能够主动澄清歧义。

环境状态的实时监控包括系统资源状态、可用工具状态、外部服务可用性、数据更新状态等。Agent需要维护一个动态的环境模型，实时更新各种状态信息，为后续的规划决策提供准确的基础数据。这种感知不是被动的，而是主动的持续监控过程。

多模态信息的融合处理现代Agent系统需要处理文本、图像、音频等多种类型的输入，将这些异构信息融合为统一的内部表示。这要求Agent具备跨模态的理解能力，能够从多个信息源中提取关键信息并进行有效整合。

> 认知规划的智能决策机制

目标层次的结构化分解将用户的高层次目标分解为具体可执行的子任务，建立清晰的目标层次结构。这个过程需要考虑任务间的依赖关系、执行顺序、资源约束等多种因素。Agent需要具备从抽象到具体的推理能力，将模糊的高层目标转化为明确的操作步骤。

策略生成的多路径规划对于同一个目标，通常存在多种实现路径。Agent需要能够生成多个可选方案，评估各方案的优劣，选择最优的执行策略。这种评估需要考虑成功率、执行时间、资源消耗、风险等多个维度，并能够根据实际情况动态调整。

动态规划的自适应调整执行过程中环境和条件可能发生变化，原定计划可能不再适用。Agent需要具备动态重新规划的能力，能够根据新的信息和反馈及时调整策略，保证目标的最终实现。这要求Agent具备灵活性和鲁棒性。

1.2 模块化架构的工程实践

> 松耦合设计的系统优势

模块间接口的标准化通过定义清晰的接口协议，使得各个模块能够独立开发、测试和维护。感知模块、规划模块、执行模块之间通过标准化的数据格式和通信协议进行交互，降低了系统的复杂性，提高了可扩展性和可维护性。

插件化工具的灵活集成将各种工具和能力设计为可插拔的组件，Agent能够根据需要动态加载和卸载不同的工具。这种设计使得系统能够轻松扩展新的功能，也便于不同场景下的定制化配置。

异步执行的性能优化采用异步架构设计，允许多个任务并行执行，提高系统整体性能。感知、规划、执行可以在不同的线程或进程中并行进行，通过消息队列等机制进行协调，最大化系统吞吐量。

> 状态管理的一致性保障

全局状态的集中管理建立统一的状态管理中心，维护Agent的全局状态信息，包括当前任务状态、环境状态、工具状态等。这种集中管理避免了状态不一致的问题，为各个模块提供了一致的状态视图。

事务性操作的原子保证对于涉及多个步骤的复杂操作，采用事务性机制保证操作的原子性。如果任何一个步骤失败，整个操作都会回滚到初始状态，避免系统处于不一致的中间状态。

版本控制的历史追踪为状态变更建立版本控制机制，记录每次状态变更的历史，支持状态回滚和变更追踪。这对于调试、审计和错误恢复都非常重要。

二、工具调用系统实现：安全可控的能力扩展机制

2.1 工具描述与注册机制

> 工具元数据的结构化定义

功能描述的语义化表达每个工具需要提供清晰的功能描述，不仅包括基本的功能说明，还要包括使用场景、输入输出格式、执行条件、副作用等详细信息。这种描述需要采用标准化的格式，便于Agent理解和选择合适的工具。

参数规范的类型安全定义严格的参数类型系统，包括基本数据类型、复合类型、可选参数、默认值等。每个参数都要有详细的说明，包括取值范围、格式要求、业务含义等。这种类型安全的设计能够在调用前发现大部分参数错误。

依赖关系的明确声明工具可能依赖于其他工具或外部服务，需要明确声明这些依赖关系。包括必需依赖、可选依赖、版本要求等。Agent在规划时需要考虑这些依赖关系，确保调用顺序的正确性。

> 动态注册的插件体系

热插拔的工具加载支持运行时动态添加或移除工具，无需重启Agent系统。新工具可以通过配置文件、API接口或自动发现机制注册到系统中，立即可用。这种热插拔能力大大提高了系统的灵活性。

版本管理的兼容策略支持同一工具的多个版本并存，根据需要选择合适的版本。建立版本兼容性检查机制，确保新版本不会破坏现有功能。提供平滑的版本升级路径，支持渐进式迁移。

命名空间的冲突避免建立工具命名空间机制，避免不同来源的工具命名冲突。支持按功能分类、按提供方分组等多种组织方式，便于管理和查找。

2.2 安全调用的防护机制

> 权限控制的多层防护

基于角色的访问控制建立细粒度的权限控制体系，不同的用户角色具有不同的工具调用权限。敏感工具需要特殊权限才能调用，普通用户默认只能访问安全的基础工具。权限可以动态调整，支持临时授权和权限委托。

资源配额的使用限制对工具调用的频率、并发数、资源消耗等设置合理的限制，防止滥用和资源耗尽。不同用户和不同工具可以有不同的配额设置，支持动态调整和弹性扩展。

调用链的审计追踪记录每次工具调用的详细信息，包括调用者、调用时间、参数内容、执行结果等。建立完整的审计日志，支持安全分析和问题排查。对于敏感操作，可以要求额外的确认和审批。

> 沙箱执行的隔离保护

进程级别的资源隔离将工具执行隔离在独立的进程中，限制其可访问的资源范围。通过操作系统的进程隔离机制，防止恶意工具影响主系统的安全性。设置合理的资源限制，包括CPU、内存、磁盘IO等。

网络访问的精确控制限制工具的网络访问权限，只允许访问必要的外部服务。建立网络白名单机制，禁止访问未经授权的地址。监控网络流量，及时发现异常访问行为。

文件系统的访问限制限制工具对文件系统的访问范围，只允许读写指定目录下的文件。建立临时工作目录机制，工具执行完成后自动清理临时文件。对于敏感文件，需要特殊权限才能访问。

三、任务规划算法应用：智能化的执行策略生成

3.1 基于大模型的任务分解

> 层次化分解的递归策略

目标导向的分解原则从用户的最终目标出发，逐层向下分解，每一层的子任务都应该明确服务于上层目标。分解过程需要保持目标的完整性和一致性，确保所有子任务的完成能够实现原始目标。同时要避免过度分解，保持合理的粒度。

依赖关系的图形化建模将任务间的依赖关系表示为有向无环图（DAG），明确哪些任务必须先完成，哪些任务可以并行执行。这种图形化建模有助于发现执行瓶颈，优化执行顺序，提高整体效率。

上下文信息的传递机制在任务分解过程中，需要考虑上下文信息的传递。子任务的执行可能需要依赖父任务或兄弟任务的结果，需要建立有效的信息传递机制。这包括参数传递、状态共享、结果复用等多种方式。

> 知识引导的智能分解

领域知识的融入利用结合特定领域的专业知识，提高任务分解的准确性和效率。例如，在软件开发任务中，可以利用软件工程的最佳实践；在数据分析任务中，可以利用统计学和机器学习的方法论。

模式识别的经验复用通过分析历史任务的分解模式，识别常见的任务类型和分解策略，建立分解模板库。对于相似的任务，可以复用已有的成功分解方案，提高分解的效率和质量。

动态调整的自适应机制在执行过程中，根据实际情况调整任务分解结果。如果发现原有分解不合理，可以动态重新分解，增加、删除或修改子任务。这种自适应机制提高了Agent对复杂环境的适应能力。

3.2 执行顺序的优化算法

> 多约束条件下的调度优化

时间约束的权衡平衡在满足依赖关系的前提下，尽可能减少总执行时间。这涉及到关键路径的识别、并行任务的安排、资源冲突的解决等多个方面。需要综合考虑任务执行时间、资源可用性、依赖约束等多种因素。

资源约束的合理分配不同任务可能需要不同类型的资源（CPU、内存、网络、特定工具等），需要合理安排资源分配，避免资源冲突和资源闲置。建立资源预约和释放机制，支持资源的动态调整。

优先级约束的策略实施根据任务的重要性和紧急性设置优先级，在调度时优先安排高优先级任务。但同时要避免低优先级任务被饥饿，建立公平性保障机制。

> 启发式搜索的路径寻优

A*算法的智能搜索将任务规划问题转化为图搜索问题，使用A*等启发式算法寻找最优执行路径。设计合适的启发式函数，能够快速收敛到较好的解。考虑多目标优化，在执行时间、资源消耗、成功率等多个目标间寻找平衡。

A*算法是一种用于寻找最短路径的搜索算法，常用于游戏开发、机器人导航和地图路径规划等场景。

它的核心思想是结合了两个信息来指导搜索：一是从起点到当前位置的实际距离成本，二是从当前位置到终点的估计距离。算法会优先探索那些总成本（实际成本+估计成本）最小的路径，这样可以更高效地找到最优解。

相比于其他搜索算法，A*的优势在于它既能保证找到最短路径，又比盲目搜索更快。它通过启发式函数来"猜测"哪个方向更有希望，从而减少不必要的搜索。

简单来说，A*就像是一个聪明的导航系统，它不仅知道你已经走了多远，还能估算还要走多远，然后选择最有希望的路线继续前进。

A*算法的基本实现步骤是这样的：

维护两个列表：开放列表（待探索的节点）和关闭列表（已探索的节点）。每个节点记录三个值：g值（从起点到该点的实际成本）、h值（从该点到终点的启发式估计）、f值（g+h的总成本）。

算法流程：先把起点放入开放列表，然后重复以下步骤：从开放列表中选择f值最小的节点作为当前节点，将其移到关闭列表；检查当前节点的所有邻居，如果邻居是终点就结束；否则计算邻居的g、h、f值，如果邻居不在开放列表中就加入，如果已在列表中但新路径更短就更新它。

关键的数据结构通常用优先队列（堆）来维护开放列表，这样可以快速找到f值最小的节点。启发式函数h常用曼哈顿距离或欧几里得距离，必须保证不会高估真实距离。

找到终点后，通过每个节点记录的父节点信息，从终点回溯到起点就得到了完整路径。整个算法的效率很大程度上取决于启发式函数的准确性。

分支限界的剪枝策略在搜索过程中，及时剪除明显不优的分支，提高搜索效率。建立有效的上界和下界估计，快速排除不可行或次优的方案。

局部搜索的优化改进在得到初始解后，通过局部搜索算法进一步优化。使用邻域搜索、模拟退火、遗传算法等方法，寻找更好的执行方案。

四、记忆系统设计实现：智能体的认知基础设施

4.1 短期记忆的实时管理

> 工作记忆的容量控制

注意力窗口的动态调整根据当前任务的复杂性和重要性，动态调整工作记忆的容量。复杂任务需要更大的记忆容量来保持更多的上下文信息，简单任务则可以适当减少记忆使用，提高效率。

信息优先级的智能排序在工作记忆容量有限的情况下，需要智能地选择保留哪些信息。根据信息的重要性、时效性、访问频率等因素进行排序，优先保留关键信息，及时清理过时或不重要的信息。

上下文切换的状态保存当Agent需要处理多个任务或在不同上下文间切换时，需要有效地保存和恢复工作记忆状态。建立任务栈机制，支持任务的嵌套和恢复。

> 会话历史的结构化存储

多轮对话的连贯维护保持多轮对话的上下文连贯性，能够理解代词指代、话题延续、隐含信息等。建立对话状态跟踪机制，记录话题变化、实体提及、情感变化等关键信息。

关键信息的提取标记从对话历史中自动提取和标记关键信息，包括用户偏好、重要决定、约束条件等。这些信息需要特殊标记，便于后续快速检索和使用。

时间衰减的遗忘模拟模拟人类记忆的时间衰减特性，对较早的信息逐渐降低权重。但同时要保留重要信息，避免关键信息的丢失。建立记忆强度模型，根据信息的重要性和使用频率调整衰减速度。

4.2 长期记忆的知识积累

> 经验案例的学习提取

成功模式的抽象总结从成功完成的任务中抽取通用的执行模式，形成可复用的经验模板。这些模板包括任务类型识别、解决步骤、关键决策点、常见问题及解决方案等。

失败经验的风险预警记录和分析失败案例，识别失败的关键因素和风险点。建立风险预警机制，在遇到类似情况时提前预警，避免重复同样的错误。

知识图谱的关系建模将经验知识组织为图结构，建立概念、实体、关系的知识网络。支持基于关系的推理和知识发现，提高知识的利用效率。

> 技能知识的持续更新

增量学习的知识融合支持知识的增量更新，新的经验能够与已有知识有机融合，避免灾难性遗忘。使用在线学习算法，持续改进Agent的能力。

知识冲突的一致性处理当新知识与已有知识冲突时，需要有效的冲突解决机制。建立知识可信度评估，支持知识的版本管理和溯源追踪。

元认知的反思机制建立对自身知识和能力的反思机制，能够评估自己的知识边界和能力局限。在面对超出能力范围的任务时，能够主动寻求帮助或学习新知识。

五、安全控制机制：可信可控的智能代理保障

5.1 权限管理的分层防护

> 基于身份的访问控制

多因子身份认证采用用户名密码、数字证书、生物特征等多种认证方式，确保用户身份的真实性。支持单点登录（SSO）和联邦身份认证，提高用户体验的同时保障安全性。

角色权限的层次管理建立层次化的角色权限体系，从普通用户到管理员，不同角色具有不同的操作权限。支持权限的继承和委托，便于大规模部署和管理。

动态权限的上下文适应根据当前上下文（时间、地点、设备、网络环境等）动态调整用户权限。例如，在内网环境下可以有更高的权限，在公网环境下则需要额外的验证。

> 最小权限原则的实施

功能权限的细粒度划分将系统功能划分为最小的权限单元，用户只能获得完成其工作所必需的最小权限。避免过度授权，减少安全风险。

时限权限的自动回收为敏感操作设置权限有效期，超过时限自动回收权限。支持权限的续期申请，但需要重新审批。

权限使用的实时监控监控权限的实际使用情况，发现异常使用模式。对于长期未使用的权限，可以考虑回收或降级。

5.2 风险控制的主动防护

> 行为模式的异常检测

基线行为的建模学习通过分析正常的用户行为和Agent行为，建立行为基线模型。包括操作频率、操作序列、资源使用模式等多个维度的特征。

异常行为的智能识别使用机器学习算法识别偏离基线的异常行为。支持无监督学习和有监督学习相结合的方式，不断提高检测准确性。

风险评分的量化计算为每个操作和行为计算风险评分，综合考虑操作的敏感性、用户的可信度、环境的安全性等因素。超过阈值的高风险操作需要额外的审核。

> 应急响应的自动化处理

威胁等级的分类响应根据威胁的严重程度制定不同的响应策略。轻微威胁可能只需要日志记录和提醒，严重威胁则需要立即阻断和报警。

自动化阻断的精准实施在检测到高风险行为时，能够精准地阻断相关操作，而不影响其他正常功能。支持黑名单和白名单机制，提高阻断的准确性。

事件溯源的完整追踪记录安全事件的完整过程，包括触发条件、响应措施、影响范围、处理结果等。支持事后分析和责任追溯。

六、智能助手实战：完整Agent系统开发示例代码

基于前面章节的理论基础，现在构建一个完整的AI Agent系统示例代码，集成感知、规划、执行、记忆等核心功能，展示一个的智能助手核心代码。

import asyncio
import json
import logging
import uuid
import time
import re
import os
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional, Callable, Union
from dataclasses import dataclass, asdict
from abc import ABC, abstractmethod
from enum import Enum
import sqlite3
import threading
from concurrent.futures import ThreadPoolExecutor, TimeoutError
import inspect
import hashlib# Web框架和API
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
import uvicorn# 模拟的大模型接口（实际使用时可以替换为真实的模型API）
class MockLLM:"""模拟的大语言模型"""def __init__(self):self.conversation_history = []async def chat(self, messages: List[Dict], max_tokens: int = 1000) -> str:"""模拟对话生成"""# 简单的模拟逻辑，实际使用时应该调用真实的LLM APIlast_message = messages[-1]['content'] if messages else ""# 模拟任务分解if "分解任务" in last_message or "制定计划" in last_message:return """我来分析这个任务并制定执行计划：1. **任务理解**: 理解用户的需求和目标
2. **资源评估**: 检查可用的工具和资源
3. **步骤分解**: 将复杂任务分解为可执行的子任务
4. **执行规划**: 确定执行顺序和依赖关系
5. **结果验证**: 检查执行结果是否符合预期具体执行步骤：
- 首先使用搜索工具获取相关信息
- 然后使用计算工具进行数据处理
- 最后整理结果并生成报告
"""# 模拟工具选择elif "选择工具" in last_message or "哪个工具" in last_message:return "基于当前任务需求，我建议使用以下工具：search_tool进行信息搜索，calculator_tool进行数值计算。"# 默认响应return f"我理解了您的需求：{last_message[:100]}...让我来处理这个任务。"# ===== 核心数据结构 =====class TaskStatus(Enum):"""任务状态枚举"""PENDING = "pending"RUNNING = "running"COMPLETED = "completed"FAILED = "failed"CANCELLED = "cancelled"class TaskPriority(Enum):"""任务优先级"""LOW = 1MEDIUM = 2HIGH = 3URGENT = 4@dataclass
class Task:"""任务数据结构"""id: strname: strdescription: strstatus: TaskStatus = TaskStatus.PENDINGpriority: TaskPriority = TaskPriority.MEDIUMcreated_at: str = Nonestarted_at: str = Nonecompleted_at: str = Noneparent_id: Optional[str] = Nonedependencies: List[str] = Nonetools_required: List[str] = Noneparameters: Dict[str, Any] = Noneresult: Any = Noneerror_message: str = Nonedef __post_init__(self):if self.created_at is None:self.created_at = datetime.now().isoformat()if self.dependencies is None:self.dependencies = []if self.tools_required is None:self.tools_required = []if self.parameters is None:self.parameters = {}@dataclass
class ToolDefinition:"""工具定义"""name: strdescription: strparameters: Dict[str, Any]function: Callablepermissions_required: List[str] = Nonerisk_level: int = 1  # 1-5，1最安全，5最危险timeout: int = 30def __post_init__(self):if self.permissions_required is None:self.permissions_required = []@dataclass
class Memory:"""记忆数据结构"""id: strcontent: strmemory_type: str  # "short_term", "long_term", "procedural", "episodic"created_at: strlast_accessed: str = Noneaccess_count: int = 0importance: float = 1.0tags: List[str] = Nonedef __post_init__(self):if self.tags is None:self.tags = []if self.last_accessed is None:self.last_accessed = self.created_at# ===== 工具系统 =====class ToolRegistry:"""工具注册中心"""def __init__(self):self.tools: Dict[str, ToolDefinition] = {}self._setup_default_tools()def _setup_default_tools(self):"""设置默认工具"""# 搜索工具def search_tool(query: str, max_results: int = 5) -> Dict[str, Any]:"""模拟搜索工具"""return {"query": query,"results": [{"title": f"搜索结果 {i}", "url": f"http://example.com/{i}", "content": f"关于 {query} 的相关信息 {i}"}for i in range(1, min(max_results + 1, 6))]}# 计算工具def calculator_tool(expression: str) -> Dict[str, Any]:"""计算工具"""try:# 简单的安全计算（实际使用中需要更严格的安全检查）allowed_chars = set('0123456789+-*/.()')if not all(c in allowed_chars or c.isspace() for c in expression):raise ValueError("不支持的字符")result = eval(expression)return {"expression": expression, "result": result}except Exception as e:return {"expression": expression, "error": str(e)}# 文件操作工具def file_tool(action: str, filename: str, content: str = None) -> Dict[str, Any]:"""文件操作工具"""if action == "read":try:with open(filename, 'r', encoding='utf-8') as f:return {"action": action, "filename": filename, "content": f.read()}except Exception as e:return {"action": action, "filename": filename, "error": str(e)}elif action == "write":try:with open(filename, 'w', encoding='utf-8') as f:f.write(content or "")return {"action": action, "filename": filename, "success": True}except Exception as e:return {"action": action, "filename": filename, "error": str(e)}else:return {"error": "不支持的操作"}# 注册工具self.register_tool(ToolDefinition(name="search_tool",description="搜索信息的工具，可以查找相关资料",parameters={"query": {"type": "string", "description": "搜索查询"},"max_results": {"type": "integer", "description": "最大结果数", "default": 5}},function=search_tool,risk_level=1))self.register_tool(ToolDefinition(name="calculator_tool", description="数学计算工具，可以进行基本的数学运算",parameters={"expression": {"type": "string", "description": "数学表达式"}},function=calculator_tool,risk_level=2))self.register_tool(ToolDefinition(name="file_tool",description="文件操作工具，可以读取和写入文件",parameters={"action": {"type": "string", "description": "操作类型：read或write"},"filename": {"type": "string", "description": "文件名"},"content": {"type": "string", "description": "写入的内容（write时必需）", "optional": True}},function=file_tool,permissions_required=["file_access"],risk_level=4))def register_tool(self, tool: ToolDefinition):"""注册工具"""self.tools[tool.name] = toollogging.info(f"工具已注册: {tool.name}")def get_tool(self, name: str) -> Optional[ToolDefinition]:"""获取工具"""return self.tools.get(name)def list_tools(self) -> Dict[str, Dict[str, Any]]:"""列出所有工具"""return {name: {"description": tool.description,"parameters": tool.parameters,"risk_level": tool.risk_level,"permissions_required": tool.permissions_required}for name, tool in self.tools.items()}# ===== 安全控制系统 =====class SecurityManager:"""安全管理器"""def __init__(self):self.user_permissions: Dict[str, List[str]] = {"admin": ["file_access", "system_control", "user_management"],"user": ["basic_tools"],"guest": []}self.risk_threshold = 3  # 风险阈值def check_permission(self, user_role: str, tool: ToolDefinition) -> bool:"""检查权限"""if not tool.permissions_required:return Trueuser_perms = self.user_permissions.get(user_role, [])return all(perm in user_perms for perm in tool.permissions_required)def assess_risk(self, tool: ToolDefinition, parameters: Dict[str, Any]) -> Dict[str, Any]:"""风险评估"""risk_score = tool.risk_levelrisk_factors = []# 基于参数内容的风险评估for param, value in parameters.items():if isinstance(value, str):# 检查危险关键词dangerous_patterns = [r'rm\s+-rf', r'del\s+/[qf]', r'format\s+c:',r'>\s*/dev/', r'sudo\s+', r'exec\s*\(']for pattern in dangerous_patterns:if re.search(pattern, value, re.IGNORECASE):risk_score += 2risk_factors.append(f"检测到危险模式: {pattern}")return {"risk_score": risk_score,"risk_factors": risk_factors,"approved": risk_score <= self.risk_threshold}def log_operation(self, user_role: str, tool_name: str, parameters: Dict[str, Any], result: Any):"""记录操作日志"""log_entry = {"timestamp": datetime.now().isoformat(),"user_role": user_role,"tool": tool_name,"parameters": parameters,"result_type": type(result).__name__,"success": "error" not in str(result)}logging.info(f"操作日志: {log_entry}")# ===== 记忆系统 =====class MemorySystem:"""记忆系统"""def __init__(self, db_path: str = "agent_memory.db"):self.db_path = db_pathself.short_term_capacity = 100  # 短期记忆容量self._init_database()self._memory_lock = threading.Lock()def _init_database(self):"""初始化数据库"""conn = sqlite3.connect(self.db_path)cursor = conn.cursor()cursor.execute("""CREATE TABLE IF NOT EXISTS memories (id TEXT PRIMARY KEY,content TEXT NOT NULL,memory_type TEXT NOT NULL,created_at TEXT NOT NULL,last_accessed TEXT,access_count INTEGER DEFAULT 0,importance REAL DEFAULT 1.0,tags TEXT)""")conn.commit()conn.close()def store_memory(self, content: str, memory_type: str, importance: float = 1.0, tags: List[str] = None) -> str:"""存储记忆"""memory_id = str(uuid.uuid4())memory = Memory(id=memory_id,content=content,memory_type=memory_type,created_at=datetime.now().isoformat(),importance=importance,tags=tags or [])with self._memory_lock:conn = sqlite3.connect(self.db_path)cursor = conn.cursor()cursor.execute("""INSERT INTO memories (id, content, memory_type, created_at, last_accessed, importance, tags)VALUES (?, ?, ?, ?, ?, ?, ?)""", (memory.id, memory.content, memory.memory_type,memory.created_at, memory.last_accessed, memory.importance,json.dumps(memory.tags)))conn.commit()conn.close()# 清理过期的短期记忆if memory_type == "short_term":self._cleanup_short_term_memory()return memory_iddef retrieve_memories(self, query: str, memory_type: Optional[str] = None, limit: int = 10) -> List[Dict[str, Any]]:"""检索记忆"""conn = sqlite3.connect(self.db_path)cursor = conn.cursor()# 简单的关键词匹配（实际应用中可以使用向量搜索）sql = "SELECT * FROM memories WHERE content LIKE ?"params = [f"%{query}%"]if memory_type:sql += " AND memory_type = ?"params.append(memory_type)sql += " ORDER BY importance DESC, last_accessed DESC LIMIT ?"params.append(limit)cursor.execute(sql, params)results = cursor.fetchall()# 更新访问记录for result in results:memory_id = result[0]cursor.execute("""UPDATE memories SET last_accessed = ?, access_count = access_count + 1 WHERE id = ?""", (datetime.now().isoformat(), memory_id))conn.commit()conn.close()# 转换为字典格式columns = ['id', 'content', 'memory_type', 'created_at', 'last_accessed', 'access_count', 'importance', 'tags']return [dict(zip(columns, result)) for result in results]def _cleanup_short_term_memory(self):"""清理过期的短期记忆"""conn = sqlite3.connect(self.db_path)cursor = conn.cursor()# 删除超过容量的最旧记忆cursor.execute("""DELETE FROM memories WHERE id IN (SELECT id FROM memories WHERE memory_type = 'short_term'ORDER BY created_at DESC LIMIT -1 OFFSET ?)""", (self.short_term_capacity,))conn.commit()conn.close()def get_memory_stats(self) -> Dict[str, Any]:"""获取记忆统计"""conn = sqlite3.connect(self.db_path)cursor = conn.cursor()cursor.execute("""SELECT memory_type, COUNT(*), AVG(importance), AVG(access_count)FROM memories GROUP BY memory_type""")results = cursor.fetchall()conn.close()stats = {}for result in results:memory_type, count, avg_importance, avg_access = resultstats[memory_type] = {"count": count,"avg_importance": avg_importance or 0,"avg_access_count": avg_access or 0}return stats# ===== 任务规划系统 =====class TaskPlanner:"""任务规划器"""def __init__(self, llm: MockLLM, memory_system: MemorySystem):self.llm = llmself.memory_system = memory_systemasync def decompose_task(self, task_description: str, context: Dict[str, Any] = None) -> List[Task]:"""任务分解"""# 构建提示prompt = f"""
请分解以下任务为可执行的子任务：任务描述: {task_description}上下文信息: {json.dumps(context or {}, ensure_ascii=False)}请以JSON格式返回子任务列表，每个子任务包含：
- name: 任务名称
- description: 详细描述
- tools_required: 需要的工具列表
- priority: 优先级（1-4）
- dependencies: 依赖的其他子任务（如果有）格式示例:
[{"name": "子任务1","description": "详细描述","tools_required": ["tool1"],"priority": 2,"dependencies": []}
]
"""# 调用LLM进行任务分解response = await self.llm.chat([{"role": "user", "content": prompt}])# 解析响应（简化处理）subtasks = []try:# 实际应用中需要更完善的JSON解析lines = response.split('\n')task_counter = 1# 简单的模拟分解（实际使用时需要更智能的解析）if "搜索" in task_description or "查找" in task_description:subtasks.append(Task(id=str(uuid.uuid4()),name=f"信息搜索",description=f"搜索关于'{task_description}'的相关信息",tools_required=["search_tool"],priority=TaskPriority.HIGH))if "计算" in task_description or "分析" in task_description:subtasks.append(Task(id=str(uuid.uuid4()),name=f"数据分析",description=f"分析和计算相关数据",tools_required=["calculator_tool"],priority=TaskPriority.MEDIUM))# 如果没有特定的工具需求，创建一个通用任务if not subtasks:subtasks.append(Task(id=str(uuid.uuid4()),name=f"执行任务",description=task_description,tools_required=[],priority=TaskPriority.MEDIUM))except Exception as e:logging.error(f"任务分解失败: {e}")# 创建一个默认任务subtasks = [Task(id=str(uuid.uuid4()),name="默认任务",description=task_description,tools_required=[],priority=TaskPriority.MEDIUM)]# 存储任务规划到记忆中self.memory_system.store_memory(content=f"任务分解: {task_description} -> {len(subtasks)} 个子任务",memory_type="procedural",importance=1.5,tags=["task_planning", "decomposition"])return subtasksdef plan_execution_order(self, tasks: List[Task]) -> List[Task]:"""规划执行顺序"""# 简单的优先级排序（实际应用中需要考虑依赖关系）sorted_tasks = sorted(tasks, key=lambda t: (-t.priority.value,  # 优先级高的先执行t.created_at        # 创建时间早的先执行))return sorted_tasks# ===== 执行引擎 =====class ExecutionEngine:"""执行引擎"""def __init__(self, tool_registry: ToolRegistry, security_manager: SecurityManager, memory_system: MemorySystem):self.tool_registry = tool_registryself.security_manager = security_managerself.memory_system = memory_systemself.executor = ThreadPoolExecutor(max_workers=5)async def execute_task(self, task: Task, user_role: str = "user") -> Dict[str, Any]:"""执行任务"""task.status = TaskStatus.RUNNINGtask.started_at = datetime.now().isoformat()try:# 记录任务开始执行self.memory_system.store_memory(content=f"开始执行任务: {task.name}",memory_type="episodic",importance=1.0,tags=["task_execution", task.id])results = []# 执行所需的工具for tool_name in task.tools_required:tool = self.tool_registry.get_tool(tool_name)if not tool:raise ValueError(f"工具不存在: {tool_name}")# 安全检查if not self.security_manager.check_permission(user_role, tool):raise PermissionError(f"没有权限使用工具: {tool_name}")# 准备参数parameters = self._prepare_parameters(tool, task.parameters)# 风险评估risk_assessment = self.security_manager.assess_risk(tool, parameters)if not risk_assessment["approved"]:raise SecurityError(f"操作风险过高: {risk_assessment['risk_factors']}")# 执行工具result = await self._execute_tool(tool, parameters)results.append({"tool": tool_name,"parameters": parameters,"result": result})# 记录操作self.security_manager.log_operation(user_role, tool_name, parameters, result)# 任务完成task.status = TaskStatus.COMPLETEDtask.completed_at = datetime.now().isoformat()task.result = results# 记录成功执行self.memory_system.store_memory(content=f"任务执行成功: {task.name}, 结果: {json.dumps(results, ensure_ascii=False)[:200]}...",memory_type="episodic",importance=1.5,tags=["task_execution", "success", task.id])return {"status": "success", "results": results}except Exception as e:# 任务失败task.status = TaskStatus.FAILEDtask.error_message = str(e)task.completed_at = datetime.now().isoformat()# 记录失败self.memory_system.store_memory(content=f"任务执行失败: {task.name}, 错误: {str(e)}",memory_type="episodic",importance=2.0,tags=["task_execution", "failure", task.id])logging.error(f"任务执行失败: {e}")return {"status": "error", "error": str(e)}def _prepare_parameters(self, tool: ToolDefinition, task_params: Dict[str, Any]) -> Dict[str, Any]:"""准备工具参数"""parameters = {}for param_name, param_config in tool.parameters.items():if param_name in task_params:parameters[param_name] = task_params[param_name]elif "default" in param_config:parameters[param_name] = param_config["default"]elif param_config.get("optional", False):continueelse:# 使用智能默认值if param_name == "query" and "query" not in task_params:parameters[param_name] = "default search"elif param_name == "expression" and "expression" not in task_params:parameters[param_name] = "1+1"return parametersasync def _execute_tool(self, tool: ToolDefinition, parameters: Dict[str, Any]) -> Any:"""执行工具"""loop = asyncio.get_event_loop()try:# 在线程池中执行工具函数（避免阻塞）result = await loop.run_in_executor(self.executor,lambda: tool.function(**parameters))return resultexcept TimeoutError:raise TimeoutError(f"工具执行超时: {tool.name}")except Exception as e:raise RuntimeError(f"工具执行失败: {tool.name}, 错误: {str(e)}")# ===== Agent核心类 =====class AIAgent:"""AI Agent核心类"""def __init__(self):# 初始化各个组件self.llm = MockLLM()self.tool_registry = ToolRegistry()self.security_manager = SecurityManager()self.memory_system = MemorySystem()self.task_planner = TaskPlanner(self.llm, self.memory_system)self.execution_engine = ExecutionEngine(self.tool_registry, self.security_manager, self.memory_system)# 任务队列self.task_queue: List[Task] = []self.completed_tasks: List[Task] = []logging.info("AI Agent初始化完成")async def process_request(self, user_input: str, user_role: str = "user", context: Dict[str, Any] = None) -> Dict[str, Any]:"""处理用户请求"""try:# 记录用户请求self.memory_system.store_memory(content=f"用户请求: {user_input}",memory_type="short_term",importance=1.0,tags=["user_request"])# 任务分解tasks = await self.task_planner.decompose_task(user_input, context)# 执行规划planned_tasks = self.task_planner.plan_execution_order(tasks)# 执行任务execution_results = []for task in planned_tasks:result = await self.execution_engine.execute_task(task, user_role)execution_results.append({"task_id": task.id,"task_name": task.name,"status": task.status.value,"result": result})# 将完成的任务加入历史self.completed_tasks.append(task)# 生成响应response = await self._generate_response(user_input, execution_results)# 记录最终响应self.memory_system.store_memory(content=f"Agent响应: {response[:200]}...",memory_type="short_term",importance=1.2,tags=["agent_response"])return {"status": "success","response": response,"tasks_executed": len(execution_results),"execution_details": execution_results}except Exception as e:error_msg = f"处理请求失败: {str(e)}"logging.error(error_msg)# 记录错误self.memory_system.store_memory(content=error_msg,memory_type="short_term",importance=2.0,tags=["error", "processing_failure"])return {"status": "error","error": error_msg,"response": "抱歉，我无法完成您的请求。请检查输入并重试。"}async def _generate_response(self, user_input: str, execution_results: List[Dict]) -> str:"""生成响应"""# 构建响应提示results_summary = []for result in execution_results:if result["status"] == "completed":results_summary.append(f"✅ {result['task_name']}: 已完成")else:results_summary.append(f"❌ {result['task_name']}: 失败")prompt = f"""
用户请求: {user_input}执行结果:
{chr(10).join(results_summary)}详细结果: {json.dumps(execution_results, ensure_ascii=False, indent=2)}请基于执行结果生成一个友好的回复，总结完成的工作和结果。
"""response = await self.llm.chat([{"role": "user", "content": prompt}])return responsedef get_status(self) -> Dict[str, Any]:"""获取Agent状态"""return {"agent_id": id(self),"available_tools": list(self.tool_registry.tools.keys()),"completed_tasks": len(self.completed_tasks),"memory_stats": self.memory_system.get_memory_stats(),"timestamp": datetime.now().isoformat()}def get_memory_context(self, query: str, memory_type: Optional[str] = None) -> List[Dict[str, Any]]:"""获取相关记忆上下文"""return self.memory_system.retrieve_memories(query, memory_type, limit=5)# ===== Web API接口 =====# Pydantic模型
class AgentRequest(BaseModel):message: struser_role: str = "user"context: Optional[Dict[str, Any]] = Noneclass AgentResponse(BaseModel):status: strresponse: strtasks_executed: intexecution_details: List[Dict]class ToolListResponse(BaseModel):tools: Dict[str, Dict[str, Any]]class StatusResponse(BaseModel):agent_id: intavailable_tools: List[str]completed_tasks: intmemory_stats: Dict[str, Any]timestamp: str# 全局Agent实例
agent = AIAgent()# 创建FastAPI应用
app = FastAPI(title="AI Agent System",description="Complete AI Agent system with planning, execution, and memory",version="1.0.0"
)@app.post("/chat", response_model=AgentResponse)
async def chat_with_agent(request: AgentRequest):"""与Agent对话"""try:result = await agent.process_request(user_input=request.message,user_role=request.user_role,context=request.context)return AgentResponse(status=result["status"],response=result["response"],tasks_executed=result["tasks_executed"],execution_details=result["execution_details"])except Exception as e:raise HTTPException(status_code=500, detail=str(e))@app.get("/tools", response_model=ToolListResponse)
async def list_tools():"""列出可用工具"""tools = agent.tool_registry.list_tools()return ToolListResponse(tools=tools)@app.get("/status", response_model=StatusResponse)
async def get_agent_status():"""获取Agent状态"""status = agent.get_status()return StatusResponse(**status)@app.get("/memory/{query}")
async def search_memory(query: str, memory_type: Optional[str] = None):"""搜索记忆"""memories = agent.get_memory_context(query, memory_type)return {"query": query, "memories": memories}@app.get("/health")
async def health_check():"""健康检查"""return {"status": "healthy","timestamp": datetime.now().isoformat(),"system": "AI Agent System"}# ===== 命令行界面 =====class AgentCLI:"""Agent命令行界面"""def __init__(self):self.agent = AIAgent()async def run_interactive(self):"""运行交互式界面"""print("=== AI Agent 智能助手系统 ===")print("输入 'help' 查看帮助，输入 'quit' 退出")print()while True:try:user_input = input("👤 您: ").strip()if user_input.lower() in ['quit', 'exit', '退出']:print("👋 再见！")breakelif user_input.lower() == 'help':self._show_help()elif user_input.lower() == 'status':self._show_status()elif user_input.lower() == 'tools':self._show_tools()elif user_input.lower().startswith('memory '):query = user_input[7:]self._show_memory(query)elif user_input:await self._process_message(user_input)except KeyboardInterrupt:print("\n👋 再见！")breakexcept Exception as e:print(f"❌ 错误: {e}")def _show_help(self):"""显示帮助信息"""help_text = """
🔧 可用命令:
- help: 显示此帮助信息
- status: 显示Agent状态
- tools: 显示可用工具
- memory <查询>: 搜索记忆
- quit/exit: 退出系统💬 直接输入您的需求即可开始对话。📌 示例:
- "帮我搜索人工智能的最新发展"
- "计算 123 + 456 * 789"
- "帮我分析这个数据文件""""print(help_text)def _show_status(self):"""显示状态"""status = self.agent.get_status()print(f"🤖 Agent状态:")print(f"   可用工具: {len(status['available_tools'])}")print(f"   完成任务: {status['completed_tasks']}")print(f"   记忆统计: {status['memory_stats']}")def _show_tools(self):"""显示工具"""tools = self.agent.tool_registry.list_tools()print("🛠️ 可用工具:")for name, info in tools.items():print(f"   {name}: {info['description']}")def _show_memory(self, query: str):"""显示记忆搜索结果"""memories = self.agent.get_memory_context(query)print(f"🧠 记忆搜索 '{query}' 结果:")for mem in memories:print(f"   [{mem['memory_type']}] {mem['content'][:100]}...")async def _process_message(self, message: str):"""处理消息"""print("🤔 正在思考...")result = await self.agent.process_request(message, "user")print(f"🤖 Agent: {result['response']}")if result['execution_details']:print(f"📋 执行了 {result['tasks_executed']} 个任务:")for detail in result['execution_details']:status_emoji = "✅" if detail['status'] == "completed" else "❌"print(f"   {status_emoji} {detail['task_name']}")# ===== 主程序 =====def start_web_server():"""启动Web服务"""logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s')print("🚀 启动AI Agent系统Web服务...")print("📚 API文档: http://localhost:8000/docs")print("🔍 系统状态: http://localhost:8000/status")print("🛠️ 可用工具: http://localhost:8000/tools")uvicorn.run(app, host="0.0.0.0", port=8000)async def start_cli():"""启动命令行界面"""logging.basicConfig(level=logging.WARNING)cli = AgentCLI()await cli.run_interactive()def main():"""主程序入口"""import sysif len(sys.argv) > 1 and sys.argv[1] == "web":start_web_server()else:asyncio.run(start_cli())if __name__ == "__main__":main()