这次我们来看一个很实用的技术主题：不用任何框架，从零开始创建自己的 Agent。对于想要深入理解 Agent 工作原理的开发者来说，跳过框架直接手写实现，是掌握核心机制的最佳路径。
很多人在学习 Agent 时容易陷入"框架依赖症"——总觉得必须用 LangChain、AutoGPT 或者某个热门框架才能开始。但实际上，Agent 的核心逻辑并不复杂，自己实现一遍反而能更清楚地理解任务分解、工具调用、状态管理等关键概念。
本文会带你完成一个完整的 Agent 实现过程，从最基础的任务解析开始，到工具调用、状态管理，最后实现一个能实际工作的简单 Agent。重点不是追求功能多强大，而是让你真正理解每个组件的作用和实现方式。
1. Agent 核心能力速览
在开始编码之前，先明确我们要实现什么样的 Agent：
| 能力项 | 实现目标 |
|---|---|
| 任务理解 | 解析用户输入，识别意图和关键参数 |
| 工具调用 | 根据任务选择合适的工具并执行 |
| 状态管理 | 跟踪任务执行进度和中间结果 |
| 决策逻辑 | 决定下一步操作或何时结束任务 |
| 硬件要求 | 普通开发环境即可，无需特殊硬件 |
| 启动方式 | 命令行或简单 Web 接口 |
| 适合场景 | 学习 Agent 原理、定制简单自动化任务 |
这个自建 Agent 不依赖任何大型模型或复杂框架，完全基于规则和简单逻辑实现，适合作为 Agent 开发的入门项目。
2. Agent 适用场景与使用边界
适合谁使用
- 初学者：想要理解 Agent 底层原理的开发者
- 教育场景：教学演示 Agent 工作机制
- 轻量级需求：需要简单自动化任务但不想引入重型框架
- 定制化开发：有特定业务逻辑需要高度定制
能解决什么问题
- 理解任务分解的基本逻辑
- 掌握工具调用的实现方式
- 学习状态机和决策循环的设计
- 为后续使用框架打下坚实基础
不适合什么场景
- 需要复杂自然语言理解的场景
- 大规模生产环境部署
- 需要连接大量外部 API 的复杂任务
- 需要高级推理能力的需求
技术边界提醒
这个实现主要演示核心概念，实际生产环境中可能需要考虑异常处理、安全验证、性能优化等更多因素。
3. 环境准备与前置条件
基础开发环境
# Python 3.8+ 环境 python --version # 输出应该显示 Python 3.8 或更高版本 # 必要的库（其实基础实现几乎不需要额外依赖） pip list | grep -E "(requests|flask)" # 如果要做 Web 接口才需要 Flask项目结构规划
simple_agent/ ├── agent_core.py # Agent 核心逻辑 ├── tools/ # 工具模块 │ ├── __init__.py │ ├── calculator.py │ └── web_search.py ├── memory.py # 记忆管理 ├── planner.py # 任务规划 └── app.py # 启动入口验证环境就绪
# 简单测试环境 import sys print(f"Python版本: {sys.version}") print("环境检查通过，可以开始开发")4. Agent 核心组件实现
4.1 基础 Agent 类定义
先从最基础的 Agent 类开始，这是整个系统的核心：
class SimpleAgent: def __init__(self, name="SimpleAgent"): self.name = name self.tools = {} # 可用工具字典 self.memory = [] # 执行记忆 self.current_task = None # 当前任务 self.max_steps = 10 # 最大执行步数，防止无限循环 def register_tool(self, tool_name, tool_function): """注册工具到 Agent""" self.tools[tool_name] = tool_function print(f"工具注册成功: {tool_name}") def parse_task(self, user_input): """解析用户输入的任务""" # 简单的关键词匹配解析 task_info = { 'original_input': user_input, 'intent': 'unknown', 'parameters': {}, 'required_tools': [] } # 基础意图识别 user_input_lower = user_input.lower() if '计算' in user_input_lower or '算' in user_input_lower: task_info['intent'] = 'calculation' # 提取数字参数 import re numbers = re.findall(r'\d+', user_input) if numbers: task_info['parameters']['numbers'] = [int(n) for n in numbers] elif '搜索' in user_input_lower or '查找' in user_input_lower: task_info['intent'] = 'search' task_info['parameters']['query'] = user_input return task_info def choose_tool(self, task_info): """根据任务信息选择合适的工具""" intent = task_info['intent'] if intent == 'calculation' and 'calculator' in self.tools: return 'calculator' elif intent == 'search' and 'web_search' in self.tools: return 'web_search' return None def execute_step(self, task_info, tool_name): """执行单个步骤""" if tool_name not in self.tools: return {'status': 'error', 'message': f'工具不存在: {tool_name}'} try: tool_function = self.tools[tool_name] result = tool_function(task_info['parameters']) # 记录执行历史 step_record = { 'step': len(self.memory) + 1, 'tool': tool_name, 'parameters': task_info['parameters'], 'result': result, 'status': 'success' } self.memory.append(step_record) return {'status': 'success', 'result': result} except Exception as e: error_record = { 'step': len(self.memory) + 1, 'tool': tool_name, 'parameters': task_info['parameters'], 'result': str(e), 'status': 'error' } self.memory.append(error_record) return {'status': 'error', 'message': str(e)} def run(self, user_input): """运行 Agent 处理用户输入""" print(f"\n=== Agent 开始处理任务 ===") print(f"输入: {user_input}") # 解析任务 task_info = self.parse_task(user_input) print(f"解析结果: 意图={task_info['intent']}, 参数={task_info['parameters']}") # 选择工具 tool_name = self.choose_tool(task_info) if not tool_name: return {'status': 'error', 'message': '没有找到合适的工具'} print(f"选择工具: {tool_name}") # 执行任务 result = self.execute_step(task_info, tool_name) # 总结执行情况 print(f"执行状态: {result['status']}") if result['status'] == 'success': print(f"执行结果: {result['result']}") print(f"=== Agent 任务完成 ===\n") return result4.2 工具模块实现
工具是 Agent 的能力扩展，这里实现两个基础工具：
# tools/calculator.py def calculator_tool(parameters): """简单的计算器工具""" numbers = parameters.get('numbers', []) if not numbers: return "错误: 没有提供数字参数" if len(numbers) == 1: return f"单个数字: {numbers[0]}" # 简单计算逻辑 total = sum(numbers) average = total / len(numbers) max_num = max(numbers) min_num = min(numbers) result = { '总和': total, '平均值': average, '最大值': max_num, '最小值': min_num, '数字个数': len(numbers) } return result # tools/web_search.py def web_search_tool(parameters): """模拟网页搜索工具""" query = parameters.get('query', '') # 模拟搜索结果 mock_results = [ f"关于 '{query}' 的结果1: 这是模拟结果内容...", f"关于 '{query}' 的结果2: 另一个模拟结果...", f"关于 '{query}' 的结果3: 第三个模拟结果..." ] return { '查询词': query, '结果数量': len(mock_results), '搜索结果': mock_results }4.3 记忆管理模块
# memory.py class MemoryManager: def __init__(self, max_memory_size=100): self.memories = [] self.max_size = max_memory_size def add_memory(self, content, memory_type="execution"): """添加记忆""" memory = { 'id': len(self.memories) + 1, 'content': content, 'type': memory_type, 'timestamp': self._get_timestamp() } self.memories.append(memory) # 限制记忆数量 if len(self.memories) > self.max_size: self.memories.pop(0) return memory['id'] def get_recent_memories(self, count=5): """获取最近的记忆""" return self.memories[-count:] if self.memories else [] def search_memories(self, keyword): """搜索相关记忆""" results = [] for memory in self.memories: if keyword.lower() in str(memory['content']).lower(): results.append(memory) return results def _get_timestamp(self): """获取时间戳""" from datetime import datetime return datetime.now().strftime("%Y-%m-%d %H:%M:%S")5. 完整 Agent 系统集成
现在把各个组件整合成一个完整的系统：
# app.py from agent_core import SimpleAgent from tools.calculator import calculator_tool from tools.web_search import web_search_tool from memory import MemoryManager class EnhancedAgent(SimpleAgent): def __init__(self, name="EnhancedAgent"): super().__init__(name) self.memory_manager = MemoryManager() self.step_count = 0 def run_with_memory(self, user_input): """带记忆的执行方法""" # 先检查是否有相关历史记录 related_memories = self.memory_manager.search_memories(user_input[:10]) if related_memories: print("找到相关历史记录:") for memory in related_memories[-3:]: # 显示最近3条 print(f" - {memory['content']}") # 执行任务 result = self.run(user_input) # 记录到记忆 memory_content = f"用户输入: {user_input}, 结果: {result}" self.memory_manager.add_memory(memory_content) return result def main(): """主函数：演示完整 Agent 工作流程""" # 创建 Agent 实例 agent = EnhancedAgent("我的第一个Agent") # 注册工具 agent.register_tool("calculator", calculator_tool) agent.register_tool("web_search", web_search_tool) print("=== Agent 系统启动 ===") print(f"Agent 名称: {agent.name}") print(f"可用工具: {list(agent.tools.keys())}") # 测试用例 test_cases = [ "计算一下 10, 20, 30 这些数字", "搜索人工智能的发展历史", "帮我算算 5, 15, 25 的平均值", "查找机器学习相关资料" ] for i, test_case in enumerate(test_cases, 1): print(f"\n{'='*50}") print(f"测试用例 {i}: {test_case}") print(f"{'='*50}") result = agent.run_with_memory(test_case) # 显示执行历史 if agent.memory: print("\n最近执行记录:") for record in agent.memory[-2:]: # 显示最近2条记录 print(f" 步骤{record['step']}: {record['tool']} -> {record['status']}") if __name__ == "__main__": main()6. 功能测试与效果验证
6.1 基础功能测试
运行上面的代码，你应该能看到类似这样的输出：
=== Agent 系统启动 === Agent 名称: 我的第一个Agent 可用工具: ['calculator', 'web_search'] ================================================== 测试用例 1: 计算一下 10, 20, 30 这些数字 ================================================== === Agent 开始处理任务 === 输入: 计算一下 10, 20, 30 这些数字 解析结果: 意图=calculation, 参数={'numbers': [10, 20, 30]} 选择工具: calculator 执行状态: success 执行结果: {'总和': 60, '平均值': 20.0, '最大值': 30, '最小值': 10, '数字个数': 3} === Agent 任务完成 ===6.2 验证逻辑正确性
通过测试用例验证 Agent 的各个组件：
# test_agent.py def test_agent_components(): """测试 Agent 各个组件的功能""" agent = SimpleAgent() # 测试任务解析 test_inputs = [ "计算 1 2 3", "搜索天气预报", "未知任务类型" ] for input_text in test_inputs: task_info = agent.parse_task(input_text) print(f"输入: {input_text}") print(f"解析: 意图={task_info['intent']}, 参数={task_info['parameters']}") print("---") # 测试工具注册和选择 agent.register_tool("calculator", calculator_tool) task_info = agent.parse_task("计算 5 10") tool_name = agent.choose_tool(task_info) print(f"选择的工具: {tool_name}") if __name__ == "__main__": test_agent_components()6.3 性能与稳定性观察
虽然这个简单 Agent 不涉及复杂计算，但仍可以观察一些基础指标：
- 启动时间：应该在秒级完成初始化
- 内存占用：基础版本内存占用很少，可以添加监控代码
- 执行速度：单个任务应该在毫秒级完成
import time import psutil import os def monitor_performance(): """监控 Agent 性能""" process = psutil.Process(os.getpid()) start_time = time.time() start_memory = process.memory_info().rss / 1024 / 1024 # MB # 执行测试 agent = EnhancedAgent() agent.register_tool("calculator", calculator_tool) for i in range(100): # 执行100次测试 agent.run(f"计算 {i} {i+1} {i+2}") end_time = time.time() end_memory = process.memory_info().rss / 1024 / 1024 print(f"执行时间: {end_time - start_time:.2f}秒") print(f"内存变化: {end_memory - start_memory:.2f}MB") print(f"平均每次任务时间: {(end_time - start_time) / 100 * 1000:.2f}毫秒")7. 扩展功能实现
7.1 添加 Web 接口
让 Agent 可以通过 HTTP API 调用：
# web_interface.py from flask import Flask, request, jsonify from agent_core import EnhancedAgent from tools.calculator import calculator_tool from tools.web_search import web_search_tool app = Flask(__name__) agent = EnhancedAgent("WebAgent") agent.register_tool("calculator", calculator_tool) agent.register_tool("web_search", web_search_tool) @app.route('/api/agent/query', methods=['POST']) def handle_query(): """处理 Agent 查询请求""" data = request.json user_input = data.get('input', '') if not user_input: return jsonify({'error': '请输入查询内容'}), 400 try: result = agent.run_with_memory(user_input) return jsonify({ 'status': 'success', 'result': result, 'memory_count': len(agent.memory_manager.memories) }) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/agent/status', methods=['GET']) def get_status(): """获取 Agent 状态""" return jsonify({ 'name': agent.name, 'tools': list(agent.tools.keys()), 'memory_size': len(agent.memory_manager.memories), 'total_executions': len(agent.memory) }) if __name__ == '__main__': app.run(host='127.0.0.1', port=5000, debug=True)7.2 批量任务处理
实现简单的批量任务处理能力：
# batch_processor.py import threading from queue import Queue class BatchProcessor: def __init__(self, agent, max_workers=3): self.agent = agent self.task_queue = Queue() self.results = {} self.max_workers = max_workers self.worker_threads = [] def add_task(self, task_id, user_input): """添加任务到队列""" self.task_queue.put((task_id, user_input)) self.results[task_id] = {'status': 'pending'} def _worker(self): """工作线程函数""" while True: try: task_id, user_input = self.task_queue.get(timeout=1) self.results[task_id] = {'status': 'processing'} # 执行任务 result = self.agent.run_with_memory(user_input) self.results[task_id] = { 'status': 'completed', 'result': result, 'completed_at': self.agent.memory_manager._get_timestamp() } self.task_queue.task_done() except: break def start_processing(self): """启动处理线程""" for i in range(self.max_workers): thread = threading.Thread(target=self._worker) thread.daemon = True thread.start() self.worker_threads.append(thread) def wait_completion(self, timeout=None): """等待所有任务完成""" self.task_queue.join() def get_result(self, task_id): """获取任务结果""" return self.results.get(task_id, {'status': 'not_found'}) # 使用示例 def demo_batch_processing(): agent = EnhancedAgent() agent.register_tool("calculator", calculator_tool) processor = BatchProcessor(agent) processor.start_processing() # 添加批量任务 tasks = [ ("task1", "计算 1 2 3"), ("task2", "计算 10 20 30"), ("task3", "计算 100 200 300") ] for task_id, user_input in tasks: processor.add_task(task_id, user_input) # 等待完成 processor.wait_completion() # 查看结果 for task_id, _ in tasks: result = processor.get_result(task_id) print(f"{task_id}: {result['status']}")8. 常见问题与排查方法
在实现和运行过程中可能会遇到以下问题：
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 工具注册失败 | 工具函数不存在或路径错误 | 检查工具模块导入路径 | 确保工具函数正确定义并可以导入 |
| 任务解析错误 | 输入格式不符合预期 | 打印解析中间结果 | 调整解析逻辑或规范输入格式 |
| 内存使用过多 | 记忆管理未清理旧数据 | 监控内存使用情况 | 实现记忆自动清理机制 |
| 执行卡住 | 任务逻辑出现无限循环 | 添加执行步数限制 | 检查任务完成条件判断 |
| Web接口无法访问 | 端口被占用或服务未启动 | 检查端口占用情况 | 更换端口或重启服务 |
8.1 调试技巧
添加详细的日志输出有助于排查问题：
import logging # 配置日志 logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) class DebuggableAgent(SimpleAgent): def __init__(self, name="DebuggableAgent"): super().__init__(name) self.logger = logging.getLogger(name) def run(self, user_input): self.logger.info(f"开始处理任务: {user_input}") task_info = self.parse_task(user_input) self.logger.debug(f"任务解析结果: {task_info}") # ... 其余逻辑添加日志记录8.2 性能优化建议
当 Agent 变得复杂时，可以考虑以下优化：
- 异步执行：对于IO密集型工具使用异步调用
- 结果缓存：对相同输入缓存计算结果
- 记忆压缩：对长期记忆进行摘要和压缩
- 懒加载：工具按需加载，减少启动时间
9. 最佳实践与使用建议
9.1 代码组织建议
advanced_agent/ ├── core/ # 核心组件 │ ├── agent.py │ ├── planner.py │ └── memory.py ├── tools/ # 工具库 │ ├── base_tool.py # 工具基类 │ ├── calculator.py │ └── web_search.py ├── utils/ # 工具函数 │ ├── logger.py │ └── validator.py ├── config/ # 配置文件 │ └── settings.py └── tests/ # 测试用例 ├── test_agent.py └── test_tools.py9.2 工具开发规范
定义工具接口规范，确保一致性：
# tools/base_tool.py from abc import ABC, abstractmethod class BaseTool(ABC): def __init__(self, name, description): self.name = name self.description = description @abstractmethod def execute(self, parameters): """执行工具的主要逻辑""" pass def validate_parameters(self, parameters): """验证输入参数""" return True # 基础验证，子类可重写 # 具体工具实现 class AdvancedCalculatorTool(BaseTool): def __init__(self): super().__init__("calculator", "高级计算器工具") def execute(self, parameters): numbers = parameters.get('numbers', []) operation = parameters.get('operation', 'sum') if operation == 'sum': return sum(numbers) elif operation == 'average': return sum(numbers) / len(numbers) if numbers else 0 # ... 其他操作 def validate_parameters(self, parameters): if 'numbers' not in parameters: return False, "缺少numbers参数" return True, "参数有效"9.3 安全考虑
即使是这样简单的 Agent，也要注意安全：
- 输入验证：对所有用户输入进行验证
- 工具权限：限制工具的执行权限
- 资源限制：防止无限循环或资源耗尽
- 错误处理：妥善处理异常，避免信息泄露
10. 总结与下一步
通过这个从零开始的 Agent 实现，你应该已经掌握了 Agent 的核心工作机制。这个简单版本包含了任务解析、工具调用、状态管理等基本要素，为理解更复杂的 Agent 系统打下了基础。
最值得尝试的扩展方向：
- 集成真实API：将模拟工具替换为真实的计算API、搜索API等
- 添加学习能力：让 Agent 能够从历史执行中学习优化策略
- 实现多步任务：处理需要多个工具协作的复杂任务
- 添加用户界面：开发更友好的Web或桌面界面
最容易踩的坑是过度设计——在初期阶段保持简单，先让核心流程跑通，再逐步添加复杂功能。这个自建 Agent 虽然功能简单，但完整演示了 Agent 技术的核心思想，建议在此基础上继续探索和实践。
