今天我们来聊聊如何用30行代码实现AI Agent的核心功能。如果你对AI Agent开发感兴趣但觉得现有框架过于复杂这篇文章将为你展示一个极简的实现方案。AI Agent的核心能力在于能够理解用户意图、调用工具完成任务并返回结果。虽然市面上的Agent框架功能丰富但学习成本较高。本文实现的30行版本保留了最核心的tool_use和tool_result机制让你快速理解Agent的工作原理。1. 核心能力速览能力项说明项目类型轻量级AI Agent核心实现核心功能工具调用(tool_use)、结果返回(tool_result)、消息处理代码规模约30行Python代码依赖环境Python 3.8无特殊硬件要求启动方式直接运行Python脚本扩展性支持自定义工具函数适合场景学习Agent原理、快速原型验证、教学演示2. Agent核心原理解析AI Agent的核心工作流程可以简化为三个关键步骤意图识别、工具调用、结果返回。从Claude等成熟平台的文档可以看出消息传递包含text、image、tool_use和tool_result四种基本块。在我们的极简实现中重点模拟tool_use和tool_result的交互过程。当用户请求需要工具协助时Agent会生成tool_use块指示要调用的工具执行完成后通过tool_result块返回结果。这种设计模式的优势在于解耦了语言模型的推理能力和具体工具的执行能力让模型专注于思考工具专注于执行。3. 环境准备与依赖安装实现这个极简Agent只需要基础的Python环境无需GPU或特殊硬件。3.1 基础环境要求Python 3.8或更高版本pip包管理工具文本编辑器或IDE3.2 依赖包安装# 创建虚拟环境可选 python -m venv agent_env source agent_env/bin/activate # Linux/Mac # agent_env\Scripts\activate # Windows # 安装必要依赖 pip install requests这个极简实现刻意保持最少的依赖仅使用Python标准库和requests用于可能的网络工具调用演示。4. 30行核心代码实现下面是完整的Agent核心实现代码包含了工具注册、消息处理和结果返回的关键逻辑class SimpleAgent: def __init__(self): self.tools {} def register_tool(self, name, function): 注册工具函数 self.tools[name] function def process_message(self, user_input): 处理用户消息的核心逻辑 # 解析用户意图判断是否需要工具调用 if 天气 in user_input: return self._call_tool(get_weather, user_input) elif 计算 in user_input: return self._call_tool(calculator, user_input) else: return {type: text, content: f收到消息: {user_input}} def _call_tool(self, tool_name, input_text): 执行工具调用 if tool_name in self.tools: result self.tools[tool_name](input_text) return { type: tool_result, tool_name: tool_name, content: result } return {type: text, content: 工具未找到} # 工具函数定义 def weather_tool(query): 模拟天气查询工具 return f查询天气: {query} - 北京晴25℃ def calculator_tool(query): 模拟计算器工具 try: # 提取数字和运算符进行简单计算 if in query: nums query.split() result sum(float(n) for n in nums if n.strip().replace(.,).isdigit()) return f计算结果: {result} except: return 计算失败 # 使用示例 agent SimpleAgent() agent.register_tool(get_weather, weather_tool) agent.register_tool(calculator, calculator_tool) # 测试Agent response agent.process_message(今天北京天气怎么样) print(response)5. 代码结构深度解析5.1 核心类设计SimpleAgent类是整个实现的核心包含三个主要方法__init__: 初始化工具注册表register_tool: 动态注册工具函数process_message: 消息处理入口_call_tool: 内部工具调用方法5.2 工具调用机制工具调用遵循标准的请求-响应模式用户输入经过意图识别匹配到合适的工具函数执行工具并获取结果格式化返回结果5.3 消息格式设计返回消息采用结构化格式包含类型标识和内容便于后续扩展支持多种消息类型text、image、tool_result等。6. 功能测试与效果验证6.1 基础对话测试# 测试普通对话 response agent.process_message(你好) print(response) # 输出: {type: text, content: 收到消息: 你好}6.2 工具调用测试# 测试天气查询工具 response agent.process_message(上海天气如何) print(response) # 输出: {type: tool_result, tool_name: get_weather, content: 查询天气: 上海天气如何 - 北京晴25℃} # 测试计算器工具 response agent.process_message(计算123456) print(response) # 输出: {type: tool_result, tool_name: calculator, content: 计算结果: 579.0}6.3 错误处理测试# 测试未注册工具 response agent.process_message(翻译这句话) print(response) # 输出: {type: text, content: 工具未找到}7. 扩展实现支持真实API调用为了让Agent具备实际应用价值我们可以扩展支持真实的API工具调用import requests import json def real_weather_tool(city): 真实天气API工具示例 # 实际使用时需要替换为真实的天气API api_url fhttps://api.weather.com/{city} try: response requests.get(api_url, timeout10) data response.json() return f{city}天气: {data[weather]}, 温度: {data[temp]}℃ except: return 天气查询服务暂不可用 def web_search_tool(query): 网络搜索工具示例 search_url https://api.search.com/search params {q: query, limit: 3} try: response requests.get(search_url, paramsparams, timeout10) results response.json()[results] return \n.join([f{r[title]}: {r[url]} for r in results]) except: return 搜索服务暂不可用 # 注册真实工具 agent.register_tool(real_weather, real_weather_tool) agent.register_tool(web_search, web_search_tool)8. 高级特性多轮对话支持基础版本支持单轮交互我们可以扩展支持多轮对话上下文class AdvancedAgent(SimpleAgent): def __init__(self): super().__init__() self.conversation_history [] def process_message_with_context(self, user_input): 支持上下文的消息处理 # 将历史对话上下文纳入考虑 context .join([msg[content] for msg in self.conversation_history[-3:]]) full_input f{context} {user_input} if context else user_input response self.process_message(full_input) # 更新对话历史 self.conversation_history.append({role: user, content: user_input}) self.conversation_history.append({role: assistant, content: response}) return response # 使用高级Agent advanced_agent AdvancedAgent() advanced_agent.register_tool(get_weather, weather_tool) # 测试多轮对话 response1 advanced_agent.process_message_with_context(今天天气怎么样) print(第一轮:, response1) response2 advanced_agent.process_message_with_context(那明天呢) print(第二轮:, response2)9. 性能优化与资源管理9.1 工具调用超时控制import signal from contextlib import contextmanager class TimeoutException(Exception): pass contextmanager def time_limit(seconds): def signal_handler(signum, frame): raise TimeoutException(工具执行超时) signal.signal(signal.SIGALRM, signal_handler) signal.alarm(seconds) try: yield finally: signal.alarm(0) def safe_tool_call(tool_func, *args, timeout30): 安全的工具调用带超时控制 try: with time_limit(timeout): return tool_func(*args) except TimeoutException: return 工具执行超时 except Exception as e: return f工具执行错误: {str(e)}9.2 内存使用优化对于长时间运行的Agent服务需要关注内存管理定期清理对话历史使用生成器代替列表处理大量数据实施连接池管理网络请求10. 实际应用场景演示10.1 客服机器人场景def customer_service_agent(): 客服场景定制Agent agent AdvancedAgent() # 注册客服专用工具 agent.register_tool(query_order, lambda x: 订单查询工具) agent.register_tool(refund_request, lambda x: 退款处理工具) agent.register_tool(contact_support, lambda x: 人工客服转接) return agent # 客服对话测试 cs_agent customer_service_agent() response cs_agent.process_message(我想查询订单状态) print(response)10.2 数据分析助手场景def data_analysis_agent(): 数据分析助手Agent agent AdvancedAgent() # 注册数据分析工具 agent.register_tool(data_summary, lambda x: 数据摘要生成) agent.register_tool(chart_generate, lambda x: 图表生成工具) agent.register_tool(trend_analysis, lambda x: 趋势分析工具) return agent11. 与成熟Agent框架对比11.1 优势分析极简设计: 30行代码清晰展示核心原理零学习成本: 无需理解复杂框架概念快速原型: 几分钟即可搭建可用的Agent易于定制: 代码完全可控方便修改扩展11.2 局限性说明功能有限: 缺少高级特性如流式响应、复杂推理等扩展性差: 相比专业框架大规模工具管理较困难缺乏生态: 没有现成的工具库和社区支持12. 常见问题与解决方案12.1 工具注册失败问题现象: 工具函数注册后无法正常调用排查步骤:检查工具名称拼写是否一致验证工具函数签名是否正确确认工具函数是否可正常执行解决方案:# 添加工具注册验证 def register_tool_with_validation(self, name, function): if not callable(function): raise ValueError(f工具 {name} 必须是可调用函数) self.tools[name] function12.2 意图识别不准确问题现象: 用户请求无法正确匹配到合适的工具解决方案: 实现更智能的意图识别算法def enhanced_intent_detection(self, user_input): 增强的意图识别 intent_keywords { weather: [天气, 气象, 温度], calculation: [计算, 算一下, , -, *, /], search: [搜索, 查找, 查询] } for intent, keywords in intent_keywords.items(): if any(keyword in user_input for keyword in keywords): return intent return general12.3 性能瓶颈处理问题现象: 工具执行时间过长影响用户体验解决方案: 实现异步执行和超时控制import asyncio async def async_tool_call(self, tool_name, input_text): 异步工具调用 if tool_name in self.tools: try: result await asyncio.wait_for( self.tools[tool_name](input_text), timeout30.0 ) return { type: tool_result, tool_name: tool_name, content: result } except asyncio.TimeoutError: return {type: text, content: 工具执行超时}13. 扩展开发指南13.1 添加新工具类型要扩展支持新的工具类型只需在工具函数中实现相应逻辑def image_processing_tool(image_path, operation): 图像处理工具示例 from PIL import Image import os if not os.path.exists(image_path): return 图片文件不存在 try: img Image.open(image_path) if operation resize: img img.resize((800, 600)) output_path image_path.replace(., _resized.) img.save(output_path) return f图片已调整大小: {output_path} except Exception as e: return f图像处理失败: {str(e)}13.2 集成外部AI服务可以轻松集成ChatGPT、Claude等大语言模型增强推理能力def llm_assistant_tool(user_query): 集成大语言模型工具 api_key your_api_key # 实际使用需要配置有效的API密钥 headers { Authorization: fBearer {api_key}, Content-Type: application/json } payload { model: gpt-3.5-turbo, messages: [{role: user, content: user_query}] } try: response requests.post( https://api.openai.com/v1/chat/completions, headersheaders, jsonpayload, timeout30 ) result response.json() return result[choices][0][message][content] except Exception as e: return fAI服务调用失败: {str(e)}14. 部署与生产化建议14.1 Web服务封装将Agent封装为HTTP服务方便集成到现有系统from flask import Flask, request, jsonify app Flask(__name__) agent AdvancedAgent() app.route(/chat, methods[POST]) def chat_endpoint(): data request.json user_input data.get(message, ) response agent.process_message(user_input) return jsonify(response) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)14.2 配置化管理使用配置文件管理工具注册和参数设置import yaml def load_agent_config(config_path): 从配置文件加载Agent配置 with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) agent AdvancedAgent() for tool_config in config[tools]: # 动态导入并注册工具 module __import__(tool_config[module]) tool_func getattr(module, tool_config[function]) agent.register_tool(tool_config[name], tool_func) return agent15. 安全与合规考虑15.1 输入验证与过滤def sanitize_input(user_input): 输入清洗与验证 # 移除潜在的危险字符 dangerous_chars [, , script, javascript] for char in dangerous_chars: user_input user_input.replace(char, ) # 限制输入长度 if len(user_input) 1000: user_input user_input[:1000] return user_input15.2 访问控制与权限管理在生产环境中需要实现API密钥认证请求频率限制操作日志记录敏感操作二次确认这个30行代码的Agent核心实现虽然简单但完整展示了AI Agent的工作原理。通过逐步扩展你可以根据实际需求添加更多高级功能。建议从基础版本开始理解核心概念再根据具体应用场景进行定制化开发。
30行Python代码实现AI Agent核心功能：工具调用与结果返回机制详解
今天我们来聊聊如何用30行代码实现AI Agent的核心功能。如果你对AI Agent开发感兴趣但觉得现有框架过于复杂这篇文章将为你展示一个极简的实现方案。AI Agent的核心能力在于能够理解用户意图、调用工具完成任务并返回结果。虽然市面上的Agent框架功能丰富但学习成本较高。本文实现的30行版本保留了最核心的tool_use和tool_result机制让你快速理解Agent的工作原理。1. 核心能力速览能力项说明项目类型轻量级AI Agent核心实现核心功能工具调用(tool_use)、结果返回(tool_result)、消息处理代码规模约30行Python代码依赖环境Python 3.8无特殊硬件要求启动方式直接运行Python脚本扩展性支持自定义工具函数适合场景学习Agent原理、快速原型验证、教学演示2. Agent核心原理解析AI Agent的核心工作流程可以简化为三个关键步骤意图识别、工具调用、结果返回。从Claude等成熟平台的文档可以看出消息传递包含text、image、tool_use和tool_result四种基本块。在我们的极简实现中重点模拟tool_use和tool_result的交互过程。当用户请求需要工具协助时Agent会生成tool_use块指示要调用的工具执行完成后通过tool_result块返回结果。这种设计模式的优势在于解耦了语言模型的推理能力和具体工具的执行能力让模型专注于思考工具专注于执行。3. 环境准备与依赖安装实现这个极简Agent只需要基础的Python环境无需GPU或特殊硬件。3.1 基础环境要求Python 3.8或更高版本pip包管理工具文本编辑器或IDE3.2 依赖包安装# 创建虚拟环境可选 python -m venv agent_env source agent_env/bin/activate # Linux/Mac # agent_env\Scripts\activate # Windows # 安装必要依赖 pip install requests这个极简实现刻意保持最少的依赖仅使用Python标准库和requests用于可能的网络工具调用演示。4. 30行核心代码实现下面是完整的Agent核心实现代码包含了工具注册、消息处理和结果返回的关键逻辑class SimpleAgent: def __init__(self): self.tools {} def register_tool(self, name, function): 注册工具函数 self.tools[name] function def process_message(self, user_input): 处理用户消息的核心逻辑 # 解析用户意图判断是否需要工具调用 if 天气 in user_input: return self._call_tool(get_weather, user_input) elif 计算 in user_input: return self._call_tool(calculator, user_input) else: return {type: text, content: f收到消息: {user_input}} def _call_tool(self, tool_name, input_text): 执行工具调用 if tool_name in self.tools: result self.tools[tool_name](input_text) return { type: tool_result, tool_name: tool_name, content: result } return {type: text, content: 工具未找到} # 工具函数定义 def weather_tool(query): 模拟天气查询工具 return f查询天气: {query} - 北京晴25℃ def calculator_tool(query): 模拟计算器工具 try: # 提取数字和运算符进行简单计算 if in query: nums query.split() result sum(float(n) for n in nums if n.strip().replace(.,).isdigit()) return f计算结果: {result} except: return 计算失败 # 使用示例 agent SimpleAgent() agent.register_tool(get_weather, weather_tool) agent.register_tool(calculator, calculator_tool) # 测试Agent response agent.process_message(今天北京天气怎么样) print(response)5. 代码结构深度解析5.1 核心类设计SimpleAgent类是整个实现的核心包含三个主要方法__init__: 初始化工具注册表register_tool: 动态注册工具函数process_message: 消息处理入口_call_tool: 内部工具调用方法5.2 工具调用机制工具调用遵循标准的请求-响应模式用户输入经过意图识别匹配到合适的工具函数执行工具并获取结果格式化返回结果5.3 消息格式设计返回消息采用结构化格式包含类型标识和内容便于后续扩展支持多种消息类型text、image、tool_result等。6. 功能测试与效果验证6.1 基础对话测试# 测试普通对话 response agent.process_message(你好) print(response) # 输出: {type: text, content: 收到消息: 你好}6.2 工具调用测试# 测试天气查询工具 response agent.process_message(上海天气如何) print(response) # 输出: {type: tool_result, tool_name: get_weather, content: 查询天气: 上海天气如何 - 北京晴25℃} # 测试计算器工具 response agent.process_message(计算123456) print(response) # 输出: {type: tool_result, tool_name: calculator, content: 计算结果: 579.0}6.3 错误处理测试# 测试未注册工具 response agent.process_message(翻译这句话) print(response) # 输出: {type: text, content: 工具未找到}7. 扩展实现支持真实API调用为了让Agent具备实际应用价值我们可以扩展支持真实的API工具调用import requests import json def real_weather_tool(city): 真实天气API工具示例 # 实际使用时需要替换为真实的天气API api_url fhttps://api.weather.com/{city} try: response requests.get(api_url, timeout10) data response.json() return f{city}天气: {data[weather]}, 温度: {data[temp]}℃ except: return 天气查询服务暂不可用 def web_search_tool(query): 网络搜索工具示例 search_url https://api.search.com/search params {q: query, limit: 3} try: response requests.get(search_url, paramsparams, timeout10) results response.json()[results] return \n.join([f{r[title]}: {r[url]} for r in results]) except: return 搜索服务暂不可用 # 注册真实工具 agent.register_tool(real_weather, real_weather_tool) agent.register_tool(web_search, web_search_tool)8. 高级特性多轮对话支持基础版本支持单轮交互我们可以扩展支持多轮对话上下文class AdvancedAgent(SimpleAgent): def __init__(self): super().__init__() self.conversation_history [] def process_message_with_context(self, user_input): 支持上下文的消息处理 # 将历史对话上下文纳入考虑 context .join([msg[content] for msg in self.conversation_history[-3:]]) full_input f{context} {user_input} if context else user_input response self.process_message(full_input) # 更新对话历史 self.conversation_history.append({role: user, content: user_input}) self.conversation_history.append({role: assistant, content: response}) return response # 使用高级Agent advanced_agent AdvancedAgent() advanced_agent.register_tool(get_weather, weather_tool) # 测试多轮对话 response1 advanced_agent.process_message_with_context(今天天气怎么样) print(第一轮:, response1) response2 advanced_agent.process_message_with_context(那明天呢) print(第二轮:, response2)9. 性能优化与资源管理9.1 工具调用超时控制import signal from contextlib import contextmanager class TimeoutException(Exception): pass contextmanager def time_limit(seconds): def signal_handler(signum, frame): raise TimeoutException(工具执行超时) signal.signal(signal.SIGALRM, signal_handler) signal.alarm(seconds) try: yield finally: signal.alarm(0) def safe_tool_call(tool_func, *args, timeout30): 安全的工具调用带超时控制 try: with time_limit(timeout): return tool_func(*args) except TimeoutException: return 工具执行超时 except Exception as e: return f工具执行错误: {str(e)}9.2 内存使用优化对于长时间运行的Agent服务需要关注内存管理定期清理对话历史使用生成器代替列表处理大量数据实施连接池管理网络请求10. 实际应用场景演示10.1 客服机器人场景def customer_service_agent(): 客服场景定制Agent agent AdvancedAgent() # 注册客服专用工具 agent.register_tool(query_order, lambda x: 订单查询工具) agent.register_tool(refund_request, lambda x: 退款处理工具) agent.register_tool(contact_support, lambda x: 人工客服转接) return agent # 客服对话测试 cs_agent customer_service_agent() response cs_agent.process_message(我想查询订单状态) print(response)10.2 数据分析助手场景def data_analysis_agent(): 数据分析助手Agent agent AdvancedAgent() # 注册数据分析工具 agent.register_tool(data_summary, lambda x: 数据摘要生成) agent.register_tool(chart_generate, lambda x: 图表生成工具) agent.register_tool(trend_analysis, lambda x: 趋势分析工具) return agent11. 与成熟Agent框架对比11.1 优势分析极简设计: 30行代码清晰展示核心原理零学习成本: 无需理解复杂框架概念快速原型: 几分钟即可搭建可用的Agent易于定制: 代码完全可控方便修改扩展11.2 局限性说明功能有限: 缺少高级特性如流式响应、复杂推理等扩展性差: 相比专业框架大规模工具管理较困难缺乏生态: 没有现成的工具库和社区支持12. 常见问题与解决方案12.1 工具注册失败问题现象: 工具函数注册后无法正常调用排查步骤:检查工具名称拼写是否一致验证工具函数签名是否正确确认工具函数是否可正常执行解决方案:# 添加工具注册验证 def register_tool_with_validation(self, name, function): if not callable(function): raise ValueError(f工具 {name} 必须是可调用函数) self.tools[name] function12.2 意图识别不准确问题现象: 用户请求无法正确匹配到合适的工具解决方案: 实现更智能的意图识别算法def enhanced_intent_detection(self, user_input): 增强的意图识别 intent_keywords { weather: [天气, 气象, 温度], calculation: [计算, 算一下, , -, *, /], search: [搜索, 查找, 查询] } for intent, keywords in intent_keywords.items(): if any(keyword in user_input for keyword in keywords): return intent return general12.3 性能瓶颈处理问题现象: 工具执行时间过长影响用户体验解决方案: 实现异步执行和超时控制import asyncio async def async_tool_call(self, tool_name, input_text): 异步工具调用 if tool_name in self.tools: try: result await asyncio.wait_for( self.tools[tool_name](input_text), timeout30.0 ) return { type: tool_result, tool_name: tool_name, content: result } except asyncio.TimeoutError: return {type: text, content: 工具执行超时}13. 扩展开发指南13.1 添加新工具类型要扩展支持新的工具类型只需在工具函数中实现相应逻辑def image_processing_tool(image_path, operation): 图像处理工具示例 from PIL import Image import os if not os.path.exists(image_path): return 图片文件不存在 try: img Image.open(image_path) if operation resize: img img.resize((800, 600)) output_path image_path.replace(., _resized.) img.save(output_path) return f图片已调整大小: {output_path} except Exception as e: return f图像处理失败: {str(e)}13.2 集成外部AI服务可以轻松集成ChatGPT、Claude等大语言模型增强推理能力def llm_assistant_tool(user_query): 集成大语言模型工具 api_key your_api_key # 实际使用需要配置有效的API密钥 headers { Authorization: fBearer {api_key}, Content-Type: application/json } payload { model: gpt-3.5-turbo, messages: [{role: user, content: user_query}] } try: response requests.post( https://api.openai.com/v1/chat/completions, headersheaders, jsonpayload, timeout30 ) result response.json() return result[choices][0][message][content] except Exception as e: return fAI服务调用失败: {str(e)}14. 部署与生产化建议14.1 Web服务封装将Agent封装为HTTP服务方便集成到现有系统from flask import Flask, request, jsonify app Flask(__name__) agent AdvancedAgent() app.route(/chat, methods[POST]) def chat_endpoint(): data request.json user_input data.get(message, ) response agent.process_message(user_input) return jsonify(response) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)14.2 配置化管理使用配置文件管理工具注册和参数设置import yaml def load_agent_config(config_path): 从配置文件加载Agent配置 with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) agent AdvancedAgent() for tool_config in config[tools]: # 动态导入并注册工具 module __import__(tool_config[module]) tool_func getattr(module, tool_config[function]) agent.register_tool(tool_config[name], tool_func) return agent15. 安全与合规考虑15.1 输入验证与过滤def sanitize_input(user_input): 输入清洗与验证 # 移除潜在的危险字符 dangerous_chars [, , script, javascript] for char in dangerous_chars: user_input user_input.replace(char, ) # 限制输入长度 if len(user_input) 1000: user_input user_input[:1000] return user_input15.2 访问控制与权限管理在生产环境中需要实现API密钥认证请求频率限制操作日志记录敏感操作二次确认这个30行代码的Agent核心实现虽然简单但完整展示了AI Agent的工作原理。通过逐步扩展你可以根据实际需求添加更多高级功能。建议从基础版本开始理解核心概念再根据具体应用场景进行定制化开发。
