4.1 Agent基础概念
文档摘要
4.1 Agent基础概念 导读：Agent（智能体）是大模型应用的高级形态，它不仅能理解和生成文本，还能感知环境、使用工具、自主规划和执行复杂任务。本节将全面介绍Agent的基础概念、设计原则和核心能力。 4.1.1 什么是Agent 从Chatbot到Agent的演进 大模型应用的发展经历了三个阶段：从简单的对话机器人（Chatbot），到RAG增强的知识助手，再到具备自主行动能力的Agent。Agent代表了当前大模型应用的最前沿形态。 Agent的定义 Agent是一个以大语言模型为核心"大脑"，具备感知、推理、规划和行动能力的自主智能系统。
4.1 Agent基础概念
导读：Agent（智能体）是大模型应用的高级形态，它不仅能理解和生成文本，还能感知环境、使用工具、自主规划和执行复杂任务。本节将全面介绍Agent的基础概念、设计原则和核心能力。
4.1.1 什么是Agent
从Chatbot到Agent的演进
大模型应用的发展经历了三个阶段：从简单的对话机器人（Chatbot），到RAG增强的知识助手，再到具备自主行动能力的Agent。Agent代表了当前大模型应用的最前沿形态。
# 三种应用形态对比 APPLICATION_FORMS = """ ┌─────────────┬──────────────────┬──────────────────────┬─────────────────────┐ │ 维度 │ Chatbot │ RAG助手 │ Agent │ ├─────────────┼──────────────────┼──────────────────────┼─────────────────────┤ │ 核心能力 │ 对话生成 │ 对话+知识检索 │ 对话+检索+工具+规划 │ │ 知识来源 │ 仅模型内置知识 │ 模型+外部知识库 │ 模型+知识+实时数据 │ │ 行动能力 │ 无 │ 无 │ 调用工具、执行操作 │ │ 自主性 │ 无 │ 无 │ 自主规划和决策 │ │ 典型应用 │ 客服问答 │ 企业知识库问答 │ 自动化任务执行 │ └─────────────┴──────────────────┴──────────────────────┴─────────────────────┘ """ print(APPLICATION_FORMS)
Agent的定义
Agent是一个以大语言模型为核心"大脑"，具备感知、推理、规划和行动能力的自主智能系统。它能够：
- 理解任务：解析用户指令，明确需要达成的目标
- 制定计划：将复杂任务分解为可执行的子任务序列
- 使用工具：调用外部API、执行代码、访问数据库等
- 观察反馈：根据工具执行结果调整后续策略
- 迭代优化：在执行过程中持续评估和改进
# Agent核心循环 class AgentCoreLoop: """Agent的核心推理-行动循环（ReAct模式）""" def __init__(self, llm, tools): self.llm = llm self.tools = {t.name: t for t in tools} self.max_iterations = 10 def run(self, task: str) -> str: """执行Agent核心循环""" context = [{"role": "user", "content": task}] for iteration in range(self.max_iterations): print(f"--- 迭代 {iteration + 1} ---") # 推理：让LLM思考下一步行动 response = self.llm.chat.completions.create( model="gpt-4-turbo-preview", messages=context, temperature=0.2 ) thought = response.choices[0].message.content print(f"思考: {thought[:200]}") # 检查是否需要调用工具 tool_call = self._parse_tool_call(thought) if tool_call is None: # 不需要工具，直接返回结果 return thought # 行动：执行工具 tool_name = tool_call["name"] tool_args = tool_call["arguments"] print(f"行动: 调用工具 {tool_name}({tool_args})") if tool_name in self.tools: result = self.tools[tool_name].execute(**tool_args) observation = f"工具 {tool_name} 执行结果: {result}" else: observation = f"错误: 未知工具 {tool_name}" print(f"观察: {observation[:200]}") # 将观察结果加入上下文，继续下一轮循环 context.append({"role": "assistant", "content": thought}) context.append({"role": "user", "content": f"观察结果: {observation}\n请继续思考下一步。"}) return "已达到最大迭代次数，任务未完成。" def _parse_tool_call(self, thought: str) -> dict: """解析LLM输出中的工具调用指令""" # 简化版解析，实际使用function calling import re import json pattern = r'Action:\s*(\w+)\s*\((.+?)\)' match = re.search(pattern, thought) if match: try: args = json.loads(match.group(2)) return {"name": match.group(1), "arguments": args} except json.JSONDecodeError: return None return None
4.1.2 Agent的设计原则
原则一：目标导向
Agent的所有行为都应围绕明确的用户目标展开。设计Agent时首先要定义清晰的目标函数。
# 目标定义框架 from dataclasses import dataclass, field from typing import List, Optional from enum import Enum class TaskStatus(Enum): PENDING = "pending" IN_PROGRESS = "in_progress" COMPLETED = "completed" FAILED = "failed" @dataclass class AgentGoal: """Agent目标定义""" description: str success_criteria: List[str] constraints: List[str] = field(default_factory=list) priority: int = 1 max_steps: int = 10 timeout_seconds: int = 300 @dataclass class TaskResult: """任务执行结果""" goal: AgentGoal status: TaskStatus output: str = "" steps_taken: int = 0 tools_used: List[str] = field(default_factory=list) error_message: str = "" # 示例：定义一个数据分析Agent的目标 data_analysis_goal = AgentGoal( description="分析用户提供的销售数据，找出Top 3畅销产品", success_criteria=[ "识别出销量最高的3个产品", "提供每个产品的销量数据和占比", "给出简短的趋势分析" ], constraints=[ "只使用提供的数据进行分析", "分析结果必须基于数据事实", "避免推测未经证实的结论" ], priority=1, max_steps=8, timeout_seconds=120 )
原则二：可观测性
Agent的决策过程必须可观测、可追踪。每一步操作都应该有日志记录。
# Agent可观测性框架 import time import json from datetime import datetime from typing import Dict, Any, List class AgentLogger: """Agent执行日志记录器""" def __init__(self, session_id: str): self.session_id = session_id self.logs: List[Dict] = [] self.start_time = time.time() def log_thought(self, iteration: int, thought: str): """记录Agent的思考过程""" entry = { "timestamp": datetime.now().isoformat(), "session_id": self.session_id, "iteration": iteration, "type": "thought", "content": thought[:1000] } self.logs.append(entry) def log_action(self, iteration: int, tool_name: str, arguments: Dict): """记录Agent执行的操作""" entry = { "timestamp": datetime.now().isoformat(), "session_id": self.session_id, "iteration": iteration, "type": "action", "tool": tool_name, "arguments": arguments } self.logs.append(entry) def log_observation(self, iteration: int, result: Any, success: bool): """记录操作结果""" entry = { "timestamp": datetime.now().isoformat(), "session_id": self.session_id, "iteration": iteration, "type": "observation", "result_preview": str(result)[:500], "success": success } self.logs.append(entry) def get_summary(self) -> Dict: """获取执行摘要""" total_time = time.time() - self.start_time thoughts = [l for l in self.logs if l["type"] == "thought"] actions = [l for l in self.logs if l["type"] == "action"] observations = [l for l in self.logs if l["type"] == "observation"] return { "session_id": self.session_id, "total_iterations": len(thoughts), "total_actions": len(actions), "total_time_seconds": round(total_time, 2), "success_rate": sum(1 for o in observations if o["success"]) / max(len(observations), 1) } def export_trace(self) -> str: """导出完整执行轨迹""" return json.dumps(self.logs, indent=2, ensure_ascii=False)
原则三：安全可控
Agent必须具备安全边界和人工干预机制。
# Agent安全控制 class AgentSafetyGuard: """Agent安全防护""" # 危险操作白名单（需要人工确认） DANGEROUS_OPERATIONS = [ "delete_database", "send_email", "execute_sql", "modify_production", "deploy_code", "transfer_funds" ] # 资源限制 MAX_TOOL_CALLS_PER_TASK = 20 MAX_TOKENS_PER_TASK = 50000 MAX_EXECUTION_TIME = 300 # 秒 def __init__(self, require_approval=True): self.require_approval = require_approval self.tool_call_count = 0 self.token_usage = 0 def check_tool_safety(self, tool_name: str, arguments: dict) -> dict: """检查工具调用的安全性""" result = {"allowed": True, "reason": "", "requires_approval": False} if tool_name in self.DANGEROUS_OPERATIONS: result["requires_approval"] = True result["allowed"] = not self.require_approval result["reason"] = f"工具 {tool_name} 属于高风险操作" if self.tool_call_count >= self.MAX_TOOL_CALLS_PER_TASK: result["allowed"] = False result["reason"] = "已达到工具调用上限" return result def record_tool_call(self, token_usage: int = 0): """记录工具调用""" self.tool_call_count += 1 self.token_usage += token_usage
4.1.3 Agent的核心能力
能力一：自然语言理解
Agent通过大模型理解用户的自然语言指令，将模糊的需求转化为精确的任务描述。
# 任务理解模块 from openai import OpenAI from pydantic import BaseModel class TaskUnderstanding(BaseModel): """结构化任务理解""" intent: str entities: dict constraints: list expected_output: str class TaskParser: """任务解析器""" def __init__(self, llm_client: OpenAI): self.client = llm_client def parse(self, user_input: str) -> TaskUnderstanding: """将用户输入解析为结构化任务""" prompt = f"""请分析以下用户指令，提取关键信息。 用户指令: {user_input} 请以JSON格式返回： - intent: 用户意图（一句话描述） - entities: 提到的实体（人名、日期、地点等） - constraints: 约束条件 - expected_output: 期望的输出形式 只返回JSON，不要其他文字。""" response = self.client.chat.completions.create( model="gpt-4-turbo-preview", messages=[{"role": "user", "content": prompt}], temperature=0.1 ) import json try: data = json.loads(response.choices[0].message.content) return TaskUnderstanding(**data) except (json.JSONDecodeError, TypeError): return TaskUnderstanding( intent=user_input, entities={}, constraints=[], expected_output="文本回答" )
能力二：工具使用
工具使用是Agent区别于普通Chatbot的核心能力。Agent可以根据需要调用各种外部工具来完成任务。
# 工具定义框架 from abc import ABC, abstractmethod from typing import Dict, Any class Tool(ABC): """工具基类""" @property @abstractmethod def name(self) -> str: """工具名称""" pass @property @abstractmethod def description(self) -> str: """工具描述（供Agent理解工具用途）""" pass @property @abstractmethod def parameters_schema(self) -> Dict: """参数JSON Schema""" pass @abstractmethod def execute(self, **kwargs) -> Any: """执行工具""" pass def to_function_spec(self) -> Dict: """转换为OpenAI function calling格式""" return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": self.parameters_schema } } # 具体工具示例 class WebSearchTool(Tool): """网络搜索工具""" @property def name(self): return "web_search" @property def description(self): return "搜索互联网上的信息，返回相关结果列表" @property def parameters_schema(self): return { "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词"}, "max_results": {"type": "integer", "description": "最大返回结果数", "default": 5} }, "required": ["query"] } def execute(self, query: str, max_results: int = 5) -> str: """执行搜索""" # 实际实现中调用搜索API return f"搜索 '{query}' 的结果: [模拟结果1, 模拟结果2, ...]" class CodeExecutorTool(Tool): """代码执行工具""" @property def name(self): return "execute_code" @property def description(self): return "执行Python代码并返回输出结果，可用于数学计算、数据处理等" @property def parameters_schema(self): return { "type": "object", "properties": { "code": {"type": "string", "description": "要执行的Python代码"}, "timeout": {"type": "integer", "description": "执行超时时间(秒)", "default": 30} }, "required": ["code"] } def execute(self, code: str, timeout: int = 30) -> str: """执行代码""" import subprocess try: result = subprocess.run( ["python3", "-c", code], capture_output=True, text=True, timeout=timeout ) if result.returncode == 0: return result.stdout[:2000] else: return f"执行错误: {result.stderr[:500]}" except subprocess.TimeoutExpired: return "执行超时" class DatabaseQueryTool(Tool): """数据库查询工具""" @property def name(self): return "query_database" @property def description(self): return "查询数据库并返回结果，仅支持SELECT语句" @property def parameters_schema(self): return { "type": "object", "properties": { "sql": {"type": "string", "description": "SQL查询语句"} }, "required": ["sql"] } def execute(self, sql: str) -> str: """执行SQL查询""" # 安全检查 sql_upper = sql.strip().upper() if not sql_upper.startswith("SELECT"): return "错误: 只允许SELECT查询" # 实际实现中连接数据库执行 return f"查询结果: [{sql} 的模拟结果]"
能力三：规划与推理
Agent能够将复杂任务分解为可执行的步骤序列。
# 任务规划器 from typing import List from dataclasses import dataclass @dataclass class PlanStep: """规划步骤""" step_number: int action: str tool_required: str expected_output: str dependencies: List[int] = None def __post_init__(self): self.dependencies = self.dependencies or [] class TaskPlanner: """任务规划器""" def __init__(self, llm_client): self.client = llm_client def create_plan(self, task: str, available_tools: List[Tool]) -> List[PlanStep]: """为任务创建执行计划""" tools_desc = "\n".join( f"- {t.name}: {t.description}" for t in available_tools ) prompt = f"""你是一个任务规划专家。请为以下任务制定执行计划。 任务: {task} 可用工具: {tools_desc} 请制定一个详细的执行计划，每个步骤包括： 1. 步骤编号 2. 执行动作 3. 需要使用的工具 4. 预期输出 5. 依赖的步骤编号 请以JSON数组格式输出。""" response = self.client.chat.completions.create( model="gpt-4-turbo-preview", messages=[{"role": "user", "content": prompt}], temperature=0.2 ) import json try: steps_data = json.loads(response.choices[0].message.content) return [PlanStep(**s) for s in steps_data] except (json.JSONDecodeError, TypeError) as e: print(f"规划解析失败: {e}") return [PlanStep(1, "分析任务", "无", "任务分析结果")] def evaluate_plan(self, plan: List[PlanStep]) -> Dict: """评估计划的可行性""" return { "total_steps": len(plan), "tools_needed": list(set(s.tool_required for s in plan if s.tool_required != "无")), "estimated_complexity": "high" if len(plan) > 5 else "medium" if len(plan) > 2 else "low" }
4.1.4 Agent架构模式
ReAct模式
ReAct（Reasoning + Acting）是最经典的Agent架构，交替进行推理和行动。
# ReAct Agent实现 class ReActAgent: """基于ReAct模式的Agent""" def __init__(self, llm, tools): self.llm = llm self.tools = {t.name: t for t in tools} self.system_prompt = """你是一个智能助手，可以使用工具来帮助用户完成任务。 你可以使用以下工具： {tools_desc} 请按以下格式回答： Thought: 思考当前情况和下一步行动 Action: 工具名称(参数) 或者直接给出最终答案： Thought: 我已经获得了足够信息 Answer: 最终答案""" def run(self, task: str, max_iterations: int = 8): """执行ReAct循环""" messages = [{"role": "user", "content": task}] for i in range(max_iterations): response = self.llm.chat.completions.create( model="gpt-4-turbo-preview", messages=messages, temperature=0.2 ) output = response.choices[0].message.content # 解析Thought和Action thought, action = self._parse_react_output(output) if action is None: # 没有Action，说明是最终答案 return thought # 执行Action result = self.tools[action["name"]].execute(**action["args"]) messages.append({"role": "assistant", "content": output}) messages.append({"role": "user", "content": f"工具返回: {result}\n请继续。"}) return "达到最大迭代次数" def _parse_react_output(self, text: str): """解析ReAct格式输出""" import re # 简化解析 thought_match = re.search(r'Thought:\s*(.+)', text) action_match = re.search(r'Action:\s*(\w+)\((.+?)\)', text) thought = thought_match.group(1) if thought_match else text action = None if action_match: try: action = {"name": action_match.group(1), "args": json.loads(action_match.group(2))} except: pass return thought, action
Plan-and-Execute模式
先制定完整计划，再逐步执行。
# Plan-and-Execute Agent class PlanExecuteAgent: """先规划后执行的Agent""" def __init__(self, llm, tools): self.llm = llm self.planner = TaskPlanner(llm) self.tools = {t.name: t for t in tools} def run(self, task: str): """执行Plan-and-Execute流程""" # 阶段1：规划 print("=== 阶段1: 制定计划 ===") plan = self.planner.create_plan(task, list(self.tools.values())) for step in plan: print(f" 步骤{step.step_number}: {step.action} (工具: {step.tool_required})") # 阶段2：执行 print("\n=== 阶段2: 执行计划 ===") results = {} for step in plan: print(f"\n执行步骤{step.step_number}: {step.action}") if step.tool_required and step.tool_required in self.tools: tool = self.tools[step.tool_required] result = tool.execute() # 简化，实际需要根据上下文传参 results[step.step_number] = result print(f" 结果: {str(result)[:200]}") else: results[step.step_number] = "无需工具" # 阶段3：汇总 print("\n=== 阶段3: 汇总结果 ===") return self._summarize(task, plan, results) def _summarize(self, task, plan, results): """汇总执行结果""" summary_parts = [f"任务: {task}", ""] for step in plan: r = results.get(step.step_number, "未执行") summary_parts.append(f"{step.step_number}. {step.action}: {str(r)[:100]}") return "\n".join(summary_parts)
本章小结
通过本节的学习，你将：
✅ 理解Agent的定义和从Chatbot到Agent的演进路径
✅ 掌握Agent设计的三大原则：目标导向、可观测性、安全可控
✅ 了解Agent的四大核心能力：自然语言理解、工具使用、规划推理、记忆管理
✅ 学会ReAct和Plan-and-Execute两种经典Agent架构模式
✅ 能够设计和实现基础的Agent系统
在4.2节中，我们将深入学习Agent的工具调用机制，在4.3节中将探讨更复杂的工作流编排技术。
[FAQ] 常见问题
Q: Agent和普通的函数调用有什么区别？
A: Agent的核心区别在于自主性。函数调用是预先编程好的，而Agent能够根据上下文自主决定调用哪些工具、传递什么参数、是否需要多步操作。Agent具备推理和规划能力。
Q: 如何控制Agent不做出危险操作？
A: 通过安全防护机制：工具调用权限控制、高风险操作需人工确认、资源使用上限设置、操作审计日志等。永远不要给Agent直接的生产环境权限。
Q: Agent需要多少Token才能工作？
A: 取决于任务复杂度和工具数量。简单的Agent任务可能每次对话消耗500-2000 Token，复杂的多步骤任务可能需要5000-20000 Token。建议做好Token使用监控。
Q: 什么时候应该用Agent而不是简单的RAG系统？
A: 当任务需要调用外部工具、执行多步骤操作、或需要动态决策时，使用Agent。如果只是简单的知识问答，RAG系统已经足够。
Q: 如何测试Agent的效果？
A: 需要从多个维度测试：任务完成率、工具调用准确率、推理合理性、安全合规性、Token效率等。建议构建标准化的评估数据集。
评论区
(0)
U