本节内容将在第四章构建的三种经典Agent范式（ReAct、Plan-and-Solve、Reflection）基础上进行框架化重构，并新增SimpleAgent作为基础对话范式。我们将把这些独立的Agent实现，改造为基于统一架构的框架组件。本次重构主要围绕以下三个核心目标展开：
- 提示词工程的系统性提升：对第四章中的提示词进行深度优化，从特定任务导向转向通用化设计，同时增强格式约束和角色定义。
- 接口与格式的标准化统一：建立统一的Agent基类和标准化的运行接口，所有Agent都遵循相同的初始化参数、方法签名和历史管理机制。
- 高度可配置的自定义能力：支持用户自定义提示词模板、配置参数和执行策略。
7.4.1 SimpleAgent
SimpleAgent是最基础的Agent实现，它展示了如何在框架基础上构建一个完整的对话智能体。我们将通过继承框架中已有的SimpleAgent类并重写其核心方法，来实现一个可扩展的版本。首先，在你的项目目录中创建一个my_simple_agent.py文件：
# my_simple_agent.py
from typing import Optional, Iterator
from hello_agents import SimpleAgent, HelloAgentsLLM, Config, Message,ToolRegistry
class MySimpleAgent(SimpleAgent):
"""
重写的简单对话Agent
展示如何基于框架基类构建自定义Agent
"""
def __init__(
self,
name: str,
llm: HelloAgentsLLM,
system_prompt: Optional[str] = None,
config: Optional[Config] = None,
tool_registry: Optional['ToolRegistry'] = None,
enable_tool_calling: bool = True
):
super().__init__(name, llm, system_prompt, config)
self.tool_registry = tool_registry
self.enable_tool_calling = enable_tool_calling and tool_registry is not None
print(f"✅ {name} 初始化完成，工具调用: {'启用' if self.enable_tool_calling else '禁用'}")
接下来，我们需要重写run方法。SimpleAgent支持可选的工具调用功能，也方便后续章节的扩展：
# 继续在 my_simple_agent.py 中添加
import re
class MySimpleAgent(SimpleAgent):
# ... 前面的 __init__ 方法
def run(self, input_text: str, max_tool_iterations: int = 3, **kwargs) -> str:
"""
重写的运行方法 - 实现简单对话逻辑，支持可选工具调用
"""
print(f"🤖 {self.name} 正在处理: {input_text}")
# 构建消息列表
messages = []
# 添加系统消息（可能包含工具信息）
enhanced_system_prompt = self._get_enhanced_system_prompt()
messages.append({"role": "system", "content": enhanced_system_prompt})
# 添加历史消息
for msg in self._history:
messages.append({"role": msg.role, "content": msg.content})
# 添加当前用户消息
messages.append({"role": "user", "content": input_text})
# 如果没有启用工具调用，使用简单对话逻辑
if not self.enable_tool_calling:
response = self.llm.invoke(messages, **kwargs)
self.add_message(Message(input_text, "user"))
self.add_message(Message(response, "assistant"))
print(f"✅ {self.name} 响应完成")
return response
# 支持多轮工具调用的逻辑
return self._run_with_tools(messages, input_text, max_tool_iterations, **kwargs)
def _get_enhanced_system_prompt(self) -> str:
"""构建增强的系统提示词，包含工具信息"""
base_prompt = self.system_prompt or "你是一个有用的AI助手。"
if not self.enable_tool_calling or not self.tool_registry:
return base_prompt
# 获取工具描述
tools_description = self.tool_registry.get_tools_description()
if not tools_description or tools_description == "暂无可用工具":
return base_prompt
tools_section = "\n\n## 可用工具\n"
tools_section += "你可以使用以下工具来帮助回答问题:\n"
tools_section += tools_description + "\n"
tools_section += "\n## 工具调用格式\n"
tools_section += "当需要使用工具时，请使用以下格式:\n"
tools_section += "`[TOOL_CALL:{tool_name}:{parameters}]`\n"
tools_section += "例如:`[TOOL_CALL:search:Python编程]` 或 `[TOOL_CALL:memory:recall=用户信息]`\n\n"
tools_section += "工具调用结果会自动插入到对话中，然后你可以基于结果继续回答。\n"
return base_prompt + tools_section
现在我们实现工具调用的核心逻辑：
# 继续在 my_simple_agent.py 中添加
class MySimpleAgent(SimpleAgent):
# ... 前面的方法
def _run_with_tools(self, messages: list, input_text: str, max_tool_iterations: int, **kwargs) -> str:
"""支持工具调用的运行逻辑"""
current_iteration = 0
final_response = ""
while current_iteration < max_tool_iterations:
# 调用LLM
response = self.llm.invoke(messages, **kwargs)
# 检查是否有工具调用
tool_calls = self._parse_tool_calls(response)
if tool_calls:
print(f"🔧 检测到 {len(tool_calls)} 个工具调用")
# 执行所有工具调用并收集结果
tool_results = []
clean_response = response
for call in tool_calls:
result = self._execute_tool_call(call['tool_name'], call['parameters'])
tool_results.append(result)
# 从响应中移除工具调用标记
clean_response = clean_response.replace(call['original'], "")
# 构建包含工具结果的消息
messages.append({"role": "assistant", "content": clean_response})
# 添加工具结果
tool_results_text = "\n\n".join(tool_results)
messages.append({"role": "user", "content": f"工具执行结果:\n{tool_results_text}\n\n请基于这些结果给出完整的回答。"})
current_iteration += 1
continue
# 没有工具调用，这是最终回答
final_response = response
break
# 如果超过最大迭代次数，获取最后一次回答
if current_iteration >= max_tool_iterations and not final_response:
final_response = self.llm.invoke(messages, **kwargs)
# 保存到历史记录
self.add_message(Message(input_text, "user"))
self.add_message(Message(final_response, "assistant"))
print(f"✅ {self.name} 响应完成")
return final_response
def _parse_tool_calls(self, text: str) -> list:
"""解析文本中的工具调用"""
pattern = r'\[TOOL_CALL:([^:]+):([^\]]+)\]'
matches = re.findall(pattern, text)
tool_calls = []
for tool_name, parameters in matches:
tool_calls.append({
'tool_name': tool_name.strip(),
'parameters': parameters.strip(),
'original': f'[TOOL_CALL:{tool_name}:{parameters}]'
})
return tool_calls
def _execute_tool_call(self, tool_name: str, parameters: str) -> str:
"""执行工具调用"""
if not self.tool_registry:
return f"❌ 错误:未配置工具注册表"
try:
# 智能参数解析
if tool_name == 'calculator':
# 计算器工具直接传入表达式
result = self.tool_registry.execute_tool(tool_name, parameters)
else:
# 其他工具使用智能参数解析
param_dict = self._parse_tool_parameters(tool_name, parameters)
tool = self.tool_registry.get_tool(tool_name)
if not tool:
return f"❌ 错误:未找到工具 '{tool_name}'"
result = tool.run(param_dict)
return f"🔧 工具 {tool_name} 执行结果:\n{result}"
except Exception as e:
return f"❌ 工具调用失败:{str(e)}"
def _parse_tool_parameters(self, tool_name: str, parameters: str) -> dict:
"""智能解析工具参数"""
param_dict = {}
if '=' in parameters:
# 格式: key=value 或 action=search,query=Python
if ',' in parameters:
# 多个参数:action=search,query=Python,limit=3
pairs = parameters.split(',')
for pair in pairs:
if '=' in pair:
key, value = pair.split('=', 1)
param_dict[key.strip()] = value.strip()
else:
# 单个参数:key=value
key, value = parameters.split('=', 1)
param_dict[key.strip()] = value.strip()
else:
# 直接传入参数，根据工具类型智能推断
if tool_name == 'search':
param_dict = {'query': parameters}
elif tool_name == 'memory':
param_dict = {'action': 'search', 'query': parameters}
else:
param_dict = {'input': parameters}
return param_dict
我们还可以为自定义Agent添加流式响应功能和便利方法：
# 继续在 my_simple_agent.py 中添加
class MySimpleAgent(SimpleAgent):
# ... 前面的方法
def stream_run(self, input_text: str, **kwargs) -> Iterator[str]:
"""
自定义的流式运行方法
"""
print(f"🌊 {self.name} 开始流式处理: {input_text}")
messages = []
if self.system_prompt:
messages.append({"role": "system", "content": self.system_prompt})
for msg in self._history:
messages.append({"role": msg.role, "content": msg.content})
messages.append({"role": "user", "content": input_text})
# 流式调用LLM
full_response = ""
print("📝 实时响应: ", end="")
for chunk in self.llm.stream_invoke(messages, **kwargs):
full_response += chunk
print(chunk, end="", flush=True)
yield chunk
print() # 换行
# 保存完整对话到历史记录
self.add_message(Message(input_text, "user"))
self.add_message(Message(full_response, "assistant"))
print(f"✅ {self.name} 流式响应完成")
def add_tool(self, tool) -> None:
"""添加工具到Agent（便利方法）"""
if not self.tool_registry:
from hello_agents import ToolRegistry
self.tool_registry = ToolRegistry()
self.enable_tool_calling = True
self.tool_registry.register_tool(tool)
print(f"🔧 工具 '{tool.name}' 已添加")
def has_tools(self) -> bool:
"""检查是否有可用工具"""
return self.enable_tool_calling and self.tool_registry is not None
def remove_tool(self, tool_name: str) -> bool:
"""移除工具（便利方法）"""
if self.tool_registry:
self.tool_registry.unregister(tool_name)
return True
return False
def list_tools(self) -> list:
"""列出所有可用工具"""
if self.tool_registry:
return self.tool_registry.list_tools()
return []
创建一个测试文件test_simple_agent.py：
# test_simple_agent.py
from dotenv import load_dotenv
from hello_agents import HelloAgentsLLM, ToolRegistry
from hello_agents.tools import CalculatorTool
from my_simple_agent import MySimpleAgent
# 加载环境变量
load_dotenv()
# 创建LLM实例
llm = HelloAgentsLLM()
# 测试1:基础对话Agent（无工具）
print("=== 测试1:基础对话 ===")
basic_agent = MySimpleAgent(
name="基础助手",
llm=llm,
system_prompt="你是一个友好的AI助手，请用简洁明了的方式回答问题。"
)
response1 = basic_agent.run("你好，请介绍一下自己")
print(f"基础对话响应: {response1}\n")
# 测试2:带工具的Agent
print("=== 测试2:工具增强对话 ===")
tool_registry = ToolRegistry()
calculator = CalculatorTool()
tool_registry.register_tool(calculator)
enhanced_agent = MySimpleAgent(
name="增强助手",
llm=llm,
system_prompt="你是一个智能助手，可以使用工具来帮助用户。",
tool_registry=tool_registry,
enable_tool_calling=True
)
response2 = enhanced_agent.run("请帮我计算 15 * 8 + 32")
print(f"工具增强响应: {response2}\n")
# 测试3:流式响应
print("=== 测试3:流式响应 ===")
print("流式响应: ", end="")
for chunk in basic_agent.stream_run("请解释什么是人工智能"):
pass # 内容已在stream_run中实时打印
# 测试4:动态添加工具
print("\n=== 测试4:动态工具管理 ===")
print(f"添加工具前: {basic_agent.has_tools()}")
basic_agent.add_tool(calculator)
print(f"添加工具后: {basic_agent.has_tools()}")
print(f"可用工具: {basic_agent.list_tools()}")
# 查看对话历史
print(f"\n对话历史: {len(basic_agent.get_history())} 条消息")
在本节中，我们通过继承 Agent 基类，成功构建了一个功能完备且遵循框架规范的基础对话智能体 MySimpleAgent。它不仅支持基础对话，还具备可选的工具调用能力、流式响应和便利的工具管理方法。
7.4.2 ReActAgent
框架化的 ReActAgent 在保持核心逻辑不变的同时，提升了代码的组织性和可维护性，主要是通过提示词优化和与框架工具系统的集成。
（1）提示词模板的改进
保持了原有的格式要求，强调"每次只能执行一个步骤"，避免混乱，并明确了两种Action的使用场景。
MY_REACT_PROMPT = """你是一个具备推理和行动能力的AI助手。你可以通过思考分析问题，然后调用合适的工具来获取信息，最终给出准确的答案。
## 可用工具
{tools}
## 工作流程
请严格按照以下格式进行回应，每次只能执行一个步骤:
Thought: 分析当前问题，思考需要什么信息或采取什么行动。
Action: 选择一个行动，格式必须是以下之一:
- `{{tool_name}}[{{tool_input}}]` - 调用指定工具
- `Finish[最终答案]` - 当你有足够信息给出最终答案时
## 重要提醒
1. 每次回应必须包含Thought和Action两部分
2. 工具调用的格式必须严格遵循:工具名[参数]
3. 只有当你确信有足够信息回答问题时，才使用Finish
4. 如果工具返回的信息不够，继续使用其他工具或相同工具的不同参数
## 当前任务
**Question:** {question}
## 执行历史
{history}
现在开始你的推理和行动:
"""
（2）重写ReActAgent的完整实现
创建my_react_agent.py文件来重写ReActAgent：
# my_react_agent.py
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_registry
self.max_steps = max_steps
self.current_history: List[str] = []
self.prompt_template = custom_prompt if custom_prompt else MY_REACT_PROMPT
print(f"✅ {name} 初始化完成，最大步数: {max_steps}")
其初始化参数的含义如下：
name： Agent的名称。llm：HelloAgentsLLM的实例，负责与大语言模型通信。tool_registry：ToolRegistry的实例，用于管理和执行Agent可用的工具。system_prompt： 系统提示词，用于设定Agent的角色和行为准则。config： 配置对象，用于传递框架级的设置。max_steps： ReAct循环的最大执行步数，防止无限循环。custom_prompt： 自定义的提示词模板，用于替换默认的ReAct提示词。
框架化的ReActAgent将执行流程分解为清晰的步骤：
def run(self, input_text: str, **kwargs) -> str:
"""运行ReAct Agent"""
self.current_history = []
current_step = 0
print(f"\n🤖 {self.name} 开始处理问题: {input_text}")
while current_step < self.max_steps:
current_step += 1
print(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. 调用LLM
messages = [{"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
通过以上重构，我们将 ReAct 范式成功地集成到了框架中。核心改进在于利用了统一的 ToolRegistry 接口，并通过一个可配置、格式更严谨的提示词模板，提升了智能体执行思考-行动循环的稳定性。对于ReAct的测试案例，由于需要调用工具，所以统一放在文末提供测试代码。
7.4.3 ReflectionAgent
由于这几类Agent已经在第四章实现过核心逻辑，所以这里只给出对应的Prompt。与第四章专门针对代码生成的提示词不同，框架化的版本采用了通用化设计，使其适用于文本生成、分析、创作等多种场景，并通过custom_prompts参数支持用户深度定制。
DEFAULT_PROMPTS = {
"initial": """
请根据以下要求完成任务:
任务: {task}
请提供一个完整、准确的回答。
""",
"reflect": """
请仔细审查以下回答，并找出可能的问题或改进空间:
# 原始任务:
{task}
# 当前回答:
{content}
请分析这个回答的质量，指出不足之处，并提出具体的改进建议。
如果回答已经很好，请回答"无需改进"。
""",
"refine": """
请根据反馈意见改进你的回答:
# 原始任务:
{task}
# 上一轮回答:
{last_attempt}
# 反馈意见:
{feedback}
请提供一个改进后的回答。
"""
}
你可以尝试根据第四章的代码，以及上文ReAct的实现，构建出自己的MyReflectionAgent。下面提供一个测试代码供验证想法。
# test_reflection_agent.py
from dotenv import load_dotenv
from hello_agents import HelloAgentsLLM
from my_reflection_agent import MyReflectionAgent
load_dotenv()
llm = HelloAgentsLLM()
# 使用默认通用提示词
general_agent = MyReflectionAgent(name="我的反思助手", llm=llm)
# 使用自定义代码生成提示词（类似第四章）
code_prompts = {
"initial": "你是Python专家，请编写函数:{task}",
"reflect": "请审查代码的算法效率:\n任务:{task}\n代码:{content}",
"refine": "请根据反馈优化代码:\n任务:{task}\n反馈:{feedback}"
}
code_agent = MyReflectionAgent(
name="我的代码生成助手",
llm=llm,
custom_prompts=code_prompts
)
# 测试使用
result = general_agent.run("写一篇关于人工智能发展历程的简短文章")
print(f"最终结果: {result}")
7.4.4 PlanAndSolveAgent
与第四章自由文本的计划输出不同，框架化版本强制要求Planner以Python列表的格式输出计划，并提供了完整的异常处理机制，确保了后续步骤能够稳定执行。框架化的Plan-and-Solve提示词：
# 默认规划器提示词模板
DEFAULT_PLANNER_PROMPT = """
你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。
请确保计划中的每个步骤都是一个独立的、可执行的子任务，并且严格按照逻辑顺序排列。
你的输出必须是一个Python列表，其中每个元素都是一个描述子任务的字符串。
问题: {question}
请严格按照以下格式输出你的计划:
```python
["步骤1", "步骤2", "步骤3", ...]
```
"""
# 默认执行器提示词模板
DEFAULT_EXECUTOR_PROMPT = """
你是一位顶级的AI执行专家。你的任务是严格按照给定的计划，一步步地解决问题。
你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。
请你专注于解决"当前步骤"，并仅输出该步骤的最终答案，不要输出任何额外的解释或对话。
# 原始问题:
{question}
# 完整计划:
{plan}
# 历史步骤与结果:
{history}
# 当前步骤:
{current_step}
请仅输出针对"当前步骤"的回答:
"""
这一节仍然给出一个综合测试文件test_plan_solve_agent.py，可以自行设计实现。
# test_plan_solve_agent.py
from dotenv import load_dotenv
from hello_agents.core.llm import HelloAgentsLLM
from my_plan_solve_agent import MyPlanAndSolveAgent
# 加载环境变量
load_dotenv()
# 创建LLM实例
llm = HelloAgentsLLM()
# 创建自定义PlanAndSolveAgent
agent = MyPlanAndSolveAgent(
name="我的规划执行助手",
llm=llm
)
# 测试复杂问题
question = "一个水果店周一卖出了15个苹果。周二卖出的苹果数量是周一的两倍。周三卖出的数量比周二少了5个。请问这三天总共卖出了多少个苹果？"
result = agent.run(question)
print(f"\n最终结果: {result}")
# 查看对话历史
print(f"对话历史: {len(agent.get_history())} 条消息")
在最后可以补充一款新的提示词，可以尝试实现custom_prompt载入自定义提示词。
# 创建专门用于数学问题的自定义提示词
math_prompts = {
"planner": """
你是数学问题规划专家。请将数学问题分解为计算步骤:
问题: {question}
输出格式:
python
["计算步骤1", "计算步骤2", "求总和"]
""",
"executor": """
你是数学计算专家。请计算当前步骤:
问题: {question}
计划: {plan}
历史: {history}
当前步骤: {current_step}
请只输出数值结果:
"""
}
# 使用自定义提示词创建数学专用Agent
math_agent = MyPlanAndSolveAgent(
name="数学计算助手",
llm=llm,
custom_prompts=math_prompts
)
# 测试数学问题
math_result = math_agent.run(question)
print(f"数学专用Agent结果: {math_result}")
如表7.2所示，通过这种框架化的重构，我们不仅保持了第四章中各种Agent范式的核心功能，还大幅提升了代码的组织性、可维护性和扩展性。所有Agent现在都共享统一的基础架构，同时保持了各自的特色和优势。
表 7.2 Agent不同章节实现对比
7.4.5 FunctionCallAgent
FunctionCallAgent是hello-agents在0.2.8之后引入的Agent，它基于OpenAI原生函数调用机制的Agent，展示了如何使用OpenAI的函数调用机制来构建Agent。 它支持以下功能：
- _build_tool_schemas:通过工具的description构建OpenAI的function calling schema
- _extract_message_content:从OpenAI的响应中提取文本
- _parse_function_call_arguments:解析模型返回的JSON字符串参数
- _convert_parameter_types:转换参数类型
这些功能可以使其具备原生的OpenAI Function Calling的能力，对比使用prompt约束的方式，具备更强的鲁棒性。
def _invoke_with_tools(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]], tool_choice: Union[str, dict], **kwargs):
"""调用底层OpenAI客户端执行函数调用"""
client = getattr(self.llm, "_client", None)
if client is None:
raise RuntimeError("HelloAgentsLLM 未正确初始化客户端，无法执行函数调用。")
client_kwargs = dict(kwargs)
client_kwargs.setdefault("temperature", self.llm.temperature)
if self.llm.max_tokens is not None:
client_kwargs.setdefault("max_tokens", self.llm.max_tokens)
return client.chat.completions.create(
model=self.llm.model,
messages=messages,
tools=tools,
tool_choice=tool_choice,
**client_kwargs,
)
#内部逻辑是对Openai 原生的functioncall作再封装
#OpenAI 原生functioncall示例
from openai import OpenAI
client = OpenAI()
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
},
"required": ["location"],
},
}
}
]
messages = [{"role": "user", "content": "What's the weather like in Boston today?"}]
completion = client.chat.completions.create(
model="gpt-5",
messages=messages,
tools=tools,
tool_choice="auto"
)
print(completion)