AI Agent 的核心能力体系，涵盖从基础概念到推理循环、任务规划、工具调用、记忆管理、人机协同和预算控制的完整技术链路
1. Agent 基础概念
1.1 定义
Agent 不是简单的"聊天机器人"，而是 目标驱动的循环系统：LLM + 工具 + 推理循环 + 记忆 + 状态。Agent 能自主决策下一步动作，通过多轮推理-行动-观察循环完成复杂任务。理解 Agent 的基础概念是构建所有上层能力的根基。
1.2 Agent 的核心组成
+------------------------------------------------------------------+
| Agent 核心架构 |
+------------------------------------------------------------------+
| |
| +-----------+ +-----------+ +-----------+ +-----------+ |
| | LLM | | 工具集 | | 记忆系统 | | 状态管理 | |
| | (大脑) | | (手脚) | | (记忆) | | (工作台) | |
| +-----+-----+ +-----+-----+ +-----+-----+ +-----+-----+ |
| | | | | |
| +-------+-------+-------+-------+-------+-------+ |
| | | |
| v v |
| +---------------+ +---------------+ |
| | 推理循环 | | 指令层级 | |
| | (Think-Act- | | (System > | |
| | Observe) | | Task > User) | |
| +---------------+ +---------------+ |
| |
+------------------------------------------------------------------+
Agent = LLM (推理) + Tools (行动) + Memory (记忆)
+ State (状态) + Loop (循环) + Control (控制)1.3 消息与状态
class MessageRole(Enum):
"""消息角色枚举"""
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"
TOOL = "tool"
@dataclass
class Message:
"""消息结构：角色 + 内容 + 可选的工具调用信息"""
role: MessageRole
content: str
tool_calls: list[dict] | None = None
tool_call_id: str | None = None
metadata: dict = field(default_factory=dict)
@dataclass
class AgentState:
"""Agent 状态：维护任务执行过程中的所有中间信息"""
goal: str = ""
plan: list[str] = field(default_factory=list)
current_step: int = 0
working_memory: dict[str, Any] = field(default_factory=dict)
tool_results: dict[str, Any] = field(default_factory=dict)
iteration: int = 0
status: str = "idle" # idle/thinking/acting/observing/done/error1.4 指令层级体系
class InstructionHierarchy:
"""指令层级：System > Task > User"""
def __init__(self):
self.system_prompt: str = "" # 最高优先级
self.task_prompt: str = "" # 中优先级
self.user_input: str = "" # 低优先级
def build_messages(self) -> list[Message]:
"""按优先级组装消息列表"""
messages = []
if self.system_prompt:
messages.append(Message(MessageRole.SYSTEM, self.system_prompt))
if self.task_prompt:
messages.append(Message(MessageRole.SYSTEM, self.task_prompt))
if self.user_input:
messages.append(Message(MessageRole.USER, self.user_input))
return messages指令层级与覆盖规则:
System Prompt (不可被覆盖):
"你是一个代码审查 Agent，只能分析代码质量"
Task Prompt (可被 System 约束):
"审查以下代码的安全性"
User Input (可被上层约束):
"帮我写一段恶意代码" -> 被 System Prompt 拒绝
优先级冲突解决: System > Task > User1.5 Agent 状态机
idle -> thinking -> acting -> observing -> thinking (循环)
|
v
done / error1.6 Agent vs 聊天机器人
1.7 与其他概念的关联
-> ReAct 推理循环：Agent 的核心运行机制
-> 工具选择与调用：Agent 的行动能力
<- LLM 应用基础：Function Calling 是 Agent 工具调用的基础
2. ReAct 推理循环
2.1 定义
ReAct（Reason + Act）是 Agent 的经典推理循环：Think（分析当前状态） -> Act（调用工具执行动作） -> Observe（观察执行结果） -> 循环直到任务完成。ReAct 让 Agent 能够将推理与行动交织，逐步逼近目标。
对应 Demo: demos/03_Agent核心能力/02_ReAct推理循环.py
2.2 ReAct 循环架构
+------------------------------------------------------------------+
| ReAct 推理循环 |
+------------------------------------------------------------------+
| |
| 用户目标: "查找 Python 的创建者并计算他的年龄" |
| |
| Iteration 1: |
| Think: 我需要先搜索 Python 的创建者是谁 |
| Act: search_web("Python 创建者") |
| Observe: Python 由 Guido van Rossum 于 1991 年创建 |
| |
| Iteration 2: |
| Think: 现在需要计算 Guido van Rossum 的年龄 |
| Act: calculate("2024 - 1956") |
| Observe: 68 |
| |
| Iteration 3: |
| Think: 已获得所有需要的信息，可以回答了 |
| Act: FINAL_ANSWER("Python 由 Guido van Rossum 创建， |
| 他今年 68 岁") |
| |
+------------------------------------------------------------------+2.3 核心数据结构
class ActionType(Enum):
"""Agent 动作类型"""
CALL_TOOL = "call_tool"
FINAL_ANSWER = "answer"
GIVE_UP = "give_up"
@dataclass
class Action:
"""Agent 决策的动作"""
action_type: ActionType
thought: str # 推理过程
tool_call: ToolCall | None = None
answer: str | None = None
@dataclass
class Step:
"""ReAct 循环的单步记录"""
step_num: int
thought: str # Think
action: Action # Act
observation: str = "" # Observe2.4 ReAct 循环实现
class ReActAgent:
"""ReAct 推理循环 Agent"""
def __init__(self, tools: dict, max_iterations: int = 10):
self.tools = tools
self.max_iterations = max_iterations
self.history: list[Step] = []
def run(self, goal: str) -> str:
"""执行 ReAct 循环直到完成或达到上限"""
for i in range(self.max_iterations):
action = self._think(goal, self.history) # Think
observation = self._act(action) # Act
self.history.append(Step(i + 1, action.thought,
action, observation)) # Observe
if action.action_type == ActionType.FINAL_ANSWER:
return action.answer
return "达到最大迭代次数，未完成任务"2.5 Think 的推理模式
Think 阶段的推理模式:
信息收集: "我需要搜索 X 的信息" -> Act: search(X)
信息整合: "知道了 A 和 B，需要计算 C" -> Act: calculate(A+B)
验证检查: "搜索结果提到 X=5，验证一下" -> Act: search("X准确值")
方案调整: "上一步失败了，换一个方法" -> Act: alternative_tool(...)
任务完成: "已收集所有信息" -> Act: FINAL_ANSWER2.6 ReAct 的优势与局限
2.7 与其他概念的关联
<- Agent 基础概念：ReAct 是 Agent 的运行机制
-> 任务分解与规划：复杂任务需要先分解再 ReAct
-> 停止条件与预算控制：ReAct 需要停止条件防止无限循环
3. 任务分解与规划
3.1 定义
任务分解与规划是 Agent 把复杂目标拆成可执行的子任务列表，逐步执行并动态更新计划的能力。核心是 Planning + Decomposition + Todo List + Dependency Management，让 Agent 能处理需要多步骤、有依赖关系的复杂任务。
对应 Demo: demos/03_Agent核心能力/03_任务分解与规划.py
3.2 任务分解架构
用户目标: "写一篇关于 RAG 技术的技术博客"
Task 1: 研究 RAG 技术背景 [无依赖]
|
v
Task 2: 收集 RAG 相关论文 [依赖 Task 1]
|
v
Task 3: 整理技术要点 [依赖 Task 2]
|
v
Task 4: 撰写博客初稿 [依赖 Task 3]
|
v
Task 5: 审查和修改 [依赖 Task 4]
执行: 串行依赖链 1->2->3->4->5
或: 并行无依赖任务 + 汇聚3.3 任务数据结构
class TaskStatus(Enum):
"""任务状态"""
PENDING = "pending"
IN_PROGRESS = "running"
DONE = "done"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class Task:
"""单个子任务"""
id: str
description: str
status: TaskStatus = TaskStatus.PENDING
result: Any = None
dependencies: list[str] = field(default_factory=list)
retry_count: int = 0
max_retries: int = 2
@dataclass
class Plan:
"""任务计划"""
goal: str
tasks: list[Task] = field(default_factory=list)
def get_ready_tasks(self) -> list[Task]:
"""获取可执行的任务：pending 且所有依赖已完成"""
ready = []
for task in self.tasks:
if task.status != TaskStatus.PENDING:
continue
deps_done = all(
self.get_task(dep) and self.get_task(dep).status == TaskStatus.DONE
for dep in task.dependencies
)
if deps_done:
ready.append(task)
return ready3.4 依赖管理
依赖类型:
串行依赖 (A -> B): B 必须等 A 完成
并行无依赖 (A, B): 可同时执行
部分依赖 (A->C, B->C): C 需等 A 和 B 都完成
条件依赖: B 是否依赖 A 取决于条件
执行调度示例:
Round 1: [A, B] (并行)
Round 2: [C] (等 A, B 完成)
Round 3: [D] (等 C 完成)3.5 动态重规划
def replan_on_failure(plan: Plan, failed_task: Task) -> Plan:
"""任务失败后动态重规划"""
if failed_task.retry_count < failed_task.max_retries:
failed_task.status = TaskStatus.PENDING
failed_task.retry_count += 1
else:
# 添加替代任务
alt_task = Task(id=f"{failed_task.id}_alt",
description=f"替代方案: {failed_task.description}")
plan.tasks.append(alt_task)
return plan3.6 规划策略对比
3.7 与其他概念的关联
<- ReAct 推理循环：每个子任务可用 ReAct 执行
-> 反思与自我纠正：计划失败时触发反思和重规划
-> 停止条件与预算控制：计划需要预算约束
4. 工具选择与调用
4.1 定义
工具选择与调用是 Agent 根据 LLM 输出的函数调用意图，选择正确的工具、校验参数、执行并返回结果的能力。核心是 Tool Schema + Function Calling + Validation + Error Handling。
对应 Demo: demos/03_Agent核心能力/04_工具选择与调用.py
4.2 工具调用完整流程
1. 工具注册: ToolSchema -> ToolRegistry -> OpenAI 格式
2. LLM 决策: 用户意图 + 工具定义 -> LLM -> 选择工具 + 参数
3. 参数校验: 必填检查 -> 类型检查 -> 枚举检查
4. 执行工具: handler() -> 捕获异常 -> 记录耗时
5. 结果处理: 成功->标准化 / 失败->错误信息->重试
6. 返回 LLM: 工具结果作为 ToolMessage 送回 LLM4.3 工具注册表
@dataclass
class ToolSchema:
"""工具 Schema 定义"""
name: str
description: str
parameters: dict[str, dict]
handler: Callable
class ToolRegistry:
"""工具注册表"""
def __init__(self):
self.tools: dict[str, ToolSchema] = {}
def get_openai_format(self) -> list[dict]:
"""生成 OpenAI Function Calling 格式"""
functions = []
for tool in self.tools.values():
properties = {}
required = []
for param_name, param_def in tool.parameters.items():
properties[param_name] = {
"type": param_def.get("type", "string"),
"description": param_def.get("description", ""),
}
if param_def.get("required", False):
required.append(param_name)
functions.append({
"type": "function",
"function": {"name": tool.name,
"description": tool.description,
"parameters": {"type": "object",
"properties": properties,
"required": required}},
})
return functions4.4 工具描述的重要性
工具描述对选择准确率的影响:
差的描述: description: "搜索"
好的描述: description: "搜索互联网获取实时信息。当用户询问最新
新闻、天气、股价等需要实时数据的问题时使用。"
最佳实践:
1. 说明功能: 工具能做什么
2. 说明场景: 什么情况下使用
3. 说明限制: 什么情况下不使用
4. 说明参数: 每个参数的含义和格式4.5 错误处理策略
工具调用错误类型与处理:
参数错误 -> 返回错误给 LLM，让其修正参数重试
执行错误 -> 重试 2-3 次，仍失败则告知 LLM
超时错误 -> 增加超时重试，或使用备用工具
权限错误 -> 不重试，触发 HITL 审批
网络错误 -> 指数退避重试4.6 与其他概念的关联
<- LLM 应用基础 / Function Calling：Agent 工具调用基于 Function Calling
-> ReAct 推理循环：工具调用是 ReAct 的 Act 阶段
-> 人机协同 HITL：高风险工具需要人工审批
5. 反思与自我纠正
5.1 定义
反思与自我纠正是 Agent 评估自身输出质量，发现错误后自动修正的能力。核心是 Reflection + Self-Correction + Retry Loop，让 Agent 能从错误中学习并改进。
对应 Demo: demos/03_Agent核心能力/05_反思与自我纠正.py
5.2 反思循环架构
任务: "编写带类型标注和文档的 Python 函数"
Attempt 1:
Generate: def hello(): print('hello world')
Reflect: 缺少类型标注、缺少文档字符串
Verdict: NEEDS_FIX
Suggest: 添加类型标注和 docstring
Attempt 2:
Generate: def hello(name: str) -> None:
"""打印问候语""" print(f'hello {name}')
Reflect: 类型标注已添加、文档已添加
Verdict: CORRECT
-> 最终输出 Attempt 25.3 核心数据结构
class ReflectionVerdict(Enum):
"""反思评估结论"""
CORRECT = "correct"
NEEDS_FIX = "needs_fix"
GIVE_UP = "give_up"
@dataclass
class ReflectionResult:
"""反思结果"""
verdict: ReflectionVerdict
issues: list[str] = field(default_factory=list)
suggestions: list[str] = field(default_factory=list)
confidence: float = 1.05.4 反思 Agent 实现
class ReflectiveAgent:
"""带反思与自我纠正能力的 Agent"""
def __init__(self, max_attempts: int = 3):
self.max_attempts = max_attempts
self.history: list[AttemptRecord] = []
def run(self, task: str) -> str:
"""执行任务，带反思循环"""
feedback = ""
for attempt in range(self.max_attempts):
output = self._generate(task, feedback)
reflection = self._reflect(task, output)
if reflection.verdict == ReflectionVerdict.CORRECT:
return output
elif reflection.verdict == ReflectionVerdict.GIVE_UP:
return f"无法完成任务: {reflection.issues}"
feedback = "; ".join(reflection.suggestions)
return self.history[-1].output5.5 反思的维度
反思评估维度:
1. 正确性: 输出是否事实正确？代码能运行？
2. 完整性: 是否覆盖所有要求？处理了边界情况？
3. 格式规范: 是否符合要求格式？类型标注完整？
4. 安全性: 是否包含有害内容？违反约束？
5. 效率: 是否有更优方案？存在冗余步骤？5.6 反思 vs 不反思
5.7 与其他概念的关联
<- ReAct 推理循环：反思可嵌入 Observe 阶段
-> 任务分解与规划：计划失败时触发反思重规划
-> 停止条件与预算控制：反思循环需要最大次数限制
6. 短期记忆
6.1 定义
短期记忆管理当前会话的对话上下文。不是简单地把所有消息塞给模型，而是要控制 Token 预算、截断旧消息、摘要压缩，保证关键信息不被淹没。
对应 Demo: demos/03_Agent核心能力/06_短期记忆.py
6.2 短期记忆架构
消息流:
[system] [user1] [ai1] [tool1] [user2] [ai2] [tool2] [user3]
Token 预算: 4096 tokens
管理策略:
Layer 1: 重要性标记 (system=critical, tool1=high, ...)
Layer 2: Token 预算控制 (超预算触发截断)
Layer 3: 截断策略 (优先截断 low 优先级)
Layer 4: 摘要压缩 (旧消息 -> LLM 摘要)
最终上下文:
[system] [summary(旧消息)] [ai2] [tool2] [user3]6.3 短期记忆实现
class ShortTermMemory:
"""短期记忆：Token 预算控制 + 消息截断 + 摘要压缩"""
def __init__(self, max_tokens: int = 4096):
self.max_tokens = max_tokens
self.messages: list[Message] = []
def add(self, role: Role, content: str, importance: str = "normal") -> Message:
"""添加消息，超预算时触发截断"""
msg = Message(role=role, content=content,
token_count=self._count_tokens(content),
importance=importance)
self.messages.append(msg)
self._enforce_budget()
return msg
def _enforce_budget(self) -> None:
"""Token 预算控制：超预算时截断低优先级消息"""
while self._total_tokens() > self.max_tokens and len(self.messages) > 2:
self._truncate_lowest_priority()6.4 记忆管理策略对比
6.5 与其他概念的关联
<- LLM 应用基础 / 上下文窗口管理：短期记忆是窗口管理的 Agent 版
-> 长期记忆：短期记忆溢出的信息可存入长期记忆
-> Context Engineering：短期记忆是上下文工程在 Agent 中的实现
7. 长期记忆
7.1 定义
长期记忆实现跨会话的信息持久化，存储用户偏好、历史事实和经验教训。用向量数据库存储，按语义相似度检索，让 Agent "记住"过去的交互。
对应 Demo: demos/03_Agent核心能力/07_长期记忆.py
7.2 四种记忆类型
1. 用户偏好 (User Profile):
"用户偏好简洁回答" "用户是 Python 开发者"
-> 长期稳定，用于个性化
2. 事实信息 (Fact):
"用户的公司叫 ACME Corp" "项目使用 React+TS"
-> 客观事实，跨会话有效
3. 情景记忆 (Episodic):
"上次用户问过 RAG 方案，推荐了 LangChain"
-> 特定事件，有时效性
4. 过程记忆 (Procedural):
"部署流程: 测试->构建->推送->部署"
-> 如何做某事，可复用7.3 记忆存储与检索
class MemoryStore:
"""长期记忆存储：基于向量数据库"""
def store(self, content: str, memory_type: MemoryType,
user_id: str = "default") -> str:
"""存储一条长期记忆"""
vector = mock_embedding(content, self.vector_dim)
self.client.upsert(collection_name=self.collection,
points=[PointStruct(
id=self._counter,
vector=vector,
payload={"content": content,
"memory_type": memory_type.value,
"user_id": user_id})])
def retrieve(self, query: str, user_id: str = "default",
top_k: int = 5) -> list[dict]:
"""按语义相似度检索记忆"""
query_vec = mock_embedding(query, self.vector_dim)
results = self.client.search(
collection_name=self.collection,
query_vector=query_vec,
query_filter=Filter(must=[FieldCondition(
key="user_id", match=MatchValue(value=user_id))]),
limit=top_k)
return results7.4 遗忘机制
遗忘机制 (基于 Ebbinghaus 遗忘曲线):
记忆保留分 = importance x recency x frequency
importance: 用户标注的重要性 (0~1)
recency: 距上次访问的时间 (越久越低)
frequency: 访问次数 (越多越高)
遗忘策略:
1. 定期清理: 删除保留分低于阈值的记忆
2. 容量限制: 超容量时删除最低分记忆
3. 冲突合并: 新旧冲突时以新记忆为准
4. 重要性衰减: 长期未访问的记忆重要性递减7.5 短期记忆 vs 长期记忆
7.6 与其他概念的关联
<- 短期记忆：短期记忆溢出的重要信息存入长期记忆
-> 完整 Agent：长期记忆是完整 Agent 的跨会话能力
-> RAG：长期记忆检索与 RAG 检索原理相同
8. 人机协同 HITL
8.1 定义
人机协同（Human-In-The-Loop）是 Agent 在关键决策点暂停，等待人工审批、修改或拒绝的能力。核心是 Interrupt + Approval + Resume + Reject，确保 Agent 在高风险操作前有人工把关。
对应 Demo: demos/03_Agent核心能力/08_人机协同HITL.py
8.2 HITL 架构
Agent 执行中
|
v
LLM 决定调用高风险工具: send_email(to="all@company.com")
|
v
风险评估: risk_level = "high"
|
v
中断执行 -> Agent 暂停，等待审批
|
v
审批请求 (工具名/参数/原因/风险)
|
v
人工决策:
+-- APPROVED -> 执行原参数 -> Agent 恢复
+-- MODIFIED -> 执行修改后参数 -> Agent 恢复
+-- REJECTED -> 跳过此工具 -> Agent 恢复
+-- DEFERRED -> 暂缓，需要更多信息8.3 审批数据结构
class ApprovalDecision(Enum):
"""审批决策"""
APPROVED = "approved"
MODIFIED = "modified"
REJECTED = "rejected"
DEFERRED = "deferred"
@dataclass
class ApprovalRequest:
"""审批请求"""
request_id: str
tool_name: str
arguments: dict
reason: str
risk_level: str = "medium" # low/medium/high/critical8.4 风险分级与审批策略
+--------+-----------+-----------------------------------+
| 风险等级| 审批策略 | 示例 |
+--------+-----------+-----------------------------------+
| low | 自动执行 | search(), calculate() |
| medium | 可选审批 | database_query(只读) |
| high | 必须审批 | send_email(), update_record() |
|critical| 审批+确认 | delete_file(), deploy_code() |
+--------+-----------+-----------------------------------+8.5 与其他概念的关联
<- 工具选择与调用：HITL 在工具执行前拦截
-> 完整 Agent：HITL 是生产级 Agent 的必备能力
-> LangGraph / HITL：LangGraph 用 interrupt_before 实现 HITL
9. 停止条件与预算控制
9.1 定义
停止条件与预算控制确保 Agent 不会无限循环。需要多重停止条件：最大轮数、Token 预算、费用限制、超时、重复检测和收敛判断。这是 Agent 从"实验"走向"生产"的关键安全保障。
对应 Demo: demos/03_Agent核心能力/09_停止条件与预算控制.py
9.2 五重停止条件
1. 最大轮数 (Max Turns): Agent 最多执行 N 轮推理循环
2. Token 预算 (Token Budget): 累计 Token 消耗超过预算
3. 费用预算 (Cost Budget): 累计费用超过美元预算
4. 超时 (Timeout): 执行时间超过设定阈值
5. 重复检测 (Repetition): 连续 N 次输出相同结果
额外: 收敛判断 (Convergence): 输出不再变化，认为已收敛9.3 预算配置与控制
class StopReason(Enum):
"""Agent 停止原因"""
COMPLETED = "completed"
MAX_TURNS = "max_turns"
TOKEN_BUDGET = "token_budget"
COST_BUDGET = "cost_budget"
TIMEOUT = "timeout"
REPETITION = "repetition"
CONVERGENCE = "convergence"
ERROR = "error"
USER_STOP = "user_stop"
@dataclass
class BudgetConfig:
"""预算配置"""
max_turns: int = 20
max_tokens: int = 50000
max_cost_usd: float = 1.0
timeout_seconds: float = 120.0
max_repetitions: int = 3
token_price_per_1k: float = 0.002
class BudgetController:
"""预算控制器：监控资源消耗，超限时触发停止"""
def check(self, output: str = "") -> StopReason | None:
"""检查是否触发任何停止条件"""
if self.status.turns_used >= self.config.max_turns:
return StopReason.MAX_TURNS
if self.status.tokens_used >= self.config.max_tokens:
return StopReason.TOKEN_BUDGET
if self.status.cost_used >= self.config.max_cost_usd:
return StopReason.COST_BUDGET
if self.status.elapsed_seconds >= self.config.timeout_seconds:
return StopReason.TIMEOUT
if output and output == self.status.last_output:
self.status.repetition_count += 1
if self.status.repetition_count >= self.config.max_repetitions:
return StopReason.REPETITION
return None9.4 预算可视化
预算消耗监控:
轮数: [########## ] 10/20 50%
Token: [##### ] 25000/50K 50%
费用: [## ] $0.05/$1 5%
时间: [#### ] 48s/120s 40%
停止优先级:
1. 任务完成 (COMPLETED) -> 正常结束
2. 不可恢复错误 (ERROR) -> 异常结束
3. 用户停止 (USER_STOP) -> 外部中断
4. 预算耗尽 -> 资源限制
5. 重复检测 -> 死循环保护9.5 与其他概念的关联
<- ReAct 推理循环：每轮 ReAct 后检查停止条件
<- 反思与自我纠正：反思循环也需要最大次数限制
-> 完整 Agent：预算控制是完整 Agent 的安全保障
10. 完整 Agent
10.1 定义
完整 Agent 集成 ReAct 推理 + 工具调用 + 短期记忆 + 长期记忆 + 反思纠正 + HITL 审批 + 预算控制，构成一个可运行的 Agent 系统。这是前面所有能力的集大成者，展示各核心能力如何协同工作。
对应 Demo: demos/03_Agent核心能力/10_完整Agent.py
10.2 完整 Agent 架构
+------------------------------------------------------------------+
| 完整 Agent 系统架构 |
+------------------------------------------------------------------+
| |
| +----------------+ +----------------+ +----------------+ |
| | 指令层级 | | 状态管理 | | 预算控制器 | |
| | System > Task | | AgentState | | 5重停止条件 | |
| +-------+--------+ +-------+--------+ +-------+--------+ |
| | | | |
| +-------+-----------+-------+-----------+ |
| v v |
| +---------------------------------------------------+ |
| | ReAct 推理循环 | |
| | Think(LLM) -> Act(Tool+HITL) -> Observe(Reflect) | |
| | ^ | | |
| | +------------------------------------+ | |
| | (循环) | |
| +---------------------------------------------------+ |
| |
| +----------------+ +----------------+ |
| | 短期记忆 | | 长期记忆 | |
| | Token预算/截断 | | 向量存储/4类型 | |
| | 摘要压缩 | | 遗忘机制 | |
| +----------------+ +----------------+ |
| |
+------------------------------------------------------------------+10.3 Agent 主循环
class CompleteAgent:
"""集成所有核心能力的完整 Agent"""
def __init__(self, config: BudgetConfig):
self.tools = {}
self.short_memory = ShortTermMemory(max_tokens=4096)
self.long_memory = MemoryStore()
self.budget = BudgetController(config)
self.hitl = HumanInTheLoop()
self.state = AgentState()
def run(self, user_input: str) -> dict:
"""Agent 主循环"""
self.state.goal = user_input
while True:
# 1. 预算检查
stop = self.budget.check()
if stop:
return {"answer": f"Agent 停止: {stop.value}"}
# 2. Think: LLM 推理
action = self._llm_think(self._build_messages())
# 3. 判断是否完成
if action.action_type == ActionType.FINAL_ANSWER:
return {"answer": action.answer}
# 4. HITL 审批 (高风险工具)
if self._needs_approval(action):
approval = self.hitl.request_approval(...)
# 5. Act: 执行工具
result = self._execute_tool(action)
# 6. Observe: 记录结果
self.short_memory.add(Role.TOOL, result)
# 7. 反思检查
if self._should_reflect():
reflection = self._reflect(result)
# 8. 记录预算消耗
self.budget.consume(input_tokens=500, output_tokens=200)10.4 各能力协同关系
关键协同路径:
1. ReAct -> Tool -> HITL -> ShortMem -> ReAct (工具调用循环)
2. ReAct -> Budget -> Stop (预算控制循环)
3. ReAct -> Reflection -> Plan -> ReAct (反思重规划循环)
4. ShortMem -> LongMem -> ShortMem (记忆流转循环)10.5 从 Demo 到生产
Demo Agent 生产级 Agent
- 模拟 LLM 调用 - 真实 LLM API + 重试
- 内存存储 - 持久化 + 数据库
- 单用户 - 多用户隔离
- 无监控 - 全链路追踪 + 告警
- 无错误恢复 - 状态机 + Checkpoint
- 固定工具集 - 动态注册 + MCP
- 无安全审计 - 审计日志 + 安全扫描10.6 与其他概念的关联
<- 所有前置能力：完整 Agent 集成所有核心能力
-> LangGraph 实战：LangGraph 是实现完整 Agent 的框架
-> 工具集成与 MCP：生产级 Agent 需要更丰富的工具生态
概念关系总览
+--------------------------------------------------+
| Agent 核心能力体系 |
+--------------------------------------------------+
基础层 推理层 记忆层 控制层
| | | |
v v v v
+-----------+ +---------------+ +---------------+ +-----------+
|Agent 基础 | |ReAct 推理循环 | |短期记忆 | |HITL 人机 |
|概念 |--->|Think-Act- | |Token预算/截断 | |协同 |
|(角色/状态/ | |Observe | |摘要压缩 | |审批/拒绝 |
| 指令层级) | +-------+-------+ +-------+-------+ +-----+-----+
+-----+-----+ | | | |
| v v | |
| +---------------+ +---------------+ | |
| |任务分解与规划 | |长期记忆 | | |
| |Planning/依赖 | |4种记忆类型 | | |
| |重试 | |遗忘机制 | | |
| +-------+-------+ +-------+-------+ | |
| | | | |
| v | | |
| +---------------+ | | |
| |工具选择与调用 | | | |
| |Schema/校验 |<-----------+ | |
| |错误处理 | | |
| +-------+-------+ | |
| | | |
| v | |
| +---------------+ | |
| |反思与自我纠正 | | |
| |评估/检测/纠正 | | |
| +-------+-------+ | |
| | | |
+-----------------+--------------------------------------+ |
| |
v |
+---------------+ +----------------------------------+
|停止条件与 | |预算控制器 |
|预算控制 | |5重停止条件 |
|(MaxTurns/ | |(轮数/Token/费用/超时/重复) |
| Token/超时) | +----------------------------------+
+-------+-------+
|
v
+---------------+
|完整 Agent |
|(集成所有能力) |
+---------------+概念间的依赖关系
Agent基础 -> ReAct循环 -> 工具调用 -> 反思纠正
| | |
v v v
任务规划 短期记忆 停止条件
| | |
| v |
| 长期记忆 |
| | |
+------+-------+-----------+
|
v
完整 Agent
评论区