1. 从零构建你的第一个智能Agent
在人工智能领域,Agent(智能体)正变得越来越重要。它们不仅能理解自然语言,还能调用工具完成任务,甚至能自主决策。今天我要分享的是如何用Python构建一个基础但功能完整的SimpleAgent,这是我多年开发经验的结晶。
这个SimpleAgent的核心能力包括:
- 基础对话功能
- 工具调用能力(最多3轮)
- 流式响应支持
- 工具动态管理
- 完整的对话历史记录
下面我会从架构设计到代码实现,一步步带你理解如何构建这样的Agent系统。即使你是AI领域的新手,也能通过这个实例快速上手。
2. SimpleAgent架构解析
2.1 核心类结构设计
我们的SimpleAgent继承自基础Agent类,采用组合而非继承的方式集成工具调用能力:
python复制class MySimpleAgent(SimpleAgent):
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
关键设计考量:
- 松耦合设计:工具注册表(tool_registry)通过依赖注入,方便替换实现
- 灵活启用:工具调用功能可动态开启/关闭(enable_tool_calling)
- 类型安全:全面使用Python类型注解(Type Hints)
提示:在实际项目中,我建议将工具调用功能设计为插件式架构,这样后续扩展新工具类型时不会影响核心逻辑。
2.2 消息处理流程
Agent的核心是消息处理机制,我们的设计采用标准的对话轮次管理:
python复制def run(self, input_text: str, max_tool_iterations: int = 3, **kwargs) -> str:
messages = []
# 系统提示词处理
if self.system_prompt:
messages.append({"role": "system", "content": self.system_prompt})
# 历史对话加载
for msg in self._history:
messages.append({"role": "message", "content": msg})
# 当前用户消息
messages.append({"role": "message", "content": input_text})
# 判断是否启用工具调用
if not self.enable_tool_calling:
response = self.llm.invoke(messages, **kwargs)
self._save_conversation(input_text, response)
return response
# 支持工具调用的处理流程
return self._run_with_tools(messages, input_text, max_tool_iterations, **kwargs)
这个设计有几个精妙之处:
- 消息角色分离:区分system/message/user角色,符合现代AI系统设计规范
- 历史对话管理:自动维护完整的对话上下文
- 条件分支:根据配置自动选择简单响应或工具调用流程
3. 工具调用实现细节
3.1 工具调用协议设计
我们采用了一种简洁而强大的工具调用标记协议:
python复制def _parse_tool_calls(self, text: str) -> list:
pattern = r"\[TOOL_CALL:(?P<tool_name>\w+):(?P<parameters>[^\]]+)\]"
matches = re.finditer(pattern, text)
return [{
'tool_name': m.group('tool_name').strip(),
'parameters': m.group('parameters').strip(),
'original': m.group(0)
} for m in matches]
这个正则表达式模式可以解析如下格式的工具调用:
code复制[TOOL_CALL:search:query=Python]
[TOOL_CALL:calculator:expression=2+2*3]
经验分享:在实际项目中,我建议使用JSON格式作为工具调用协议,虽然正则表达式在这种简单场景下够用,但JSON更易于扩展复杂参数结构。
3.2 多轮工具调用控制
实现安全的多轮工具调用是Agent系统的关键:
python复制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:
response = self.llm.invoke(messages, **kwargs)
tool_calls = self._parse_tool_calls(response)
if 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(f"工具: {call['tool_name']}\n结果: {result}")
clean_response = clean_response.replace(call['original'], "")
if tool_results:
messages.append({"role": "tool", "content": "\n\n".join(tool_results)})
current_iteration += 1
continue
final_response = response if not final_response else final_response
break
self._save_conversation(input_text, final_response)
return final_response
关键安全措施:
- 迭代次数限制:防止无限循环(max_tool_iterations)
- 结果清理:移除原始工具调用标记,保持响应干净
- 上下文维护:将工具执行结果作为新消息加入对话历史
4. 高级功能实现
4.1 流式响应支持
对于需要实时显示的场景,我们实现了流式响应:
python复制def stream_run(self, input_text: str, **kwargs) -> Iterator[str]:
messages = []
# 构建消息上下文(同上)
full_response = ""
for chunk in self.llm.stream_invoke(messages, **kwargs):
full_response += chunk
yield chunk
self._save_conversation(input_text, full_response)
使用示例:
python复制for chunk in agent.stream_run("请介绍Python的优缺点"):
print(chunk, end="")
性能提示:流式响应会带来额外的网络开销,在不需要实时显示的场景下,建议使用普通run方法。
4.2 动态工具管理
我们提供了完整的工具生命周期管理API:
python复制def add_tool(self, tool) -> None:
"""动态添加工具"""
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)
def remove_tool(self, tool_name: str) -> bool:
"""移除工具"""
if self.tool_registry:
return self.tool_registry.unregister(tool_name)
return False
def list_tools(self) -> list:
"""列出所有可用工具"""
return self.tool_registry.list_tools() if self.tool_registry else []
5. 实战技巧与避坑指南
5.1 系统提示词优化
好的系统提示词能显著提升Agent表现:
python复制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
tool_descriptions = "\n".join(
[f"{tool.name}: {tool.description}" for tool in self.tool_registry.get_all_tools()]
)
return f"""{base_prompt}
你可以使用以下工具来帮助用户:
{tool_descriptions}
工具调用格式:
使用工具: <工具名称>
参数: <参数>
重要规则:
1. 每次只能调用一个工具
2. 必须严格按指定格式调用
3. 工具结果返回前不要回复用户"""
5.2 常见问题排查
-
工具调用不被识别
- 检查工具名称是否完全匹配
- 验证调用格式是否符合[TOOL_CALL:name:params]规范
- 确认enable_tool_calling是否为True
-
无限循环问题
- 设置合理的max_tool_iterations(通常3-5次)
- 在工具结果中加入"最终答案"标记
- 添加超时机制
-
上下文混乱
- 定期清理历史对话
- 为不同会话分配唯一ID
- 实现对话总结功能(long-term memory)
6. ReAct Agent扩展实现
除了SimpleAgent,我还实现了更高级的ReActAgent:
python复制class MyReactAgent(ReActAgent):
def __init__(
self,
name: str,
llm: HelloAgentsLLM,
system_prompt: Optional[str] = None,
config: Optional[Config] = None,
tool_registry: Optional['ToolRegistry'] = 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 = []
self.prompt_template = custom_prompt or MY_REACT_PROMPT
ReAct模式的关键优势:
- 更明确的思考过程(Thought/Action/Observation循环)
- 更好的可解释性
- 复杂任务处理能力更强
构建Agent系统最关键的不仅是实现功能,更要考虑:
- 错误处理机制
- 性能监控
- 安全边界控制
- 用户体验优化
经过多个项目的实践验证,这套架构在保持简单的同时,能够满足大多数业务场景的需求。你可以基于这个基础框架,继续扩展如多Agent协作、长期记忆等高级功能。