my_plan_solve_agent.py
代码实现
# my_plan_solve_agent.py
import re
import ast
from typing import List, Dict, Optional
from hello_agents.core.llm import HelloAgentsLLM# 默认规划器提示词模板
DEFAULT_PLANNER_PROMPT = """
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。问题: {question}请严格按照以下格式输出你的计划:
```python
["步骤1", "步骤2", "步骤3", ...]
"""
默认执行器提示词模板
DEFAULT_EXECUTOR_PROMPT = """
你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。
你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
请你专注于解决"当前步骤"，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。
原始问题:
完整计划:
历史步骤与结果:
当前步骤:
请仅输出针对"当前步骤"的回答:
"""
class MyPlanAndSolveAgent:
"""
Plan-and-Solve (规划与执行) 智能体
"""
def init(
self,
name: str,
llm: HelloAgentsLLM,
custom_prompts: Optional[Dict[str, str]] = None
):
self.name = name
self.llm = llm
self._history = [] # 用于记录长期的对话历史
# 灵活加载提示词：有自定义就用自定义，没有就用默认self.prompts = custom_prompts if custom_prompts else {"planner": DEFAULT_PLANNER_PROMPT,"executor": DEFAULT_EXECUTOR_PROMPT}print(f"✅ {self.name} 初始化完成 (Plan-and-Solve 架构)")def get_history(self) -> List[dict]:"""获取对话历史"""return self._historydef _parse_plan(self, text: str) -> List[str]:"""核心辅助方法：使用正则表达式和ast从LLM的纯文本回复中提取Python列表"""match = re.search(r'\[.*\]', text, re.DOTALL)if match:try:# ast.literal_eval 能安全地将字符串形态的列表转化为真正的 Python Listplan_list = ast.literal_eval(match.group(0))if isinstance(plan_list, list):return plan_listexcept Exception:pass# 兜底机制：如果大模型没有按格式输出，就按换行符强制分割return [line.strip('- *') for line in text.split('\n') if line.strip()]def run(self, question: str) -> str:"""核心引擎：处理规划与执行循环"""print(f"\n🤖 {self.name} 开始处理问题: {question}")# === 阶段 1: 规划 (Planning) ===print("\n--- 阶段 1: 规划 (Planning) ---")planner_prompt = self.prompts["planner"].format(question=question)plan_response = self.llm.invoke([{"role": "user", "content": planner_prompt}])# 解析提取步骤plan_list = self._parse_plan(plan_response)print("📝 生成的计划步骤:")for i, step in enumerate(plan_list):print(f" {i+1}. {step}")# === 阶段 2: 执行 (Solving) ===print("\n--- 阶段 2: 执行 (Solving) ---")history_str = ""final_answer = ""for i, current_step in enumerate(plan_list):print(f"\n▶ 执行步骤 {i+1}: {current_step}")# 组装上下文环境（包含历史结果，形成思维链条）executor_prompt = self.prompts["executor"].format(question=question,plan=str(plan_list),history=history_str if history_str else "无",current_step=current_step)step_result = self.llm.invoke([{"role": "user", "content": executor_prompt}])print(f" 💡 结果: {step_result}")# 将当前结果累加到历史中，供下一步使用history_str += f"步骤 {i+1} ({current_step}): {step_result}\n"# 不断覆盖，直到循环结束，拿到的就是终局答案final_answer = step_result # === 任务收尾 ===# 将本次完整任务归档到 Agent 记忆中self._history.append({"role": "user", "content": question})self._history.append({"role": "assistant", "content": final_answer})print("\n✨ 任务全部执行完成！")return final_answer
my_react_agent.py
代码实现
MY_REACT_PROMPT = """你是一个具备推理和行动能力的AI助手。你可以通过思考分析问题，然后调用合适的工具来获取信息，最终给出准确的答案。
可用工具
{tools}
工作流程
请严格按照以下格式进行回应，每次只能执行一个步骤：
Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。
Action: 你决定采取的行动，必须是以下格式之一：
{{tool_name}}[{{tool_input}}]- 调用指定工具Finish[最终答案]- 当你有足够信息给出最终答案时
重要提醒
- 每次回应必须包含Thought和Action两部分
- 工具调用的格式必须严格遵循：工具名[参数]
- 只有当你确信有足够信息回答问题时，才使用Finish
- 如果工具返回的信息不够，继续使用其他工具或相同工具的不同参数
当前任务
Question: {question}
执行历史
{history}
现在开始你的推理和行动：
"""
import re
from typing import Optional, List, Tuple
from hello_agents import ReActAgent, HelloAgentsLLM, Config, Message, ToolRegistry
class MyReActAgent(ReActAgent):
"""
重写的ReAct Agent - 推理与行动结合的智能体
"""
def __init__(self,name: str,llm: HelloAgentsLLM,tool_registry: ToolRegistry,system_prompt: Optional[str] = None,config: Optional[Config] = None,max_steps: int = 5,custom_prompt: Optional[str] = None
):super().__init__(name, llm, system_prompt, config)self.tool_registry = tool_registryself.max_steps = max_stepsself.current_history: List[str] = []self.prompt_template = custom_prompt if custom_prompt else MY_REACT_PROMPTprint(f"✅ {name} 初始化完成，最大步数: {max_steps}")def run(self, input_text: str, **kwargs) -> str:"""运行ReAct Agent"""self.current_history = []current_step = 0print(f"\n🤖 {self.name} 开始处理问题: {input_text}")while current_step < self.max_steps:current_step += 1print(f"\n--- 第 {current_step} 步 ---")# 1. 构建提示词tools_desc = self.tool_registry.get_tools_description()history_str = "\n".join(self.current_history)prompt = self.prompt_template.format(tools=tools_desc,question=input_text,history=history_str)# 2. 调用LLMmessages = [{"role": "user", "content": prompt}]response_text = self.llm.invoke(messages, **kwargs)# 3. 解析输出thought, action = self._parse_output(response_text)# 4. 检查完成条件if action and action.startswith("Finish"):final_answer = self._parse_action_input(action)self.add_message(Message(input_text, "user"))self.add_message(Message(final_answer, "assistant"))return final_answer# 5. 执行工具调用if action:tool_name, tool_input = self._parse_action(action)observation = self.tool_registry.execute_tool(tool_name, tool_input)self.current_history.append(f"Action: {action}")self.current_history.append(f"Observation: {observation}")# 达到最大步数final_answer = "抱歉，我无法在限定步数内完成这个任务。"self.add_message(Message(input_text, "user"))self.add_message(Message(final_answer, "assistant"))return final_answer