我用这个Python库一天搭建了个Agent平台!
1今天要跟大家分享一个超级有趣的项目 - 用Python搭建一个AI智能助手平台。这个平台可以让AI自主思考和行动,就像钢铁侠的贾维斯一样!我们会用到OpenAI的API,但不需要复杂的Langchain等第三方库,用最基础的Python代码就能实现。让我们开始动手吧!
2
3## 1. 项目准备工作
4
5首先需要准备以下环境:
6
7- Python 3.7+
8- OpenAI API密钥
9- 代码编辑器(推荐VS Code)
10
11创建一个新的Python虚拟环境并安装必要的包:
12
13```python
14# 创建虚拟环境
15python -m venv agent_env
16source agent_env/bin/activate # Linux/Mac
17# Windows用:.\agent_env\Scripts\activate
18
19# 安装必要的包
20pip install openai python-dotenv
2. 项目结构搭建
我们需要创建3个核心文件:
1project/
2 ├── main.py # 主程序
3 ├── actions.py # 功能函数
4 ├── prompts.py # 提示词模板
5 └── .env # 环境变量
在.env文件中添加你的OpenAI API密钥:
1OPENAI_API_KEY=sk-your-api-key-here
3. 构建核心功能
先在actions.py中定义Agent可以执行的操作:
1def get_website_info(url):
2 “”“模拟获取网站信息的函数”“”
3 info = {
4 “google.com”: {“response_time”: 0.3, “status”: “online”},
5 “github.com”: {“response_time”: 0.5, “status”: “online”}
6 }
7 return info.get(url, {“response_time”: 1.0, “status”: “unknown”})
8
9def search_knowledge(query):
10 “”“模拟知识库搜索”“”
11 knowledge_base = {
12 “python”: “Python是一种高级编程语言”,
13 “openai”: “OpenAI是一家AI研究公司”
14 }
15 return knowledge_base.get(query, “未找到相关信息”)
4. 配置提示词模板
在prompts.py中设置ReAct提示词模板:
1SYSTEM_PROMPT = “”“你是一个智能AI助手,运行在Thought(思考) -> Action(行动) -> Response(响应)的循环中。
2
3可用的操作有:
41. get_website_info: 获取网站信息
52. search_knowledge: 搜索知识库
6
7示例格式:
8Thought: 我需要了解网站状态
9Action: {”name“: ”get_website_info“, ”args“: {”url“: ”google.com“}}
10Response: 根据获得的信息进行回答
11
12每次行动后请说”PAUSE“等待响应。
13”“”
5. 主程序实现
在main.py中实现Agent的核心逻辑:
1from openai import OpenAI
2import json
3import os
4from dotenv import load_dotenv
5from actions import *
6from prompts import SYSTEM_PROMPT
7
8load_dotenv()
9
10client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
11
12def run_conversation(user_input):
13 messages = [
14 {“role”: “system”, “content”:SYSTEM_PROMPT},
15 {“role”: “user”, “content”:user_input}
16 ]
17
18 for _ in range(5): # 最多5轮对话
19 response = client.chat.completions.create(
20 model=“gpt-3.5-turbo”,
21 messages=messages
22 )
23
24 assistant_message = response.choices[0].message.content
25 if “PAUSE” not in assistant_message:
26 return assistant_message
27
28 # 解析Action并执行
29 try:
30 action = json.loads(assistant_message.split(“Action: ”)[1].split(“PAUSE”)[0])
31 if action[“name”] == “get_website_info”:
32 result = get_website_info(**action[“args”])
33 elif action[“name”] == “search_knowledge”:
34 result = search_knowledge(**action[“args”])
35
36 messages.append({“role”: “assistant”, “content”:assistant_message})
37 messages.append({“role”: “user”, “content”:f“Action result: {result}”})
38 except Exception as e:
39 messages.append({“role”: “user”, “content”:f“Error: {str(e)}”})
40
41 return “抱歉,我没能完成这个任务”
42
43# 测试运行
44if __name__ == “__main__”:
45 test_input = “google.com的响应时间是多少?”
46 print(run_conversation(test_input))
实用小贴士
测试时建议先用gpt-3.5-turbo模型,费用更低
可以通过调整system_prompt来控制Agent的行为方式
记得做好错误处理,避免程序崩溃
建议加入日志记录功能,方便调试
进阶优化建议
添加记忆功能,让Agent能记住历史对话
扩展更多action功能,比如文件操作、数据分析等
增加并发处理能力,提高响应速度
添加超时机制,避免死循环
小伙伴们,今天的Python实战就到这里啦!我们用简单的代码就实现了一个基础的AI Agent平台。记得动手实践,遇到问题随时在评论区问我哦。祝大家编码愉快,Python学习节节高!🚀
Python学习 #AI编程 #实战教程# 让我们继续优化Agent平台,添加更多实用功能!
1## 6. 添加记忆模块
2
3为了让我们的Agent能够记住对话历史,我们新建一个memory.py:
4
5```python
6class ConversationMemory:
7 def __init__(self, max_tokens=1000):
8 self.conversations = []
9 self.max_tokens = max_tokens
10
11 def add_memory(self, role, content):
12 self.conversations.append({
13 “role”:role,
14 “content”:content
15 })
16 # 简单的记忆管理,超过最大限制就移除最早的记忆
17 while self._estimate_tokens() > self.max_tokens:
18 self.conversations.pop(0)
19
20 def _estimate_tokens(self):
21 # 粗略估算token数量
22 return sum(len(conv[“content”].split()) * 1.3 for conv in self.conversations)
23
24 def get_relevant_memory(self, query):
25 # 简单的相关性搜索
26 relevant = []
27 for conv in self.conversations:
28 if any(word in conv[“content”].lower()
29 for word in query.lower().split()):
30 relevant.append(conv)
31 return relevant
7. 增强错误处理和日志功能
创建utils.py添加辅助功能:
1import logging
2import time
3from functools import wraps
4
5# 配置日志
6logging.basicConfig(
7 level=logging.INFO,
8 format='%(asctime)s - %(levelname)s - %(message)s',
9 filename='agent.log'
10)
11
12def retry_on_error(max_retries=3, delay=1):
13 “”“错误重试装饰器”“”
14 def decorator(func):
15 @wraps(func)
16 def wrapper(*args, **kwargs):
17 for i in range(max_retries):
18 try:
19 return func(*args, **kwargs)
20 except Exception as e:
21 logging.error(f“尝试 {i+1}/{max_retries} 失败: {str(e)}”)
22 if i < max_retries - 1:
23 time.sleep(delay)
24 raise Exception(f“在{max_retries}次尝试后失败”)
25 return wrapper
26 return decorator
8. 添加更多实用Action
扩展actions.py with更多功能:
1import pandas as pd
2import requests
3from bs4 import BeautifulSoup
4from utils import retry_on_error
5
6class AgentActions:
7 @retry_on_error(max_retries=3)
8 def web_search(self, query):
9 “”“模拟网络搜索”“”
10 # 实际项目中可以接入搜索API
11 return f“搜索结果: {query}”
12
13 def analyze_data(self, data_str):
14 “”“简单的数据分析”“”
15 try:
16 # 将字符串转换为DataFrame
17 data = pd.read_json(data_str)
18 analysis = {
19 “rows”:len(data),
20 “columns”:list(data.columns),
21 “summary”: data.describe().to_dict()
22 }
23 return analysis
24 except Exception as e:
25 return f“数据分析错误: {str(e)}”
26
27 @retry_on_error()
28 def fetch_webpage(self, url):
29 “”“获取网页内容”“”
30 response = requests.get(url)
31 soup = BeautifulSoup(response.text, 'html.parser')
32 return {
33 “title”:soup.title.string if soup.title else “No title”,
34 “text”:soup.get_text()[:500] # 只返回前500字符
35 }
9. 改进主程序
更新main.py使用新功能:
1from memory import ConversationMemory
2from actions import AgentActions
3import logging
4from utils import retry_on_error
5
6class Agent:
7 def __init__(self):
8 self.memory = ConversationMemory()
9 self.actions = AgentActions()
10 self.client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
11
12 @retry_on_error()
13 def get_response(self, user_input):
14 # 获取相关记忆
15 relevant_memory = self.memory.get_relevant_memory(user_input)
16
17 # 构建消息
18 messages = [
19 {“role”: “system”, “content”:SYSTEM_PROMPT},
20 *relevant_memory,
21 {“role”: “user”, “content”:user_input}
22 ]
23
24 try:
25 response = self.client.chat.completions.create(
26 model=“gpt-3.5-turbo”,
27 messages=messages
28 )
29
30 response_content = response.choices[0].message.content
31 self.memory.add_memory(“assistant”, response_content)
32 return response_content
33
34 except Exception as e:
35 logging.error(f“获取响应失败: {str(e)}”)
36 return “抱歉,我遇到了一些问题,请稍后再试”
37
38# 使用示例
39if __name__ == “__main__”:
40 agent = Agent()
41
42 # 测试对话
43 conversations = [
44 “Python是什么编程语言?”,
45 “它有哪些主要特点?”,
46 “给我看一个简单的Python代码示例”
47 ]
48
49 for conv in conversations:
50 print(f“\n用户: {conv}”)
51 response = agent.get_response(conv)
52 print(f“AI: {response}”)
小贴士
内存管理 :
定期清理不相关的记忆
可以使用向量数据库(如Faiss)优化记忆搜索
错误处理 :
为不同类型的错误设置不同的重试策略
添加详细的错误日志便于调试
性能优化 :
使用异步编程处理I/O操作
缓存常用的响应
实战练习
试试实现这些功能:
添加文件处理能力,让Agent能读写文件
接入真实的搜索API
实现对话导出功能
添加简单的Web界面
小伙伴们,到这里我们的Agent平台就基本完善啦!虽然还比不上钢铁侠的贾维斯,但已经是个很棒的开始了。记得关注代码中的注释说明,遇到问题也可以查看日志文件哦。祝大家编码愉快!🚀