上周调试一个智能家居控制Agent，遇到一个典型问题：用户说“把客厅温度调到24度然后打开扫地机器人”，结果Agent直接调用“调温+清扫”的复合接口，忽略了扫地机器人需要先检查地面是否有障碍物的前置条件。这种场景让我重新思考：Agent面对复杂指令时，到底该直接执行，还是先拆解再规划？

规划不是简单拆句子

很多初级实现会把“先A后B”的指令切成两个独立任务，但真实场景的依赖关系要复杂得多。比如“下载最新日志并分析错误趋势”：

• 表面看是两个动作
• 实际隐含依赖：下载完成才能分析
• 还可能隐藏条件：需要确保磁盘空间>1GB才下载

早期版本我写过这样的代码：

def handle_command(command):
    tasks = command.split('然后')  # 字符串切分太脆弱了
    for task in tasks:
        execute(task)  # 没有依赖检查

结果经常因为任务顺序错乱导致执行失败。这种字符串切割的方式对自然语言的理解太肤浅了。

规划的本质是状态空间搜索

后来在嵌入式调度系统中找到了灵感：复杂任务规划其实是搜索从初始状态到目标状态的动作序列。以“准备会议”为例：

目标状态：会议室已预订、投影已连接、材料已打印
初始状态：当前时间9:00，会议室占用到10:00
动作集：[预订会议室，连接投影，打印材料...]
约束：打印需要15分钟，投影连接需会议室空闲

这时候规划器需要：

1. 识别打印必须在会议开始前完成
2. 发现10:00前会议室不可用
3. 生成序列：[10:00预订会议室，9:45开始打印，10:05连接投影]

实现时我用状态机来建模：

class TaskPlanner:
    def plan(self, goal_state, current_state):
        # 这里踩过坑：一定要先检查可达性
        if not self.is_reachable(goal_state):
            return self.suggest_alternative(goal_state)  # 别直接抛异常
        
        # 反向链式规划：从目标倒推
        path = []
        state = goal_state
        while state != current_state:
            action = self.find_action_leading_to(state)  # 找能到达当前状态的动作
            if not action:
                break  # 规划失败
            path.insert(0, action)  # 逆序插入
            state = self.apply_action_reverse(state, action)  # 反向应用动作
        return path

这个反向规划的方法在资源有限的嵌入式系统里特别有用，因为可以从目标约束出发，避免探索无效分支。

执行时的动态调整

规划好的任务在执行时经常遇到意外。上周的物流Agent就遇到这种情况：

规划路径：A->B->C
执行时发现：B点临时关闭

这时候需要重新规划，但不能完全推倒重来。我的做法是局部重规划：

def execute_with_monitoring(task_sequence):
    context = {}
    for i, task in enumerate(task_sequence):
        try:
            result = execute(task, context)
            context.update(result)
        except ConditionChanged as e:  # 环境变化
            # 只重规划剩余部分
            remaining = task_sequence[i:]
            new_plan = replan(remaining, context, e.new_condition)
            task_sequence = task_sequence[:i] + new_plan
            # 注意这里要回退一步重新执行当前任务
            continue

这里有个细节：重规划时要保留已执行任务产生的上下文（比如已经获取到的数据），否则会出现信息断层。

多层分解策略

复杂任务需要分层处理。我设计过的一个芯片测试Agent采用三级分解：

用户指令：“全面测试通信模块”
L1分解：射频测试 + 协议栈测试 + 压力测试
L2分解：协议栈测试 = 连接测试 + 数据传输测试 + 异常恢复测试  
L3分解：连接测试 = 发起连接 + 验证握手 + 确认参数

每层使用不同的规划策略：

• L1用依赖图分析（哪些测试可以并行）
• L2用时间预估（避免超时）
• L3用实时资源分配（GPIO、DMA通道等）

代码结构大致这样：

class HierarchicalPlanner:
    def decompose(self, task, level):
        if level == "high":
            return self.split_by_domain(task)  # 按功能域切分
        elif level == "middle":
            return self.split_by_workflow(task)  # 按工作流切分
        else:
            return self.split_to_atomic(task)  # 拆到原子操作
    
    def plan(self, task):
        # 自顶向下分解
        high_level_tasks = self.decompose(task, "high")
        full_plan = []
        for hl_task in high_level_tasks:
            mid_level = self.decompose(hl_task, "middle")
            for ml_task in mid_level:
                atomic_tasks = self.decompose(ml_task, "atomic")
                full_plan.extend(self.sequence_atomic_tasks(atomic_tasks))
        return self.optimize_parallelism(full_plan)  # 最后做并行优化

注意原子任务的识别边界：一个操作如果可能失败且需要独立回滚，就应该作为原子任务。

---

实现ReAct模式：一步步构建会思考的Agent

ReAct到底在解决什么？

你肯定遇到过这种场景：让Agent查“北京明天适合穿什么衣服”，它可能直接调天气API返回一串数据给你，让你自己判断。这显然不够智能。ReAct的核心思想是让Agent学会“边想边做”——先推理当前该做什么，再执行动作，观察结果，继续推理下一步。

拆开看就是Reasoning和Acting的交替循环。这个模式特别适合需要多步骤决策的场景，比如故障诊断、复杂查询、动态规划类任务。下面我们动手搞一个。

---

搭建基础思考循环

先定义Agent的思考状态机。别一上来就搞复杂的状态模式，咱们从最简单的字典开始：

class ReActAgent:
    def __init__(self, tools):
        self.tools = tools  # 可调用的工具集
        self.memory = []    # 历史记录
        self.max_steps = 10  # 防死循环
        
    def run(self, query):
        # 初始提示词，这个模板我调了三天才稳定
        prompt = f"""
        目标：{query}
        
        你可以使用以下工具：
        {self._format_tools()}
        
        历史记录：
        {self._format_memory()}
        
        请按以下格式回应：
        思考：<你的推理>
        动作：<工具名>|<输入参数>
        或
        最终答案：<结论>
        """
        
        for step in range(self.max_steps):
            # 调用LLM生成响应
            response = self._call_llm(prompt)
            
            if "最终答案：" in response:
                return self._extract_answer(response)
                
            # 解析动作并执行
            tool_name, input_arg = self._parse_action(response)
            observation = self._use_tool(tool_name, input_arg)
            
            # 关键步骤：把本轮思考和观察记下来
            self.memory.append({
                "step": step,
                "thought": self._extract_thought(response),
                "action": (tool_name, input_arg),
                "observation": observation
            })
            
            # 更新提示词，加入新的历史
            prompt = self._update_prompt(prompt, observation)
            
        return "超过最大步数，任务失败"  # 这里踩过坑，必须设上限

注意那个max_steps，没它的话Agent可能陷入“思考-查询-再思考-再查询”的无限循环。特别是当工具返回的结果不明确时，LLM容易钻牛角尖。

---

工具调用那点坑

工具层看起来简单，但细节能坑死人。先看一个反面教材：

# 别这样写！
def call_tool(tool_name, arg):
    if tool_name == "search_web":
        return google_search(arg)  # 可能超时、可能被封
    elif tool_name == "query_db":
        return db.execute(arg)     # SQL注入警告！

正确的写法要包含异常处理、超时控制、结果清洗：

def safe_tool_call(tool_name, arg):
    # 工具映射表，避免if-else链
    tool_map = {
        "search": self._safe_search,
        "calculate": self._calculator,
        "get_time": lambda x: datetime.now().isoformat()
    }
    
    if tool_name not in tool_map:
        return f"未知工具：{tool_name}"
    
    try:
        # 关键：限制执行时间
        with timeout(seconds=5):
            result = tool_map[tool_name](arg)
            # 结果不能太长，否则会撑爆上下文
            return str(result)[:500]
    except Exception as e:
        # 返回错误信息供Agent推理
        return f"工具执行失败：{type(e).__name__}"

这里有个经验：工具返回的观察结果要简洁但信息完整。曾经因为返回了整页HTML，Agent居然开始解析HTML标签，完全跑偏了。

---

提示词工程实战

ReAct的提示词需要精心设计。分享一个我目前在用的模板：

REACT_TEMPLATE = """
你是一个善于分步解决问题的助手。请交替进行思考和行动。

当前任务：{task}

可用工具：
{tool_descs}

格式要求：
1. 先写“思考：<推理过程>”
2. 再写“动作：<工具名>|<输入>”
3. 看到工具返回后，继续思考下一步
4. 当足够确信时，输出“最终答案：<答案>”

历史记录：
{history}

现在开始：
"""

注意几个要点：

• 工具描述要写清楚输入输出格式，比如“search(query: str) -> str”
• 历史记录要倒序显示最近几条，避免上下文过长
• 在最后加“现在开始：”能显著提高格式遵循率

我试过让GPT-4自己生成思考过程，但效果不稳定。后来改成要求“用中文简短思考”，反而更符合预期。

---

调试技巧：给Agent装个仪表盘

当Agent行为异常时，别光看日志。我写了个简单的监控装饰器：

def debug_agent(func):
    def wrapper(*args, **kwargs):
        print(f"[Agent调用] {func.__name__}")
        print(f"输入：{kwargs.get('query', '无')}")
        
        result = func(*args, **kwargs)
        
        print(f"步数：{len(result['steps'])}")
        for step in result['steps'][-3:]:  # 只看最后三步
            print(f"  思考：{step['thought'][:50]}...")
            print(f"  动作：{step['action']}")
            
        return result
    return wrapper

通过观察最后几步的思考-动作对，能快速发现Agent是在原地打转还是在推进问题。常见的问题模式：

1. 思考越来越长但动作不变 -> 陷入过度推理
2. 动作在不同工具间跳变 -> 目标不明确
3. 观察结果被忽略 -> 历史记录有问题

---

性能优化实战

当步数增多时，上下文长度会爆炸。我的解决方案是选择性记忆：

def compress_memory(self, memory_list):
    """只保留关键历史"""
    compressed = []
    for m in memory_list:
        # 跳过纯查询无结果的步骤
        if "未找到" in m["observation"] and "搜索" in m["action"]:
            continue
        # 保留有决策价值的步骤
        if "确定" in m["thought"] or "选择" in m["thought"]:
            compressed.append(m)
    return compressed[-6:]  # 最多保留6步

另一个技巧是让Agent自己总结进度。在每5步后插入一个总结步骤：

if step % 5 == 0:
    summary_prompt = f"请用一句话总结当前进展：{history[-5:]}"
    summary = self._call_llm(summary_prompt)
    self.memory.append({"summary": summary})

这样即使处理长任务，上下文也不会无限增长。

---

个人经验建议

1. 从简单工具开始：先实现搜索、计算、时间查询这三个基础工具，验证思考循环能跑通，再加复杂工具。

2. 设置明确的停止条件：除了最大步数，还可以让Agent在连续两次动作相同时自动停止，防止死循环。

3. 给工具分类：我把工具分为“信息获取类”和“动作执行类”，在提示词中明确告知Agent：前者可以多次调用，后者确认无误后再执行。

4. 接受不完美：早期版本能达到70%的任务完成率就可以继续往下做，过度优化单点不如完善整体架构。

5. 人机协作设计：在关键决策点让Agent给出选项，由用户选择。完全自主的Agent在复杂场景下反而不可靠。

最后说句实话：ReAct不是银弹。对于简单查询，它比直接调用工具慢；但对于需要多源信息整合、动态决策的场景，这种“慢思考”恰恰是价值所在。就像老工程师解决问题时，总会先停下来想清楚再动手——我们现在只是把这种能力赋予了机器。