Vanilla AI代理实现
AI 代理就像超级智能助手,能够自主做出决策以达成目标。虽然人类设定目标,但 AI 会找出最佳实现路径,无需持续监督。例如,一个智能旅行规划 AI 可以建议目的地、查找最佳航班和酒店交易、根据你的偏好创建行程,并在发生取消时重新预订航班——这一切都不需要你动手。如果遇到障碍,比如不寻常的要求,它知道何时求助于人类。
如果你对 AI 感兴趣,并想深入了解一篇关于 Gemini vs. OpenAI Deep Research 的精彩比较文章,最近的 多伦多 AI 二月聚会 对此进行了深入讨论,重点介绍了这些模型如何使用代理 AI 在幕后解决问题。点击此处观看录像。
让我们回到它们是如何构建的。虽然像 PhiData、AutoGen、CrewAI、LangGraph、BeeAgent 和 Swarm 这样的框架提供了强大的捷径,但没有什么能比从头开始构建自己的 AI 代理更能带来兴奋感和深刻理解。
从零开始不仅仅是制造,更是发现可能性。
AI 代理的五大核心支柱:
- 感知 —— AI 代理如何处理和解释其环境中的输入。
- 推理 —— 逻辑处理和决策能力。
- 记忆 —— 存储和检索信息及经验。
- 行动 —— 根据决策采取行动的能力。
- 学习 —— 随时间适应和改进。
注意:本文档的目标不是创建终极代理,而是通过一个简单的、纯 Vanilla 的代理来说明核心概念。我已经尽量简化了解释,使其易于大众理解。可以根据具体用例添加复杂性。一旦理解了基本原理,我们可以利用任何框架来扩展它们。
这个 ItineraryAgent 提供了一种简单的方法来自动生成代码并执行,特别适用于生成旅行行程、可视化或其他模板任务,而无需大量手动干预。
1、导入语句和环境设置
from openai import OpenAI
import os
from google.colab import userdata
import base64
import requests
from PIL import Image
from io import BytesIO
import subprocess
import tempfile
import re
import importlib
import sys
import matplotlib.pyplot as plt
import matplotlib.image as mpimg
- OpenAI:提供与 OpenAI API(用于代码生成)的接口。
- os, sys:用于环境操作,如设置环境变量和处理文件路径。
- google.colab.userdata, base64, requests:分别帮助数据检索、编码/解码和进行 HTTP 请求的工具。
- PIL(Python 图像库):帮助图像处理和显示。
- tempfile:用于创建和管理生成代码的临时文件。
- re:用于正则表达式(如提取代码段或搜索文件模式)。
- importlib:用于动态导入模块(尽管在这里使用较少)。
- subprocess:允许脚本从 Python 中运行命令和其他 Python 文件。
- matplotlib:提供图像显示和绘图功能。
2、ItineraryAgent 类
该类封装了以下逻辑:
- 基于用户提示生成 Python 代码。
- 自动安装必要的库。
- 在临时文件中安全地执行生成的代码。
- 显示任何生成的图像(或其他输出文件)。
方法概述:
generate_code(prompt)
install_libraries(code)
execute_code(code)
display_image(filename)
run(prompt)
以下是每个方法的详细描述。
2.1 generate_code(self, prompt)
def generate_code(self, prompt):
from openai import OpenAI # 确保正确导入 OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "You are a Python code generator. Respond only with executable Python code, ..."},
{"role": "user", "content": f"Generate Python code to {prompt}. ..."}
],
max_tokens=4000,
temperature=0.7,
top_p=1,
frequency_penalty=0,
presence_penalty=0
)
code = re.sub(r'^```python\n|^```\n|```$', '', response.choices[0].message.content, flags=re.MULTILINE)
code_lines = code.split('\n')
while code_lines and not (code_lines[0].startswith('import') or code_lines[0].startswith('from') or code_lines[0].strip().startswith('#')):
code_lines.pop(0)
return '\n'.join(code_lines)
它的作用是:
- 创建一个 客户端 来与 OpenAI 库交互。
- 向 OpenAI 的聊天完成端点发送 提示,带有特定指令:
- “系统”角色指示模型生成 可执行的 Python 代码。
- “用户”角色传递实际提示(例如:“生成创建行程的 Python 代码…”)。
- 接收模型的响应,这应该是一个被三重反引号(```)包围的 Python 代码。
- 使用 正则表达式 删除任何 markdown 格式(三重反引号或其他多余文本)。
- 删除开头的非
import
语句、from
语句或注释行。这一步主要是修剪不需要的文本或免责声明,以便剩下的大部分是直接的 Python 代码。
该方法仅返回由模型生成的精炼代码字符串。
2.2 install_libraries(self, code)
def install_libraries(self, code):
import pkg_resources
...
# 从 import 语句中提取库名
imports = re.findall(r'^\s*(?:import|from) ([\w\d_]+)', code, re.MULTILINE)
# 从 pip install 注释中提取库名
libraries = re.findall(r'#\s*pip install\s+([\w-]+)', code)
all_libraries = set(imports + libraries)
...
if all_libraries:
print("Checking required libraries...")
for lib in all_libraries:
package_name = lib.replace('-', '_')
try:
pkg_resources.require(package_name)
print(f"{lib} is already installed.")
except pkg_resources.DistributionNotFound:
print(f"Installing {lib}...")
subprocess.check_call([sys.executable, "-m", "pip", "install", lib])
print("All required libraries are installed.")
关键步骤:
- 提取库名:
- 读取代码中的
import ...
或from ... import ...
行以猜测可能需要的库。 - 查找带有
# pip install ...
的行,查看用户(或 AI)是否已明确注释了 pip 安装。
2. 检查库是否已安装:
- 使用
pkg_resources.require
查看每个库是否存在于环境中。
3. 安装缺失的库:
- 运行子进程 pip 命令(
subprocess.check_call([...])
)以安装未找到的任何内容。
这种自动依赖安装有助于确保 AI 生成的任何代码都可以在没有手动库设置的情况下运行。
2.3 execute_code(self, code)
def execute_code(self, code):
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as temp_file:
temp_file.write(code)
temp_file_path = temp_file.name
try:
result = subprocess.run([sys.executable, temp_file_path],
capture_output=True, text=True, timeout=50)
output = result.stdout
error = result.stderr
except subprocess.TimeoutExpired:
output = ""
error = "Execution timed out after 50 seconds."
finally:
os.unlink(temp_file_path)
return output, error
它的作用是:
- 创建一个 临时文件(.py),并将生成的代码写入其中。
- 使用
subprocess.run
运行该临时文件:
- capture_output 用于捕获 stdout/stderr。
- timeout 防止脚本无限期运行(50 秒)。
- 返回脚本的 stdout 和 stderr。
最后,删除临时文件以保持环境整洁。
2.4 display_image(self, filename)
def display_image(self, filename):
try:
img = mpimg.imread(filename)
plt.imshow(img)
plt.axis('off')
plt.show()
except Exception as e:
print(f"Could not display image: {e}")
它的作用是:
- 尝试使用 Matplotlib 加载并显示图像文件(例如 PNG 或 JPG)。
- 如果发生错误(例如文件未找到、格式不支持),则打印错误消息。
这是一个方便的方法,如果生成的代码产生或期望显示图像(如流程图或行程图)。
2.5 run(self, prompt)
def run(self, prompt):
print(f"Generating code for: {prompt}")
code = self.generate_code(prompt)
print("Generated code:")
print(code)
print("\nInstalling required libraries if needed...")
self.install_libraries(code)
print("\nExecuting code...")
output, error = self.execute_code(code)
if output:
print("Output:")
print(output)
if error:
print("Error:")
print(error)
高级过程:
- 提示 用户或开发人员某些功能(例如:“从城市 A 到城市 B 创建行程”)。
- 生成 代码,使用 OpenAI 的 API 调用
generate_code
。 - 安装 代码中推断出的任何库。
- 执行 生成的代码,捕获其输出或错误。
3、if __name__ == "__main__": 块
此最终块是一种常见的 Python 模式,表示 “仅当此文件作为脚本执行而不是作为模块导入时才运行此代码。”
if __name__ == "__main__":
agent = ItineraryAgent()
agent.run("Create an itinerary outlining all the essential details for a trip from Orangeville to Niagara Falls, ...")
它的作用是:
- 实例化一个
ItineraryAgent
。 - 调用
agent.run(...)
并传递特定提示,指示代理生成什么代码。 - 这触发整个流程:生成代码、安装库并执行它。
4、输出
看看代理生成的代码的输出。
生成代码用于:创建一个行程图,概述从 Orangeville 到尼亚加拉瀑布的所有必要细节,包括沿途所有可能的停留点。确保重要考虑事项被记录,并将最终文档保存为 PDF。 生成的代码:
# 需要的 pip 安装:
# pip install reportlab
from reportlab.lib.pagesizes import letter
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer
from reportlab.lib.styles import getSampleStyleSheet
def create_itinerary():
doc = SimpleDocTemplate("itinerary.pdf", pagesize=letter)
styles = getSampleStyleSheet()
title = Paragraph("Trip Itinerary: Orangeville to Niagara Falls", styles['Title'])
intro = Paragraph("This itinerary outlines the essential details and possible stops along the trip from Orangeville to Niagara Falls.", styles['Normal'])
stops = [
"1. Orangeville",
"2. Belfountain Conservation Area - Scenic nature spot",
"3. Forks of the Credit Provincial Park - Hiking trails",
"4. Terra Cotta Conservation Area - Nature area with trails",
"5. Royal Botanical Gardens, Burlington - Botanical museum",
"6. Hamilton - Known for its waterfalls",
"7. St. Catharines - Known for its wineries",
"8. Niagara-on-the-Lake - Historic town with shops and restaurants",
"9. Niagara Falls - Main destination"
]
considerations = [
"Check weather conditions before departure.",
"Ensure vehicle is fueled and in good condition.",
"Plan for rest stops and meals.",
"Verify opening hours of parks and attractions.",
"Consider purchasing tickets in advance for popular attractions.",
"Take into account possible traffic delays."
]
elements = [title, Spacer(1, 12), intro, Spacer(1, 12)]
elements.append(Paragraph("Possible Stops:", styles['Heading2']))
for stop in stops:
elements.append(Paragraph(stop, styles['Normal']))
elements.append(Spacer(1, 12))
elements.append(Paragraph("Important Considerations:", styles['Heading2']))
for consideration in considerations:
elements.append(Paragraph(consideration, styles['Normal']))
doc.build(elements)
create_itinerary()
这段代码会默默地为你创建 PDF 文档。
看看生成的 PDF 文档的截图。
如果你要求它以图形格式生成输出,它会像这样生成。(根据输入类型不同)
你可以玩一下这段代码。这里是 notebook 的链接。
在下一篇文章中,我们将探讨 ReAct 和 CoAct 代理,深入研究它们的基本概念、优点和缺点。我们还将讨论何时使用每个代理以及更多内容。敬请期待!
原文链接:A Vanilla AI Agent from Scratch
汇智网翻译整理,转载请标明出处