Agentic AI 基础

概述

Agentic AI（智能体AI）代表人工智能从被动响应工具向主动规划、自主决策系统的根本性范式转变。与传统AI助手不同，智能体AI能够：

• 主动规划：将复杂任务分解为可执行的子步骤
• 持续记忆：保持跨对话的上下文状态
• 自主行动：在环境中执行操作并根据反馈迭代
• 工具使用：调用外部工具和API扩展能力边界

“Agentic AI系统被定义为具有主动规划、上下文记忆和目标导向行为的自主系统，能够在复杂环境中独立完成多步骤任务。” — Stanford HAI Agent Survey 2025

---

Agentic AI vs 传统AI

核心区别

维度	传统AI助手	Agentic AI
交互模式	被动响应	主动规划
任务范围	单轮问答	多步任务执行
记忆能力	无状态	持久记忆
错误恢复	无	自我反思与修正
工具使用	有限或无	动态工具调用
自主程度	低	高

典型应用场景

传统AI助手：
用户: "帮我写一封邮件"
助手: "以下是邮件内容..."

Agentic AI：
用户: "帮我联系供应商并确定下周的项目会议时间"
Agent: 
  1. 搜索供应商联系人
  2. 起草多封邮件草稿
  3. 检查日历可用性
  4. 发送邀请并跟踪确认
  5. 将会议详情添加到项目管理系统

---

双范式框架

根据最新研究（Springer AI Review 2025），Agentic AI可分为两大范式：

1. 符号/古典智能体 (Symbolic/Classical Agents)

核心特征：

• 基于算法规划（PDDL、HTN规划）
• 持久状态管理
• 确定性推理
• 可解释性强

代表系统：

• 经典规划器（Gripper、Blocksworld）
• 知识图谱增强智能体
• 混合专家系统

# 简化的符号规划伪代码
class SymbolicAgent:
    def __init__(self):
        self.state = WorldState()
        self.planner = PDDLPlanner()
        self.knowledge_base = KnowledgeGraph()
    
    def execute_task(self, goal):
        plan = self.planner.plan(self.state, goal)
        for action in plan:
            self.state = action.execute(self.state)
            self.verify_preconditions()

2. 神经/生成式智能体 (Neural/Generative Agents)

核心特征：

• 基于LLM的推理
• 随机生成与探索
• 提示驱动的行为编排
• 灵活性高但可解释性较低

代表系统：

• ReAct智能体
• AutoGPT/BabyAGI
• Voyager/MinerU

# 神经智能体核心循环
class NeuralAgent:
    def __init__(self, llm, tools, memory):
        self.llm = llm
        self.tools = tools
        self.memory = memory
    
    def run(self, task, max_steps=10):
        observation = ""
        for step in range(max_steps):
            # 思考阶段
            thought = self.llm.think(task, observation, self.memory)
            
            # 行动选择
            if thought.needs_tool:
                result = self.tools.execute(thought.tool, thought.args)
                observation += f"Tool result: {result}\n"
            else:
                return thought.response
            
            # 更新记忆
            self.memory.add(thought, result)
            
            # 自我反思
            if self.should_reflect():
                self.reflect_and_correct()

---

核心组件

1. 规划模块 (Planning)

层次化规划：

高层目标
    ↓
┌─────────────────────────────────────┐
│     任务分解 (Task Decomposition)     │
└─────────────────────────────────────┘
    ↓
子任务1  子任务2  子任务3  ...
    ↓       ↓       ↓
行动1   行动2   行动3   ...
    ↓       ↓       ↓
执行反馈 → 重新规划（如需）

主要方法：

• Chain-of-Thought：显式推理链
• Tree-of-Thought：探索多路径
• Graph-of-Thought：复杂依赖建模
• LLM Compiler：并行规划执行

2. 记忆系统 (Memory)

记忆类型	容量	持久性	用途
感官记忆	短	瞬时	原始感知输入
工作记忆	中	当前会话	推理中间结果
情景记忆	长	跨会话	经验存储
语义记忆	无限	永久	知识存储

记忆实现示例：

class MemorySystem:
    def __init__(self):
        self.sensory = deque(maxlen=100)  # 原始输入
        self.working = {}                   # 当前推理
        self.episodic = VectorDB()          # 经验向量化
        self.semantic = KnowledgeGraph()     # 知识图谱
    
    def remember(self, experience, memory_type="episodic"):
        embedding = self.encode(experience)
        
        if memory_type == "sensory":
            self.sensory.append(experience)
        elif memory_type == "working":
            self.working[experience.id] = experience
        elif memory_type == "episodic":
            self.episodic.insert(embedding, experience)
        elif memory_type == "semantic":
            self.semantic.add(experience)
    
    def retrieve(self, query, memory_type="all", k=5):
        query_emb = self.encode(query)
        
        if memory_type == "all":
            results = []
            results.extend(self.episodic.search(query_emb, k))
            results.extend(self.semantic.search(query_emb, k))
            return self.rerank(results)
        elif memory_type == "episodic":
            return self.episodic.search(query_emb, k)

3. 工具系统 (Tool Use)

工具分类：

• 计算工具：计算器、代码执行器
• 信息工具：搜索、数据库查询、API调用
• 操作工具：文件操作、发送消息、控制系统
• 知识工具：知识图谱检索、RAG

工具调用框架：

from typing import Callable, Any
from dataclasses import dataclass
 
@dataclass
class Tool:
    name: str
    description: str
    parameters: dict
    function: Callable
 
class ToolSystem:
    def __init__(self):
        self.tools: dict[str, Tool] = {}
        self.execution_history = []
    
    def register(self, tool: Tool):
        self.tools[tool.name] = tool
    
    async def execute(self, tool_name: str, **kwargs) -> Any:
        if tool_name not in self.tools:
            raise ValueError(f"Unknown tool: {tool_name}")
        
        tool = self.tools[tool_name]
        
        # 参数验证
        self.validate_params(tool.parameters, kwargs)
        
        # 执行
        result = await tool.function(**kwargs)
        
        # 记录
        self.execution_history.append({
            "tool": tool_name,
            "input": kwargs,
            "output": result,
            "timestamp": time.time()
        })
        
        return result
    
    def get_available_tools(self, task_context: str) -> list[Tool]:
        """根据上下文返回相关工具"""
        # 可使用LLM进行工具选择
        relevant_tools = []
        for tool in self.tools.values():
            if self.is_relevant(tool, task_context):
                relevant_tools.append(tool)
        return relevant_tools

---

评估方法

1. 基准测试

基准	关注点	代表任务
GAIA	真实世界助手	多步骤网页任务
AgentBench	多领域智能体	代码、数据分析
WebArena	网页导航	真实网站操作
MiniWob++	微操作	简单UI交互
ALFWorld	家庭任务	物体操作推理

2. 评估维度

核心指标：

• 任务完成率：是否成功达成目标
• 效率：完成任务所需的步骤/时间
• 鲁棒性：对干扰的抵抗能力
• 可复用性：跨任务泛化能力
• 安全性：对恶意输入的防护

class AgentEvaluator:
    def evaluate(self, agent, benchmark, num_episodes=100):
        results = {
            "success_rate": [],
            "avg_steps": [],
            "avg_time": [],
            "error_types": defaultdict(int)
        }
        
        for episode in range(num_episodes):
            task = benchmark.sample_task()
            trace = agent.run(task)
            
            # 评估
            results["success_rate"].append(trace.success)
            results["avg_steps"].append(len(trace.steps))
            results["avg_time"].append(trace.total_time)
            
            # 错误分析
            for error in trace.errors:
                results["error_types"][error.type] += 1
        
        return self.summarize(results)

---

架构设计原则

1. 模块化与可组合性

┌──────────────────────────────────────────────────────┐
│                    Agent Controller                    │
│  (Orchestration, State Management, Error Handling)     │
└────────────────────────┬─────────────────────────────┘
                         │
         ┌───────────────┼───────────────┐
         ↓               ↓               ↓
    ┌─────────┐    ┌─────────┐    ┌─────────┐
    │Planner  │    │ Memory  │    │  Tools  │
    │         │    │         │    │         │
    │- CoT    │    │- Work   │    │- Search │
    │- ToT    │    │- Episodic│    │- Code  │
    │- Graph  │    │- Semantic│    │- API  │
    └─────────┘    └─────────┘    └─────────┘

2. 容错与恢复

class ResilientAgent:
    def __init__(self, max_retries=3):
        self.max_retries = max_retries
    
    async def execute_with_retry(self, action_fn, *args, **kwargs):
        for attempt in range(self.max_retries):
            try:
                return await action_fn(*args, **kwargs)
            except TransientError as e:
                # 短暂错误，等待后重试
                await asyncio.sleep(2 ** attempt)
                continue
            except FatalError as e:
                # 致命错误，触发恢复
                await self.recover_from_error(e)
                raise
            except MaxRetriesExceeded:
                # 回退策略
                return await self.fallback_strategy()
    
    async def recover_from_error(self, error):
        """错误恢复机制"""
        # 1. 保存当前状态
        checkpoint = self.save_checkpoint()
        
        # 2. 分析错误原因
        error_analysis = self.analyze_error(error)
        
        # 3. 选择恢复策略
        if error_analysis.type == "tool_failure":
            await self.replace_tool()
        elif error_analysis.type == "planning_failure":
            await self.replan()
        elif error_analysis.type == "memory_corruption":
            await self.restore_memory()

---

挑战与局限

1. 主要挑战

挑战	描述	当前解决方案
规划复杂性	长 horizon 任务规划困难	层次化分解、课程学习
幻觉问题	LLM生成错误规划	验证机制、外部知识
计算成本	多步推理开销大	缓存、提前终止
安全风险	自主行为潜在危害	沙盒、权限控制
评估困难	开放任务难以评测	多维度基准

2. 安全考量

Agentic AI的自主性带来了新的安全挑战：

• 目标对齐：确保智能体追求正确的目标
• 权限控制：限制关键操作的执行
• 审计追溯：记录所有决策和行动
• 紧急停止：随时中止智能体行为

class SecureAgentWrapper:
    def __init__(self, agent, safety_config):
        self.agent = agent
        self.safety = SafetyMonitor(safety_config)
    
    async def run(self, task):
        # 前置检查
        if not self.safety.preflight_check(task):
            raise SafetyViolation("Preflight check failed")
        
        # 带监控的执行
        async for step in self.agent.run_streaming(task):
            if not self.safety.check_step(step):
                await self.safety.intervene()
                break
            yield step
        
        # 后置审计
        self.safety.audit(self.agent.execution_trace)

---

参考文献