Hello Agents | 第四章习题

[matrixcloud]

第四章习题

1. 本章介绍了三种经典的智能体范式:ReAct、Plan-and-Solve 和 Reflection。请分析

a. 这三种范式在"思考"与"行动"的组织方式上有什么本质区别？

• ReAct: 思考后进行行动，如果未完成目标，则将行动的结果作为新的观察作为输入，开始新轮的思考-行动。思考与行动交错执行，思考直接驱动行动。
• Plan-and-Solve: 先由计划单元思考和规划完成目标的需要的步骤，交由执行单元行动。行动主要由事先规划驱动。
• Reflection: 先由初始器给出初始解，由反思器在此基础上进行思考并提出优化建议，最后优化器根据优化建议进行优化，直到反思器认为以达到最优解。行动由优化器执行，依据反思器的思考结果行动。

b. 如果要设计一个"智能家居控制助手"（需要控制灯光、空调、窗帘等多个设备，并根据用户习惯自动调节），你会选择哪种范式作为基础架构？为什么？

首先明确，什么是依据用户习惯自动调节？可以理解为，智能体根据用户的历史操作数据进行自动调节，例如近三天用户习惯于晚上 10 点打开卧室的灯，然后接着打开空调并设置到 25 度。当智能体识别到这个习惯后，应该调整设定好的规则。这样的场景，非常适用于使用 Reflection范式，用户可能在最初设定了一个使用的规则集，现在根据用户的习惯反思并提出优化建议，最后由执行层去执行，最后得到一个最终版本用于设定新的规则集。

c. 是否可以将这三种范式进行组合使用？若可以，请尝试设计一个混合范式的智能体架构，并说明其适用场景。

可以，例如 coding agent。使用 Plan-and-Solve 范式来规划新需求的任务，接着用 ReAct范式来对子任务进行开发，在提交代码的时候，可以使用 Reflection范式来优化相应的代码。

2. 在4.2节的 ReAct 实现中，我们使用了正则表达式来解析大语言模型的输出（如 Thought 和 Action）。请思考

a. 当前的解析方法存在哪些潜在的脆弱性？在什么情况下可能会失败？

• 当前的正则很容易出现匹配匹配不了的情况，例如 Thought 和 Action 中间没有换行符的情况。
• 如果字符串过长，可能会导致正则表达式的性能问题，导致解析速度变慢或崩溃。
• 安全问题，ReDOS 攻击，导致系统资源耗尽。

b. 除了正则表达式，还有哪些更鲁棒的输出解析方案？

使用严格标准的返回格式，例如 JSON 或 XML。

c. 尝试修改本章的代码，使用一种更可靠的输出格式，并对比两种方案的优缺点

REACT_PROMPT_TEMPLATE = """
请注意，你是一个有能力调用外部工具的智能助手。

可用工具如下：
{tools}

请严格按照以下格式进行回应，**只返回纯JSON，不要包含任何其他文字**：

{{
    "thought": "你的思考过程，用于分析问题、拆解任务和规划下一步行动。",
    "action": {{
        "type: "action_type",
        "payload": action_payload
    }}
}}

action 字段说明：

- action_type: enum (tool_use, finish)
    - tool_use: 调用一个工具。
    - finish: 当你认为已经获得最终答案时。
- action_payload: action 为 finish 时为 string 类型的研究结论, 否则定义如下：

{{
    "name": "tool name",
    "input": "tool input"
}}

现在，请开始解决以下问题：
Question: {question}
History: {history}
"""

• JSON 格式: 优点是简单、易读、支持嵌套结构。缺点是需要模型输出 JSON 格式，可能会增加额外的计算成本。
• 纯文本格式: 优点是简单、易读、不需要额外的计算成本。缺点是脆弱，容易出错。

3. 工具调用是现代智能体的核心能力之一。基于4.2.2节的 ToolExecutor 设计，请完成以下扩展实践

• 为 ReAct 智能体添加一个"计算器"工具，使其能够处理复杂的数学计算问题（如"计算 (123 + 456) × 789/ 12 = ? 的结果"）

def calculator(expression: str) -> str:
    """
    一个安全的计算器工具，支持基本的数学表达式计算。
    """
    print(f"🧮 正在执行 [Calculator] 计算: {expression}")
    
    def eval_node(node: ast.expr) -> int | float:
        if isinstance(node, ast.Constant):
            if isinstance(node.value, (int, float)):
                return node.value
            raise ValueError(f"非数字类型: {type(node.value)}")
        elif isinstance(node, ast.BinOp):
            left_val = eval_node(node.left)
            right_val = eval_node(node.right)
            
            if type(node.op) in SUPPORTED_OPERATORS:
                return SUPPORTED_OPERATORS[type(node.op)](left_val, right_val)
            raise NotImplementedError(f"不支持的二元运算符: {node.op}")
        elif isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
            return -eval_node(node.operand)
        else:
            raise NotImplementedError(f"不支持的节点类型: {type(node)}")
        
    try:
        tree = ast.parse(expression, mode='eval')
        return str(eval_node(tree.body))
    except Exception as e:
        return f"计算时发生错误: {e}"

• 设计并实现一个"工具选择失败"的处理机制: 当智能体多次调用错误的工具或提供错误的参数时，系统应该如何引导它纠正？

实现 DEMO

REACT_PROMPT_TEMPLATE = """
请注意，你是一个有能力调用外部工具的智能助手。

可用工具如下：
{tools}

请严格按照以下格式进行回应，**只返回纯JSON，不要包含任何其他文字**：

{{
    "thought": "你的思考过程，用于分析问题、拆解任务和规划下一步行动。",
    "action": {{
        "type: "action_type",
        "payload": action_payload
    }}
}}

action 字段说明：

- action_type: enum (tool_use, finish)
    - tool_use: 调用一个工具。
    - finish: 当你认为已经获得最终答案时。
- action_payload: action 为 finish 时为 string 类型的研究结论, 否则定义如下：

{{
    "name": "tool name",
    "input": "tool input"
}}

现在，请开始解决以下问题：
Question: {question}
History: {history}
Corrections: {corrections}
"""

@dataclass
class ToolFailure:
    step: int
    tool: str | None = None
    input: str | None = None
    exception: Exception | None = None

class ReActAgent:
    def __init__(
        self,
        llm_client: HelloAgentsLLM,
        tool_executor: ToolExecutor,
        max_steps: int = 5,
    ):
        self.llm_client = llm_client
        self.tool_executor = tool_executor
        self.max_steps = max_steps
        self.history: List[str] = []
        self.failures: List[ToolFailure] = []

    def run(self, question: str):
        self.history = []
        current_step = 0

        while current_step < self.max_steps:
            current_step += 1
            print(f"\n--- 第 {current_step} 步 ---")

            tools_desc = self.tool_executor.getAvailableTools()
            history_str = "\n".join(self.history)
            corrections = self._get_corrections()

            prompt = REACT_PROMPT_TEMPLATE.format(
                tools=tools_desc,
                question=question,
                history=history_str,
                corrections=corrections,
            )

            messages = [{"role": "user", "content": prompt}]
            response_text = self.llm_client.think(messages=messages)
            if not response_text:
                print("错误：LLM未能返回有效响应。")
                break

            thought, action = self._parse_output(response_text)
            if thought:
                print(f"🤔 思考: {thought}")
            if not action:
                print("警告：未能解析出有效的Action，流程终止。")
                break

            action_type, action_payload = self._parse_action(action)
            if action_type == "finish":
                # 如果是Finish指令，提取最终答案并结束
                final_answer = action_payload
                print(f"🎉 最终答案: {final_answer}")
                return final_answer
            
            assert isinstance(action_payload, dict), (
                "tool_use类型的payload必须是一个字典。"
            )
            tool_name, tool_input = (
                action_payload.get("name"),
                action_payload.get("input"),
            )
            if not tool_name or not tool_input:
                self._record_failure(current_step, tool_name, tool_input)
                continue

            observation = self._call_tool(current_step, tool_name, tool_input)

            print(f"👀 观察: {observation}")
            self.history.append(f"Action: {action}")
            self.history.append(f"Observation: {observation}")

        print("已达到最大步数，流程终止。")
        return None

    def _call_tool(self, step: int, tool_name: str, tool_input: str) -> str | None:
        print(f"🎬 行动: {tool_name}[{tool_input}]")
        tool_function = self.tool_executor.getTool(tool_name)
        if not tool_function:
            self._record_failure(step, tool_name, tool_input, exception=ValueError(f"tool not found: {tool_name}"))
        try:
            return tool_function(tool_input)
        except Exception as e:
            self._record_failure(step, tool_name, tool_input, exception=e)

    def _get_corrections(self) -> str:
        if not self.failures:
            return ""
        last = self.failures[-1]
        corrections = ""
        
        if not last.tool:
            corrections = f"缺失工具名称，请检查。可用工具列表: {self.tool_executor.getAvailableTools()}"
        elif not last.input:
            corrections = f"工具'{last.tool}'参数错误: '{last.input}'。输入参数不能为空"
        elif not last.exception:
            corrections = f"工具调用失败: {last.tool}[{last.input}]，但未捕获到异常信息。"
        else:
            corrections = f"工具调用失败: {last.tool}[{last.input}]。具体错误信息: {last.exception}"
        
        consecutive_failures = self._consecutive_failures()
        
        return corrections if consecutive_failures < 3 else f"已连续失败{consecutive_failures}次，请直接 finish 并给出最终答案。"
    
    def _record_failure(self, step: int, tool_name: str | None = None, input: str | None = None, exception: Exception | None = None):
        self.failures.append(ToolFailure(step=step, tool=tool_name, input=input, exception=exception))
        
    def _consecutive_failures(self) -> int:
        """计算连续失败次数（从后往前数）"""
        count = 0
        for f in reversed(self.failures):
            if f.step == len(self.failures) - count:  # 连续编号判断
                count += 1
            else:
                break
        return count

    def _parse_output(self, text: str):
        # 尝试从响应文本中提取JSON部分
        # 改进正则表达式，支持嵌套的大括号
        json_match = re.search(r"\{.*\}", text, re.DOTALL)
        if json_match:
            json_str = json_match.group()
            try:
                res: dict = json.loads(json_str)
                thought: str = res.get("thought", "")
                action: dict = res.get("action", {})
                return thought, action
            except json.JSONDecodeError as e:
                print(f"JSON解析错误: {e}")
                print(f"尝试解析的JSON字符串: {json_str}")

        # 如果JSON解析失败，返回默认值
        return "", {}

    def _parse_action(self, action: dict[str, Any]):
        type: str = action.get("type", "")

        match type:
            case "finish":
                answer: str = action.get("payload", "")
                return type, answer
            case "tool_use":
                payload: dict[str, Any] = action.get("payload", {})
                return type, payload
            case _:
                raise ValueError(f"未知的Action类型: {type}")


if __name__ == "__main__":
    print("=" * 60)
    print("ReActAgent Error Handler Mock 测试")
    print("=" * 60)

    # 注册工具
    tool_executor = ToolExecutor()
    tool_executor.registerTool(
        "Calculator", "一个简单的计算器工具，用于计算数学表达式。", calculator
    )

    # 创建 MockLLM，按顺序返回预设响应
    class MockLLM(HelloAgentsLLM):
        def __init__(self, responses: list):
            self.responses = responses
            self.call_count = 0

        def think(self, messages: list) -> str:
            if self.call_count < len(self.responses):
                resp = self.responses[self.call_count]
                self.call_count += 1
                print(f"   [MockLLM] 返回第 {self.call_count} 个预设响应")
                return resp
            # 兜底：直接 finish
            return '{"thought":"结束","action":{"type":"finish","payload":"测试完成"}}'

    # ===== 测试1: 验证各类 _get_corrections 分支 =====
    print("\n--- 测试1: 各类错误纠正信息 ---")
    agent = ReActAgent(llm_client=HelloAgentsLLM(), tool_executor=tool_executor)

    # 1.1 工具名称缺失
    agent._record_failure(1, None, "some input", None)
    print(f"工具名称缺失: {agent._get_corrections()}")
    agent.failures.clear()

    # 1.2 输入参数为空
    agent._record_failure(1, "Calculator", "", None)
    print(f"输入参数为空: {agent._get_corrections()}")
    agent.failures.clear()

    # 1.3 工具执行异常
    agent._record_failure(1, "Calculator", "1/0", ZeroDivisionError("division by zero"))
    print(f"工具执行异常: {agent._get_corrections()}")
    agent.failures.clear()

    # ===== 测试2: 3次连续失败的端到端测试 =====
    print("\n--- 测试2: 3次连续失败触发 finish 提示 ---")

    # 预设响应：连续3次空参数导致失败，第4次 finish
    mock_responses = [
        '{"thought":"尝试计算","action":{"type":"tool_use","payload":{"name":"Calculator","input":""}}}',
        '{"thought":"再次尝试","action":{"type":"tool_use","payload":{"name":"Calculator","input":""}}}',
        '{"thought":"第三次尝试","action":{"type":"tool_use","payload":{"name":"Calculator","input":""}}}',
        '{"thought":"连续失败，直接结束","action":{"type":"finish","payload":"由于工具调用失败，无法给出答案"}}',
    ]

    mock_llm = MockLLM(mock_responses)
    agent = ReActAgent(llm_client=mock_llm, tool_executor=tool_executor, max_steps=5)

    result = agent.run("测试问题：1 + 1 = ?")

    print(f"\n【测试结果】")
    print(f"最终答案: {result}")
    print(f"LLM 调用次数: {mock_llm.call_count}")
    print(f"总失败记录数: {len(agent.failures)}")
    print(f"连续失败次数: {agent._consecutive_failures()}")

    # 断言验证
    assert len(agent.failures) == 3, f"期望3次失败，实际 {len(agent.failures)} 次"
    assert agent._consecutive_failures() == 3, f"期望连续失败3次，实际 {agent._consecutive_failures()} 次"
    assert result is not None, "应该获得最终答案"

    print("\n✓ 所有测试通过！")

• 思考: 如果可调用工具的数量增加到50个甚至100个，当前的工具描述方式是否还能有效工作？在可调用工具数量随业务需求显著增加时，从工程角度如何优化工具的组织和检索机制？

当调用工具急剧增加时，当前的工具描述方式可能有效，但是效率会下降，另外如果用户本身的输入很长，再加上工具的描述，可能导致 LLM 出现健忘的情况，甚至导致输入超出模型的最大长度。我认为，可以定义两层索引，第一层索引是工具的分类，例如"计算工具"、"数据工具"等；第二层索引是具体工具的名称、使用场景以及加载位置；LLM 先找分类，然后再找需要加载的工具，以实现懒加载。

4. Plan-and-Solve 范式将任务分解为"规划"和"执行"两个阶段。请深入分析

• 在4.3节的实现中，规划阶段生成的计划是"静态"的（一次性生成，不可修改）。如果在执行过程中发现某个步骤无法完成或结果不符合预期，应该如何设计一个"动态重规划"机制？

首先将存储步骤的数据结构改成有向图，每个节点表示一个步骤，每次规划实际上就是从起点规划一条路径到终点，对于某个步骤无法完成的情况，可以让 LLM 回溯到上一个成功节点，重新规划路径。

• 对比 Plan-and-Solve 与 ReAct:在处理"预订一次从北京到上海的商务旅行（包括机票、酒店、租车）"这样的任务时，哪种范式更合适？为什么？

ReAct 范式更合适，因为它能够动态地调用工具，而 Plan-and-Solve 是静态的，只能在规划阶段就确定好所有的步骤。比如 ReAct 能处理机票库存不足的情况，而 Plan-and-Solve 不太适合。

• 尝试设计一个"分层规划"系统:先生成高层次的抽象计划，然后针对每个高层步骤再生成详细的子计划。这种设计有什么优势？

通过高层的抽象可以指定做什么（What），而低层完成怎么做（How）。这样做有如下优点：

• 更高的复用性和模块化：低层会变成的更精细和更通用。
• 更好的可读性和可维护性：高层的计划可以更加清晰和简洁，更容易理解和维护。
• 良好的错误隔离：如果子计划执行失败，只需在局部重新规划即可，无需推翻整个计划。
• 良好的异步和分布式支持：子计划可以并行执行，也可以在不同的节点上进行。

5. Reflection 机制通过"执行-反思-优化"循环来提升输出质量。请思考

• 在4.4节的代码生成案例中，不同阶段使用的是同一个模型。如果使用两个不同的模型（例如，用一个更强大的模型来做反思，用一个更快的模型来做执行），会带来什么影响？

首先，反思出来的建议会更好；其次，执行模型会更快，能够更快地响应用户的需求。Agent 自我优化的轮次会更少，能够更快地收敛到最优解。

• Reflection 机制的终止条件是"反馈中包含无需改进"或"达到最大迭代次数"。这种设计是否合理？能否设计一个更智能的终止条件？

不是太合理的设计。反馈中包含无需改进，实际上是一个主观的判断，达到最大迭代次数，可能会导致智能体提前结束，而没有机会得到最优解。一个更智能的终止条件需要考虑多方面因素：

a. 消耗的 Token 数量
b. 智能体的执行时间
c. 定义多维度的评判指标
d. 基于语义相似度的判断
e. ...

总之，要考虑到模型求解的边际成本，添加多维度的评判指标。

• 假设你要搭建一个"学术论文写作助手"，它能够生成初稿并不断优化论文内容。请设计一个多维度的Reflection机制，从段落逻辑性、方法创新性、语言表达、引用规范等多个角度进行反思和改进。

TODO

6. 提示词工程是影响智能体最终效果的关键技术。本章展示了多个精心设计的提示词模板。请分析

a. 对比4.2.3节的 ReAct 提示词和4.3.2节的 Plan-and-Solve 提示词，它们显然存在结构设计上的明显不同，这些差异是如何服务于各自范式的核心逻辑的？

• ReAct 在提示词主要包含：任务描述，可用的工具集，定义的 LLM 的输出格式(Thought-Action)，历史Action 和 执行结果Observation；
• Plan-and-Solve 包含两个提示词：
  • 规划提示词：包含任务描述，还有生成步骤的格式。
  • 执行提示词：包含任务描述，执行步骤，当前执行步骤，历史执行步骤和结果。

b. 在4.4.3节的 Reflection 提示词中，我们使用了"你是一位极其严格的代码评审专家"这样的角色设定。尝试修改这个角色设定（如改为"你是一位注重代码可读性的开源项目维护者"），观察输出结果的变化，并总结角色设定对智能体行为的影响。

TODO

c. 在提示词中加入 few-shot 示例往往能显著提升模型对特定格式的遵循能力。请为本章的某个智能体尝试添加 few-shot 示例，并对比其效果。

TODO

7. 某电商初创公司现在希望使用"客服智能体"来代替真人客服实现降本增效，它需要具备以下功能

• a. 理解用户的退款申请理由
• b. 查询用户的订单信息和物流状态
• c. 根据公司政策智能地判断是否应该批准退款
• d. 生成一封得体的回复邮件并发送至用户邮箱
  e. 如果判断决策存在一定争议（自我置信度低于阈值），能够进行自我反思并给出更审慎的建议

此时作为该产品的负责人:

• 你会选择本章的哪种范式（或哪些范式的组合）作为系统的核心架构？
• 这个系统需要哪些工具？请列出至少3个工具及其功能描述。
• 如何设计提示词来确保智能体的决策既符合公司利益，又能保持对用户的友好态度？
• 这个产品上线后可能面临哪些风险和挑战？如何通过技术手段来降低这些风险？

[Skitre]

1

[zd08135]

my flask app 代码在哪里？这个代码维护助手如何运行？

[EthanCherry]

smithery那里根本操作不了，天呐