第02章 Agent的三要素：感知、思考与执行
第02章 Agent的三要素：感知、思考与执行
“一个真正的Agent就像一个新员工：你给它一个目标，它知道怎么利用手头的资源把事情搞定，而不是每隔5分钟来问你下一步该怎么做。” —— 某工程团队负责人
上一章我们理解了Agent是什么，以及它和Chatbot的根本区别。
这一章，我们拆解Agent的内部结构：它是怎么工作的？每个部分在做什么？
2.1 感知层：Agent如何理解世界
# 感知层：Agent可以接收的输入类型
class AgentPerception:
"""
感知层负责：把各种输入转化为Agent可以理解的"状态"
"""
INPUT_TYPES = {
"文本指令": {
"示例": "帮我添加一个子域名 blog.example.com",
"处理方式": "直接传入LLM，提取意图和参数",
},
"工具执行结果": {
"示例": '{"success": true, "record_id": "abc123", "message": "DNS记录已创建"}',
"处理方式": "格式化后追加到对话历史，影响下一步决策",
},
"环境状态": {
"示例": "当前系统时间、用户的账户计划、工具的速率限制",
"处理方式": "注入System Prompt或作为工具调用的前置上下文",
},
"错误和异常": {
"示例": '{"error": "403 Forbidden", "message": "Token权限不足"}',
"处理方式": "触发错误处理路径，Agent需要决定重试/降级/报告",
},
"用户反馈": {
"示例": "不对，我需要的是CNAME记录，不是A记录",
"处理方式": "更新任务理解，重新规划",
},
}
def build_context(
self,
user_goal: str,
available_tools: list[dict],
current_skills: list[str],
conversation_history: list[dict],
) -> list[dict]:
"""
构建Agent的完整上下文
这是感知层的核心产出
"""
system_message = {
"role": "system",
"content": f"""你是一个自主AI助手，可以使用工具完成任务。
可用工具：
{self._format_tools(available_tools)}
已掌握技能（过去成功完成的任务）：
{', '.join(current_skills) if current_skills else '暂无'}
执行原则：
- 不确定时先查询，再操作
- 每次工具调用后验证结果
- 出错时说明原因并尝试恢复
- 完成后总结执行结果
"""
}
return [system_message] + conversation_history + [
{"role": "user", "content": user_goal}
]
def _format_tools(self, tools: list[dict]) -> str:
return "\n".join(
f"- {t['name']}: {t['description']}"
for t in tools
)
# 感知的质量决定Agent的质量
PERCEPTION_QUALITY_FACTORS = {
"上下文完整性": "Agent需要知道目标、工具、当前状态",
"噪音过滤": "不相关的信息会导致Agent分心或出错",
"格式一致性": "工具输出需要标准化，LLM才能稳定解析",
"错误可读性": "错误信息要对LLM友好，而不只是技术错误码",
}
2.2 思考层：Agent如何规划和决策
# 思考层：Agent的推理引擎
class AgentReasoning:
"""
思考层负责：
1. 理解用户真实意图（不只是字面意思）
2. 制定行动计划（选择工具、确定顺序）
3. 处理不确定性（何时询问用户、何时自己决定）
4. 从失败中恢复（识别错误类型、选择恢复策略）
"""
# === 意图理解 ===
INTENT_LAYERS = {
"表层意图": "用户说了什么",
"深层意图": "用户真正想要什么",
"隐含约束": "用户没说但理所当然的要求",
}
# 示例：
# 用户说："把我的博客部署上去"
# 表层意图：执行部署操作
# 深层意图：让博客对外可访问，并且正常运行
# 隐含约束：不要把现有的数据清空；尽量不要停机；出错先告诉我
# === 规划方法 ===
PLANNING_PATTERNS = {
"直接执行（简单任务）": {
"适用": "1-2个步骤，工具调用很确定",
"示例": "查询当前DNS记录",
"代码": """
response = await llm.complete(messages, tools=tools)
# 直接执行tool_call
""",
},
"Sequential（顺序规划）": {
"适用": "步骤之间有依赖，必须顺序执行",
"示例": "获取zone_id → 添加记录 → 验证生效",
"代码": """
# Agent在每步的tool_call结果后重新规划下一步
""",
},
"Plan-then-Execute（先规划后执行）": {
"适用": "复杂任务，需要全局视角",
"示例": "迁移整个DNS配置",
"代码": """
# 第一次调用：让LLM生成完整计划（不执行工具）
# 审查计划后再执行
""",
},
"ReAct（交替推理和行动）": {
"适用": "探索性任务，结果不可预测",
"示例": "排查为什么某个子域名不生效",
"代码": """
# 每步: Thought → Action → Observation → Thought → ...
""",
},
}
async def reason(
self,
context: list[dict],
tools: list,
previous_results: list = None,
) -> dict:
"""
核心推理：决定下一步做什么
"""
# 注入已有结果
if previous_results:
context.append({
"role": "assistant",
"content": f"已完成步骤: {previous_results}"
})
response = await self.llm.complete(
messages=context,
tools=tools,
tool_choice="auto", # 让LLM自己决定用不用工具
)
return {
"tool_calls": response.tool_calls, # 要执行的工具
"thinking": self._extract_thinking(response.content), # 推理过程
"final_answer": response.content if not response.tool_calls else None,
}
def _extract_thinking(self, content: str) -> str:
"""从<思考>标签中提取推理过程"""
import re
match = re.search(r"<思考>(.*?)</思考>", content, re.DOTALL)
return match.group(1).strip() if match else ""
# 思考质量的关键因素
THINKING_QUALITY = {
"明确性": "Agent需要清楚地知道当前目标是什么",
"步骤分解": "复杂目标需要拆成可执行的小步骤",
"不确定性处理": "不确定时询问，而不是猜测后执行",
"错误识别": "能识别出哪些结果是错误的，哪些是正常的",
}
2.3 执行层：Agent如何与世界交互
# 执行层：工具执行引擎
import asyncio
from typing import Any
class AgentExecutor:
"""
执行层负责：
1. 解析LLM的tool_call请求
2. 调用对应工具
3. 处理执行结果（包括错误）
4. 把结果格式化返回给LLM
"""
def __init__(self, tools: dict):
self.tools = tools # {tool_name: tool_function}
async def execute_tool_call(self, tool_call: dict) -> dict:
"""
执行单个工具调用
"""
tool_name = tool_call["name"]
arguments = tool_call["arguments"]
if tool_name not in self.tools:
return {
"success": False,
"error": f"工具 '{tool_name}' 不存在",
"available_tools": list(self.tools.keys()),
}
try:
tool_fn = self.tools[tool_name]
result = await tool_fn(**arguments)
return {
"success": True,
"tool": tool_name,
"arguments": arguments,
"result": result,
}
except TypeError as e:
# 参数不对
return {
"success": False,
"error": f"参数错误: {str(e)}",
"hint": "检查参数名称和类型是否正确",
}
except Exception as e:
# 执行出错
return {
"success": False,
"error": str(e),
"error_type": type(e).__name__,
}
async def execute_all(self, tool_calls: list[dict]) -> list[dict]:
"""
执行多个工具调用
注意：某些工具必须顺序执行（有依赖关系），某些可以并行
"""
results = []
for call in tool_calls:
result = await self.execute_tool_call(call)
results.append(result)
# 如果关键步骤失败，停止后续执行
if not result["success"] and call.get("critical", True):
results.append({
"stopped": True,
"reason": "关键步骤失败，已中止后续操作",
})
break
return results
def format_results_for_llm(self, results: list[dict]) -> str:
"""
把执行结果格式化，让LLM容易理解
"""
formatted = []
for r in results:
if r.get("success"):
formatted.append(f"✅ {r['tool']}({r['arguments']})\n 结果: {r['result']}")
elif r.get("stopped"):
formatted.append(f"⏹️ 执行已中止: {r['reason']}")
else:
formatted.append(f"❌ 执行失败: {r['error']}")
return "\n".join(formatted)
2.4 三层结构的完整循环
# 把三层组合在一起：一个完整的Agent运行一次任务
async def run_agent_task(
goal: str,
tools: dict,
max_iterations: int = 10,
) -> str:
"""
完整的Agent执行循环
"""
perception = AgentPerception()
reasoning = AgentReasoning(llm)
executor = AgentExecutor(tools)
# 构建初始上下文
context = perception.build_context(
user_goal=goal,
available_tools=list(tools.values()),
current_skills=[],
conversation_history=[],
)
print(f"🎯 目标: {goal}")
print("=" * 50)
for iteration in range(max_iterations):
print(f"\n🔄 迭代 {iteration + 1}")
# 思考：决定下一步
decision = await reasoning.reason(context, list(tools.values()))
if decision["thinking"]:
print(f"💭 思考: {decision['thinking']}")
# 检查是否完成
if decision["final_answer"]:
print(f"\n✅ 任务完成")
return decision["final_answer"]
# 执行：调用工具
if decision["tool_calls"]:
results = await executor.execute_all(decision["tool_calls"])
formatted_results = executor.format_results_for_llm(results)
print(f"🔧 执行结果:\n{formatted_results}")
# 把结果加入上下文
context.append({
"role": "tool",
"content": formatted_results,
})
return "已达到最大迭代次数"
# 运行示例
# asyncio.run(run_agent_task(
# goal="帮我把 blog.example.com 的DNS指向 1.2.3.4，并验证生效",
# tools={"dns_add": add_dns_record, "dns_verify": verify_dns},
# ))
本章小结
五个核心认知：
-
感知层决定Agent"看到什么"：System Prompt的质量、工具描述的清晰度、错误信息的可读性——都直接影响Agent的行为；垃圾进，垃圾出
-
思考层是Agent的"大脑"：同一个LLM，加上好的推理框架（ReAct、CoT）和清晰的目标，表现会天差地别；思考层的设计是Agent工程最关键的部分
-
执行层需要防御性设计：工具执行会出错；Agent需要能够识别错误类型（权限问题、参数问题、网络问题）并做出不同的恢复策略
-
三层的质量相互依赖：感知不清楚→思考会偏；思考不好→执行会错；执行不稳定→思考无法正确判断；需要整体设计，不能只优化某一层
-
迭代循环是Agent的本质：不是一次性的感知-思考-执行，而是循环，直到目标达成；这也是为什么"最大迭代次数"和"完成检测"是必须设计的安全机制
核心行动：
# 今天的实验：
# 1. 选一个你日常用LLM辅助的任务
# 2. 画出它的"感知-思考-执行"循环图
# 3. 标出每层的输入和输出
# 4. 找出哪层是当前的瓶颈（为什么需要人工干预）
本章提示词模板
模板一：分析任务的三层结构
我有一个任务：[描述你的任务]
请帮我分析这个任务的Agent三层结构：
感知层分析：
- Agent需要知道哪些信息才能开始执行？
- 哪些信息是已知的？哪些需要先查询获取？
- 执行过程中会有哪些中间状态需要感知？
思考层分析：
- 这个任务需要哪种规划模式？（直接执行/顺序/ReAct）
- 可能的决策分支是什么？（成功路径 vs 失败恢复路径）
- 什么情况下Agent应该暂停并询问用户？
执行层分析：
- 需要哪些工具？（列出工具名和功能）
- 工具之间有什么依赖关系？
- 哪些步骤的失败是可恢复的？哪些是不可逆的？
模板二：设计Agent的System Prompt
我要给一个Agent设计System Prompt，Agent的用途是：[描述用途]
Agent可以使用的工具：[列出工具]
目标用户：[描述用户]
典型任务类型：[举3个例子]
必须避免的操作：[列出危险操作]
请帮我写一个高质量的System Prompt，要求：
1. 明确Agent的角色和能力边界
2. 说明决策原则（何时用工具、何时询问用户）
3. 错误处理规范（失败时如何报告）
4. 安全约束（哪些事情绝对不做）
5. 输出格式要求
然后指出这个System Prompt可能的弱点和改进建议。