自主智能体

自主代理 (Autonomous Agents)

自主代理是指能够独立分解目标、规划行动、执行工具并在无需人类持续引导的情况下进行自我修正的 AI 系统。挑战不在于提升其能力，而在于提升其可靠性。每一个额外的决策都会增加失败的概率。

本技能涵盖代理循环（ReAct, Plan-Execute）、目标分解、反思模式以及生产环境的可靠性。核心洞察：复合错误率是自主代理的杀手。每步 95% 的成功率在第 10 步时会下降到 60%。应优先构建可靠性，其次才是自主性。

2025 年的教训：赢家是那些具有明确边界、受限且领域特定的代理，而非“全能自主”的代理。将 AI 的输出视为“提案”而非“真理”。

原则

• 可靠性高于自主性 —— 每一步都会复合错误概率

• 限制范围 —— 领域特定优于通用目的

• 将输出视为提案，而非真理

• 在扩展能力之前先构建护栏（Guardrails）

• 关键决策必须引入人机协同（Human-in-the-loop）

• 全量日志记录 —— 每项操作必须可审计

• 安全失败（支持回滚），而非静默失败（导致数据损坏）

能力

• autonomous-agents

• agent-loops

• goal-decomposition

• self-correction

• reflection-patterns

• react-pattern

• plan-execute

• agent-reliability

• agent-guardrails

范围

• multi-agent-systems → multi-agent-orchestration

• tool-building → agent-tool-builder

• memory-systems → agent-memory-systems

• workflow-orchestration → workflow-automation

工具链

框架

• LangGraph - 适用场景：需要状态管理的生产级代理。注：1.0 版本于 2025 年 10 月发布，支持检查点（checkpointing）和人机协同。

• AutoGPT - 适用场景：研究/实验，开放式探索。注：生产环境需要外部护栏。

• CrewAI - 适用场景：基于角色的代理团队。注：适用于专业代理协作。

• Claude Agent SDK - 适用场景：Anthropic 生态系统代理。注：支持计算机使用（Computer use）和工具执行。

模式

• ReAct - 适用场景：推理与行动交替进行。注：大多数现代代理的基础。

• Plan-Execute - 适用场景：将规划与执行分离。注：更适用于复杂的多步任务。

• Reflection - 适用场景：自我评估与修正。注：评估者-优化者循环。

模式

ReAct 代理循环

推理与行动步骤交替进行。

适用场景：交互式问题解决、工具使用、探索。

REACT 模式：

"""
ReAct 循环：
1. Thought（思考）：推理下一步该做什么
2. Action（行动）：选择并执行一个工具
3. Observation（观察）：接收结果
4. 重复直到达成目标

关键点：显式的推理轨迹使调试成为可能
"""

基础 ReAct 实现

定义 ReAct 提示词模板

Question: 输入的问题
Thought: 对如何操作进行推理
Action: tool_name
Action Input: 工具的输入
Observation: 行动的结果
... (根据需要重复 Thought/Action/Observation)
Thought: 我现在知道了最终答案
Final Answer:
"""
答案
'''

创建 agent

设置步骤限制执行

LangGraph ReAct (生产环境)

生产环境 checkpointer

agent = create_react_agent(
model=llm,
tools=tools,
checkpointer=checkpointer, # 持久化状态
)

使用 thread 以实现状态持久化

Plan-Execute 模式（计划-执行）

将规划阶段与执行阶段分离

适用场景：复杂的多步骤任务，且需要完整可见计划的情况

PLAN-EXECUTE 模式：

"""
两阶段方法：
1. 规划 (Planning)：将目标分解为子任务
2. 执行 (Execution)：执行子任务，必要时重新规划

优点：

• 执行前对计划有完全的可见性

• 可以由人工验证/修改计划

• 关注点分离更清晰

缺点：

• 对任务执行过程中的新发现适应性较差

• 计划可能会过时

LangGraph Plan-Execute

Planner 创建任务列表

Executor 处理单个步骤

agent = create_plan_and_execute_agent(
planner=planner_llm,
executor=executor_llm,
tools=tools,
replan_on_error=True, # 步骤失败时重新规划
)

人工审核计划

第一次调用创建计划

审核计划，然后继续

分解策略

先分解后执行 (Decomposition-First)：规划所有步骤，然后统一执行

最佳适用：稳定任务，需要完整计划审批

交替执行 (Interleaved)：规划一步，执行一步，循环往复

最佳适用：动态任务，在执行过程中学习/调整

def interleaved_execute(goal, max_steps=10):
state = {"goal": goal, "completed": [], "remaining": [goal]}

for step in range(max_steps):
# 根据当前状态规划下一步行动
next_action = planner.plan_next(state)

if next_action == "DONE":
break

# 执行并更新状态
result = executor.execute(next_action)
state["completed"].append((next_action, result))

# 重新评估剩余工作
state["remaining"] = planner.reassess(state)

return state
"""

Reflection 模式（反思）

自我评估与迭代改进

适用场景：对质量要求高、输出复杂或创意类任务

REFLECTION 模式：

"""
自我修正循环：
1. 生成初始输出
2. 根据标准进行评估
3. 批判并识别问题
4. 根据批判结果进行优化
5. 重复直至满意

也称为：评估者-优化者 (Evaluator-Optimizer)、自我批判 (Self-Critique)
"""

基础反思 (Basic Reflec)

for i in range(max_iterations):
# 评估输出
critique = evaluator.critique(
task=task,
output=output,
criteria=[
"正确性",
"完整性",
"清晰度",
]
)

if critique["passes_all"]:
return output

# 根据评估结果进行优化
output = generator.refine(
task=task,
previous_output=output,
critique=critique["feedback"],
)

return output # 达到最大迭代次数后的最佳尝试
"""

LangGraph 反思机制

def build_reflection_graph():
graph = StateGraph(ReflectionState)

# 节点
graph.add_node("generate", generate_node)
graph.add_node("reflect", reflect_node)
graph.add_node("output", output_node)

# 边
graph.add_edge("generate", "reflect")
graph.add_conditional_edges(
"reflect",
should_continue,
{
"continue": "generate", # 循环回溯
"end": "output",
}
)

return graph.compile()

def should_continue(state):
if state["iteration"] >= 3:
return "end"
if state["score"] >= 0.9:
return "end"
return "continue"
"""

独立评估器（更鲁棒）

使用不同的模型进行评估以避免自我偏差

或使用专门的评估器

带护栏的自主性 (Guardrailed Autonomy)

具有安全边界的受限智能体

适用场景：生产系统、关键业务操作

带护栏的自主性：

"""
生产环境中的智能体需要多层安全保障：
1. 输入验证
2. 动作约束
3. 输出验证
4. 成本限制
5. 人工干预/升级
6. 回滚能力
"""

多层护栏

async def execute(self, goal):
total_cost = 0
steps = 0

while steps < self.max_steps:
# 获取下一步动作
action = await self.agent.plan_next(goal)

# 验证动作是否被允许
if action.name not in self.allowed_actions:
raise ActionNotAllowedError(action.name)

# 检查是否需要审批
if action.name in self.require_approval:
approved = await self.request_human_approval(action)
if not approved:
return {"status": "rejected", "action": action}

# 预估成本
estimated_cost = self.estimate_cost(action)
if total_cost + estimated_cost > self.max_cost:
raise CostLimitExceededError(total_cost)

# 执行并具备回滚能力
checkpoint = await self.save_checkpoint()
try:
result = await self.agent.execute(action)
total_cost += self.actual_
"""
cost(action)
steps += 1
except Exception as e:
await self.rollback_to(checkpoint)
raise

if result.is_complete:
break

return {"status": "complete", "total_cost": total_cost}
"""

最小权限原则

为每种任务类型定义最小权限

def create_scoped_agent(task_type):
allowed = TASK_PERMISSIONS.get(task_type, [])
tools = [t for t in ALL_TOOLS if t.name in allowed]
return Agent(tools=tools)
"""

成本控制

上下文长度的成本呈平方级增长

上下文翻倍 = 成本 4 倍

def trim_context(messages, max_tokens=4000):
# 保留系统消息和最近的消息
system = messages[0]
recent = messages[-10:]

# 如有必要，对中间部分进行总结
if len(messages) > 11:
middle = messages[1:-10]
summary = summarize(middle)
return [system, summary] + recent

return messages
"""

持久化执行模式 (Durable Execution Pattern)

能够从失败中恢复并继续执行的 Agent

适用场景：长时间运行的任务、生产系统、跨多日流程

持久化执行：

"""
生产级 Agent 必须：

• 在服务器重启后能生存

• 从精确的失败点恢复

• 处理运行时间长达数小时/数天的任务

• 允许在流程中间进行人工干预

LangGraph 1.0 原生提供了此功能。
"""

LangGraph 检查点 (Checkpointing)

生产级检查点保存器（而非 MemorySaver！）

构建带有检查点的图

... 添加节点和边 ...

agent = graph.compile(checkpointer=checkpointer)

每次调用都会保存状态

开始任务

如果服务器崩溃，稍后恢复：

人机协作中断 (Human-in-the-Loop Interrupts)

在特定节点暂停

首次调用在中断点暂停

人员审核状态

时光回溯调试 (Time-Travel Debugging)

LangGraph 存储完整历史记录

回到之前的任何状态

从该点开始，在修改后重新执行

潜在陷阱 (Sharp Edges)

错误概率呈指数级复合

严重程度：极高 (CRITICAL)

场景：构建多步骤自主 Agent

症状：
Agent 在 Demo 中运行良好，但在生产环境中失败。简单任务能成功，
复杂任务莫名失败。随着任务复杂度增加，成功率剧烈下降。用户失去信任。

失效原因：
每一步都有独立的失败概率。如果每步成功率为 95%...
分步执行听起来很棒，直到你意识到：

• 5 个步骤：77% 成功率 (0.95^5)

• 10 个步骤：60% 成功率 (0.95^10)

• 20 个步骤：36% 成功率 (0.95^20)

这是自主代理（autonomous agents）的根本限制。每增加一个步骤，失败概率就会乘积增长。

建议修复方案：

减少步骤数量

尽可能合并步骤

倾向于使用少量高能力步骤，而非大量微小步骤

提高单步可靠性

使用结构化输出 (JSON schemas)

在每一步增加验证机制

为关键步骤使用更强大的模型

为失败而设计

拆分为带检查点的分段

每个分段进行人工审核

从最后一个正确的检查点恢复

上下文增长导致 API 成本爆炸

严重程度：极高 (CRITICAL)

场景：运行具有增长对话上下文的代理

症状：
关闭单个支持工单需花费 47 美元。收到数千美元的意外 API 账单。
代理运行时间越长，速度越慢。Token 数量超过模型限制。

失效原因：
Transformer 的成本随上下文长度呈平方级增长。上下文翻倍，计算量增加四倍。一个每轮都重新发送完整对话的长运行代理，其资金消耗呈指数级增长。

大多数代理在追加上下文时没有进行裁剪。上下文持续增长：

• 第 1 轮：500 tokens → $0.01

• 第 10 轮：5000 tokens → $0.10

• 第 50 轮：25000 tokens → $0.50

• 第 100 轮：50000 tokens → 每条消息 $1.00+

建议修复方案：

设置硬性成本限制

def __init__(self):
self.total_cost = 0

def before_call(self, estimated_tokens):
estimated_cost = self.estimate_cost(estimated_tokens)
if self.total_cost + estimated_cost > self.MAX_COST_PER_TASK:
raise CostLimitExceeded(
f"Would exceed ${self.MAX_COST_PER_TASK} limit"
)

def after_call(self, response):
self.total_cost += self.calculate_actual_cost(response)

激进地裁剪上下文

system = messages[0]
recent = messages[-5:]
middle = messages[1:-5]

if middle:
summary = summarize(middle) # 压缩历史记录
return [system, summary] + recent

return [system] + recent

使用流式传输实时跟踪成本

预算消耗 50% 时预警，90% 时停止

Demo 成功但生产环境失败

严重程度：极高 (CRITICAL)

场景：从原型阶段迁移到生产环境

症状：
向利益相关者演示时效果惊人。但在生产环境中持续数月失败。
对创始人的使用场景有效，对真实用户失效。边缘情况（edge cases）压垮了系统。

失效原因：
Demo 展示的是经过精心挑选输入后的“理想路径”。而生产环境意味着：

• 不可预见的输入（拼写错误、歧义、对抗性输入）

• 规模化（1000 个用户，而非 3 个）

• 可靠性（99.9% 的可用性，而非“通常有效”）

• 边缘情况（那 1% 导致整体崩溃的异常）

虽然这种分析方法有待商榷，但核心问题是真实的。一个可运行的 Demo 与一个可靠的生产系统之间的差距是巨大的。
这是项目失败的地方。

建议修复方案：

在生产前进行大规模测试

运行 1000+ 个测试用例，而非 10 个

衡量 P95/P99 成功率，而非平均值

包含对抗性输入

优先构建可观测性

class ObservableAgent:
def execute(self, task):
with logger.bind(task_id=task.id):
logger.info("task_started")
try:
result = self._execute(task)
logger.info("task_completed", result=result)
return result
except Exception as e:
logger.error("task_failed", error=str(e))
raise

设置逃生机制

当置信度 < 阈值时，由人工接管

优雅降级至更简单的行为

“我不知道”是一个有效的回答

渐进式部署

先 1% 的流量，然后 10%，最后 50%

在每个阶段监控错误率

Agent 在卡住时伪造数据

严重程度：高

场景：Agent 无法利用现有信息完成任务

症状：
Agent 编造看起来很真实的数据。例如：报销单上出现虚构的餐厅名称；报告中出现捏造的统计数据；给出自信但完全错误的答案。

失效原因：
LLM 的训练目标是提供帮助并产生看似合理的输出。当卡住时，它们不会说“我无法完成”，而是会进行伪造。自主 Agent 由于在没有人工审核的情况下基于伪造数据采取行动，从而放大了这一问题。

那个伪造报销条目的 Agent 是为了达成目标（完成报销报告），它通过编造数据“解决”了问题。

建议修复方案：

基于事实真相（Ground Truth）进行验证

# 检查可疑模式
if expense.amount == round(expense.amount, -1):
flag_for_review("Suspiciously round amount")

要求提供证据

使用结构化输出

class VerifiedClaim(BaseModel):
claim: str
source: str # 必须引用工具输出
confidence: float

检测不确定性

训练其输出置信度分数

将低置信度的输出标记为人工审核

绝不要在不确定的数据上自动执行

集成是 Agent 的死穴

严重程度：高

场景：将 Agent 连接到外部系统

症状：
使用 Mock API 时正常，使用真实 API 时失败。频率限制（Rate limits）导致崩溃。任务执行中途 Auth Token 过期。数据格式不匹配。部分失败导致系统处于不一致状态。

失效原因：
那些承诺“能与你整个技术栈集成的自主 Agent”的公司，并没有构建过大规模的生产系统。真实的集成面临着：

• 频率限制（任务中途出现 429 错误）

• 认证复杂性（OAuth 刷新、Token 过期）

• 数据格式差异（API v1 与 v2 的区别）

• 部分失败（收到 Webhook 但处理失败）

• 最终一致性（数据无法立即获取）

建议修复方案：

构建健壮的 API 客户端

class RobustAPIClient:
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=60)
)
async def call(self, endpoint, data):

response = await self.client.post(endpoint, json=data)
        if response.status_code == 429:
            retry_after = response.headers.get("Retry-After", 60)
            await asyncio.sleep(int(retry_after))
            raise RateLimitError()
        return response

处理认证生命周期

class TokenManager:
    def __init__(self):
        self.token = None
        self.expires_at = None
async def get_token(self):
        if self.is_expired():
            self.token = await self.refresh_token()
        return self.token
def is_expired(self):
        buffer = timedelta(minutes=5)  # 提前刷新
        return datetime.now() > (self.expires_at - buffer)

使用幂等键

每个外部操作都应该是幂等的

如果 Agent 重试，外部系统应能处理重复请求

为部分失败而设计

每个步骤都应能独立恢复

在调用外部接口前建立检查点 (Checkpoint)

为每个集成提供回滚能力

Agent 执行危险操作

严重程度：高 (HIGH)

场景：Agent 拥有过宽的权限

症状：
Agent 删除生产数据；向错误的接收者发送邮件；
在未经批准的情况下进行购买；修改了不该触碰的设置；
执行了不可逆的操作。

失效原因：
Agent 会为了达成目标而进行优化。如果没有护栏，它们会选择最短路径——即使该路径具有破坏性。例如，被告知“清理数据库”的 Agent 可能会将其理解为“删除所有内容”。

宽泛权限 + 自主权 + 目标优化 = 危险。

推荐修复方案：

最小权限原则

PERMISSIONS = {
    "research_agent": ["read_web", "read_docs"],
    "code_agent": ["read_file", "write_file", "run_tests"],
    "email_agent": ["read_email", "draft_email"],  # 仅限草拟，不可发送
    "admin_agent": ["all"],  # 极少使用
}

读写权限分离

Agent 可以读取任何内容

写入操作需要明确批准

危险操作需要确认

DANGEROUS_ACTIONS = [
    "delete_*",
    "send_email",
    "transfer_money",
    "modify_production",
    "revoke_access",
]
async def execute_action(action):
    if matches_dangerous_pattern(action):
        approval = await request_human_approval(action)
        if not approval:
            return ActionRejected(action)
    return await actually_execute(action)

测试时的干跑 (Dry-run) 模式

Agent 描述它将要执行的操作

人类批准该计划

然后 Agent 再执行

全量审计日志

记录每个操作及其上下文

谁授权的

更改了什么

如何撤销

Agent 耗尽上下文窗口

严重程度：中 (MEDIUM)

场景：长时间运行的 Agent 任务

症状：
Agent 忘记之前的指令；前后矛盾；丢失目标；开始重复自己的话；模型报 token 限制错误。

失效原因：
每条消息、观察结果和思考过程都会消耗上下文。长任务会耗尽窗口。当上下文被截断时：

• 系统提示词 (System Prompt) 可能会丢失

• 早期重要的上下文丢失

• Agent 失去连贯性

推荐修复方案：

跟踪上下文使用情况

class ContextManager:
    def __init__(self, max_tokens=100000):
        self.max_tokens = max_tokens
        self.messages = []
def add(self, message):
        self.messages.append(message)
        self.maybe_compact()
def maybe_compact(self):
        if self.token_count() > self.max_tokens * 0.8:
            self.compact()
def compact(self):
        # 始终保留：系统提示词
        system = self.messages[0]
# 始终保留：最后 N 条消息

recent = self.messages[-10:]

# 总结：其余所有内容
        middle = self.messages[1:-10]
        if middle:
            summary = summarize_messages(middle)
            self.messages = [system, summary] + recent

使用外部记忆

不要将所有内容都保留在上下文中

存储在向量数据库中，在需要时检索

参考 agent-memory-systems 技能

分层总结

最近：完整细节

中期：关键点

远期：压缩总结

看不见就无法调试

严重程度：中 (MEDIUM)

场景：Agent 神秘失败

症状：
“它就是没起作用。” 不知道 Agent 为什么失败。无法复现问题。用户报告的问题无法解释。调试全靠猜测。

失效原因：
Agent 会做出数十个内部决策。如果无法看到每一步的操作，你将无法感知失败模式。没有追踪记录（traces），生产环境的调试几乎是不可能的。

建议修复方案：

结构化日志

import structlog
logger = structlog.get_logger()
class TracedAgent:
    def think(self, context):
        with logger.bind(step="think"):
            thought = self.llm.generate(context)
            logger.info("thought_generated",
                thought=thought,
                tokens=count_tokens(thought)
            )
            return thought
def act(self, action):
        with logger.bind(step="act", action=action.name):
            logger.info("action_started")
            try:
                result = action.execute()
                logger.info("action_completed", result=result)
                return result
            except Exception as e:
                logger.error("action_failed", error=str(e))
                raise

使用 LangSmith 或类似工具

from langsmith import trace
@trace
def agent_step(state):
    # 自动追踪输入/输出
    return next_state

保存完整追踪记录

每一步，每个决策

输入和输出

每一步的延迟

Token 使用量

验证检查

Agent 循环缺乏步数限制

严重程度：高 (ERROR)

自主 Agent 必须设有最大步数限制。

提示信息：Agent 循环缺乏步数限制。请添加 max_steps 以防止死循环。

缺乏成本追踪或限制

严重程度：高 (ERROR)

Agent 应当追踪并限制 API 成本。

提示信息：Agent 使用 LLM 时缺乏成本追踪。请添加成本限制以防止费用失控。

Agent 缺乏超时机制

严重程度：警告 (WARNING)

长时间运行的 Agent 需要超时设置。

提示信息：Agent 调用缺乏超时机制。请添加 timeout 以防止任务挂起。

在生产环境中使用 MemorySaver

严重程度：高 (ERROR)

MemorySaver 仅用于开发环境。

提示信息：MemorySaver 不是持久化的。生产环境请使用 PostgresSaver 或 SqliteSaver。

长时间运行的 Agent 缺乏检查点 (Checkpointing)

严重程度：警告 (WARNING)

运行多个步骤的 Agent 需要检查点机制。

提示信息：多步 Agent 缺乏检查点。请添加 checkpointer 以保证持久性。

Agent 缺乏线程 ID (Thread ID)

严重程度：警告 (WARNING)

带有检查点的 Agent 需要唯一的线程 ID。

提示信息：Agent 调用缺乏 thread_id。状态将无法正确持久化。

直接使用 Agent 输出而未进行验证

严重程度：警告 (WARNING)

Agent 的输出在被使用前应经过验证。

提示信息：Agent 输出在未验证的情况下被使用。请在执行结果前进行验证。

Agent 缺乏结构化输出

严重程度：信息 (INFO)

结构化输出更加可靠。

提示信息：考虑使用结构化输出（如 Pydantic）以获得更可靠的解析。

Agent 缺乏错误恢复机制

严重程度：警告 (WARNING)

Agent 应当能够处理并从错误中恢复。
消息：Agent 调用缺乏错误处理。请添加 try/catch 或错误处理器。

缺乏回滚的破坏性操作

严重程度：WARNING

修改状态的操作应该是可逆的

消息：破坏性操作缺乏回滚能力。请在修改前保存状态。

协作

委派触发条件

• 用户需要多 Agent 协同 $\rightarrow$ multi-agent-orchestration（多个 Agent 协同工作）

• 用户需要测试/评估 Agent $\rightarrow$ agent-evaluation（基准测试与评估）

• 用户需要为 Agent 提供工具 $\rightarrow$ agent-tool-builder（工具设计与实现）

• 用户需要持久化记忆 $\rightarrow$ agent-memory-systems（长期记忆架构）

• 用户需要工作流自动化 $\rightarrow$ workflow-automation（当 Agent 对该任务而言过于冗余时）

• 用户需要计算机控制 $\rightarrow$ computer-use-agents（GUI 自动化、屏幕交互）

相关技能

协同效果佳：agent-tool-builder, agent-memory-systems, multi-agent-orchestration, agent-evaluation

使用场景

• 用户提到或暗示：自主 Agent (autonomous agent)

• 用户提到或暗示：autogpt

• 用户提到或暗示：babyagi

• 用户提到或暗示：自我提示 (self-prompting)

• 用户提到或暗示：目标分解 (goal decomposition)

• 用户提到或暗示：ReAct 模式 (react pattern)

• 用户提到或暗示：Agent 循环 (agent loop)

• 用户提到或暗示：自纠错 Agent (self-correcting agent)

• 用户提到或暗示：反思 Agent (reflection agent)

• 用户提到或暗示：langgraph

• 用户提到或暗示：Agentic AI

• 用户提到或暗示：Agent 规划 (agent planning)

局限性

• 仅在任务明确符合上述范围时使用此技能。

• 不要将输出视为针对特定环境的验证、测试或专家评审的替代方案。

• 如果缺失必要的输入、权限、安全边界或成功标准，请停止并请求澄清。