这是关于使用 Python SDK 通过 OpenAI 的 Assistant API 构建代理的多部分系列中的第一部分。

什么是代理？

在我看来，代理实际上只是一个利用 LLM（大型语言模型）并试图模仿人类行为的软件。这意味着它不仅可以交谈和理解语言，还可以执行对现实世界产生影响的行动。这些操作通常称为工具。

在这篇博文中，我们将探讨如何使用 OpenAI 的 Assistant API 和 Python SDK 构建代理。第 1 部分只是助手的骨架。也就是说，只是对话部分。

我选择构建一个 CLI 应用程序是为了与框架无关。我们将有目的地将我们的实现称为 Agent，并将 OpenAI SDK 实现称为 Assistant，以便轻松区分两者。

当谈到代理能够调用的函数时，我交替使用术语“工具”和“函数” 。第 2 部分将更详细地介绍函数调用。

先决条件

要学习本教程，您将需要以下内容：

• 你的机器上安装了Python3
• OpenAI API 密钥
• Python 编程基础知识

OpenAI 助手概念

Assistant ：Assistant API 中的 Assistant 是一个配置为响应用户消息的实体。它使用指令、选定的模型和工具与功能交互并提供答案。

线程：线程代表 Assistant API 中的对话或对话。它是为每个用户交互创建的，可以包含多个消息，充当正在进行的对话的容器。

消息：消息是线程中的通信单元。它包含文本（以及将来可能的文件），用于在线程内传达用户查询或助理响应。

Run ：Run 是处理线程的助手的实例。它涉及读取线程、决定是否调用工具以及根据模型对线程消息的解释生成响应。

设置开发环境

第一步是使用venv创建虚拟环境并激活它。这将确保我们的依赖项与系统 Python 安装隔离：

 python3 -m venv venv source venv/bin/activate

让我们安装唯一的依赖项： openai包：

 pip install openai

创建一个main.py文件。让我们为 CLI 应用程序填充一些基本的运行时逻辑：

 while True: user_input = input("User: ") if user_input.lower() == 'exit': print("Exiting the assistant...") break print(f"Assistant: You said {user_input}")

通过运行python3 main.py来尝试一下：

 python3 main.py User: hi Assistant: You said hi

正如您所看到的，CLI 接受用户消息作为输入，而我们的天才助手还没有大脑🧠，所以他只是立即重复该消息。还没那么聪明。

中介

现在，有趣的😁（或头痛的🤕）开始了。我现在将提供最后一堂课所需的所有导入，因此您不必绞尽脑汁地思考事物的来源，因为为了简洁起见，我将导入保留在代码示例之外。让我们首先在新文件agent.py中构建Agent类：

 import time import openai from openai.types.beta.threads.run import Run class Agent: def __init__(self, name: str, personality: str): self.name = name self.personality = personality self.client = openai.OpenAI(api_key="sk-*****") self.assistant = self.client.beta.assistants.create( name=self.name, model="gpt-4-turbo-preview" )

在类构造函数中，我们通过传递 OpenAI API 密钥将 OpenAI 客户端初始化为类属性。接下来，我们创建一个映射到新创建的助手assistant类属性。我们将name和personality存储为类属性以供以后使用。

我们传递给 create 方法的name参数仅用于识别 OpenAI 仪表板中的 Assistant，而 AI 此时实际上并不知道它。实际上，您必须将名称传递给我们稍后将看到的instructions 。

您可以在创建助手时设置instructions ，但这实际上会使您的助手对动态变化的灵活性降低。

您可以通过调用client.beta.assistants.update来更新 Assistant，但是有一个更好的地方可以传递动态值，我们将在运行时看到这些值。

请注意，如果您在创建运行时多次传递instructions ，则助手的instructions将被运行的instructions覆盖。它们并不相辅相成，因此请根据您的需求进行选择：静态指令的助理级别或动态指令的运行级别。

对于模型，我选择了gpt-4-turbo-preview模型，以便我们可以在本系列的第 2 部分中添加函数调用。如果您想节省几分钱，同时在我们实现工具时让自己感到纯粹的挫败感，您可以使用gpt-3.5-turbo 。

GPT 3.5在调用工具方面很糟糕；我在处理这个问题上花费的时间让我可以这么说。 😝 我就先到此为止，稍后再详细介绍。

创建线程、添加消息和检索最后一条消息

创建代理后，我们需要启动对话线程。

 class Agent: # ... (rest of code) def create_thread(self): self.thread = self.client.beta.threads.create()

我们需要一种向该线程添加消息的方法：

 class Agent: # ... (rest of code) def add_message(self, message): self.client.beta.threads.messages.create( thread_id=self.thread.id, role="user", content=message )

请注意，目前只能使用角色user添加消息。我相信 OpenAI 计划在未来的版本中改变这一点，因为这是相当有限的。

现在，我们可以获取线程中的最后一条消息：

 class Agent: # ... (rest of code) def get_last_message(self): return self.client.beta.threads.messages.list( thread_id=self.thread.id ).data[0].content[0].text.value

接下来，我们创建一个入口点run_agent方法来测试到目前为止我们所拥有的内容。目前， run_agent方法仅返回线程中的最后一条消息。它实际上并不执行运行。还是没脑子。

 class Agent: # ... (rest of code) def run_agent(self): message = self.get_last_message() return message

回到main.py ，我们创建代理和第一个线程。我们向线程添加一条消息。然后将相同的消息返回给用户，但这一次，来自该活动线程。

 from agent import Agent agent = Agent(name="Bilbo Baggins", personality="You are the accomplished and renowned adventurer from The Hobbit. You act like you are a bit of a homebody, but you are always up for an adventure. You worry a bit too much about breakfast.") agent.create_thread() while True: user_input = input("User: ") if user_input.lower() == 'exit': print("Exiting the agent...") break agent.add_message(user_input) answer = agent.run_agent() print(f"Assistant: {answer}")

让我们运行一下：

 python3 main.py User: hi Assistant: hi

还是不太聪明。比霍比特人更接近鹦鹉🦜。在下一部分中，真正的乐趣开始了。

创建并轮询运行

创建运行时，需要定期检索Run对象以检查运行的状态。这就是所谓的轮询，这很糟糕。您需要进行民意调查才能确定您的代理下一步应该做什么。 OpenAI 计划添加对流式传输的支持，以使这一切变得更简单。同时，我将在下一节中向您展示如何设置轮询。

请注意以下方法名称上的_ ，这是 Python 中的标准，表示该方法仅供内部使用，不应由外部代码直接访问。

首先，我们创建一个辅助方法_create_run来创建Run ，并更新run_agent来调用此方法：

 class Agent: # ... (rest of code) def get_breakfast_count_from_db(self): return 1 def _create_run(self): count = self.get_breakfast_count_from_db() return self.client.beta.threads.runs.create( thread_id=self.thread.id, assistant_id=self.assistant.id, instructions=f""" Your name is: {self.name} Your personality is: {self.personality} Metadata related to this conversation: {{ "breakfast_count": {count} }} """, ) def run_agent(self): run = self._create_run() # add this line message = self.get_last_message() return message

请注意我们如何传递thread.id和assistant.id来创建运行。

还记得我一开始说过有更好的地方来传递动态指令和数据吗？这将是创建运行时的instructions参数。在我们的例子中，我们可以从数据库中获取早餐count 。这将允许您在每次想要触发答案时轻松地传递不同的相关动态数据。

现在，您的代理意识到周围的世界正在发生变化，并可以采取相应的行动。我喜欢在我的指令中包含一个元数据 JSON 对象来保留相关的动态上下文。这使我能够以法学硕士非常容易理解的格式传递数据，同时不那么冗长。

还不要运行这个；它不起作用，因为当我们收到最后一条消息时，我们没有等待运行完成，因此它仍然是最后一条用户消息。

让我们通过构建轮询机制来解决这个问题。首先，我们需要一种方法来重复且轻松地检索运行，因此让我们添加一个_retrieve_run方法：

 class Agent: # ... (rest of code) def _retrieve_run(self, run: Run): return self.client.beta.threads.runs.retrieve( run_id=run.id, thread_id=self.thread.id)

请注意我们如何需要传递run.id和thread.id来查找特定的运行。

将_poll_run方法添加到我们的 Agent 类中：

 class Agent: # ... (rest of code) def _cancel_run(self, run: Run): self.client.beta.threads.runs.cancel( run_id=run.id, thread_id=self.thread.id) def _poll_run(self, run: Run): status = run.status start_time = time.time() while status != "completed": if status == 'failed': raise Exception(f"Run failed with error: {run.last_error}") if status == 'expired': raise Exception("Run expired.") time.sleep(1) run = self._retrieve_run(run) status = run.status elapsed_time = time.time() - start_time if elapsed_time > 120: # 2 minutes self._cancel_run(run) raise Exception("Run took longer than 2 minutes.")

🥵 唷，这么多……我们来拆包吧。

_poll_run接收Run对象作为参数并提取当前的 Run status 。所有可用状态都可以在 OpenAI文档中找到。我们将只使用一些适合我们当前目的的。

现在，我们运行 while 循环来检查完成状态，同时处理一些错误情况。 Assistant API 的实际计费有点模糊，因此为了安全起见，我选择在 2 分钟后取消运行。

即使 OpenAI 在 10 分钟后取消运行时也会出现expired状态。如果跑步时间超过 2 分钟，那么您可能就遇到了问题。

由于我也不想每隔几毫秒轮询一次，因此我通过仅每 1 秒轮询一次来限制我的请求，直到达到 2 分钟标记并取消运行。您可以将其调整为您认为合适的任何内容。

延迟后的每次迭代，我们都会再次获取运行状态。

现在，让我们将所有这些插入到run_agent方法中。您会注意到，我们首先使用_create_run创建运行，然后使用_poll_run进行轮询，直到得到答案或抛出错误，最后当轮询完成时，我们从线程中检索最后一条消息，该消息现在来自代理。

然后，我们将消息返回到运行时循环，以便将其发送回用户。

 class Agent: # ... (rest of code) def run_agent(self): run = self._create_run() self._poll_run(run) # add this line message = self.get_last_message() return message

瞧，现在当您再次运行代理时，您将收到我们友好的代理的回复：

 python3 main.py User: hi Assistant: Hello there! What adventure can we embark on today? Or perhaps, before we set out, we should think about breakfast. Have you had yours yet? I've had mine, of course – can't start the day without a proper breakfast, you know. User: how many breakfasts have you had? Assistant: Ah, well, I've had just 1 breakfast today. But the day is still young, and there's always room for a second, isn't there? What about you? How can I assist you on this fine day?

在第 2 部分中，我们将为代理添加调用工具的功能。

您可以在我的GitHub上找到完整的代码。

感谢您的阅读。很高兴听到评论中的任何想法和反馈。在 Linkedin 上关注我，了解更多类似内容： https://www.linkedin.com/in/jean-marie-dalmasso-1b5473141/