这是关于使用 Python SDK 通过 OpenAI 的 Assistant API 构建代理的多部分系列中的第一部分。 什么是代理？ 在我看来，代理实际上只是一个利用 LLM（大型语言模型）并试图模仿人类行为的软件。这意味着它不仅可以交谈和理解语言，还可以执行对现实世界产生影响的行动。这些操作通常称为工具。 在这篇博文中，我们将探讨如何使用 OpenAI 的 Assistant API 和 Python SDK 构建代理。第 1 部分只是助手的骨架。也就是说，只是对话部分。 我选择构建一个 CLI 应用程序是为了与框架无关。我们将有目的地将我们的实现称为 Agent，并将 OpenAI SDK 实现称为 Assistant，以便轻松区分两者。 当谈到代理能够调用的函数时，我交替使用术语 ”和  。第 2 部分将更详细地介绍函数调用。 “工具 “函数” 先决条件 要学习本教程，您将需要以下内容： 你的机器上安装了Python3  OpenAI API 密钥 Python 编程基础知识 OpenAI 助手概念  ：Assistant API 中的 Assistant 是一个配置为响应用户消息的实体。它使用指令、选定的模型和工具与功能交互并提供答案。 Assistant ：线程代表 Assistant API 中的对话或对话。它是为每个用户交互创建的，可以包含多个消息，充当正在进行的对话的容器。 线程 ：消息是线程中的通信单元。它包含文本（以及将来可能的文件），用于在线程内传达用户查询或助理响应。 消息    ：Run 是处理线程的助手的实例。它涉及读取线程、决定是否调用工具以及根据模型对线程消息的解释生成响应。 Run 设置开发环境 第一步是使用 创建虚拟环境并激活它。这将确保我们的依赖项与系统 Python 安装隔离： venv   python3 -m venv venv source venv/bin/activate 让我们安装唯一的依赖项：  包： openai   pip install openai 创建一个 文件。让我们为 CLI 应用程序填充一些基本的运行时逻辑： main.py   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 方法的 参数仅用于识别 OpenAI 仪表板中的 Assistant，而 AI 此时实际上并不知道它。实际上，您必须将名称传递给我们稍后将看到的  。 name instructions 您可以在创建助手时设置  ，但这实际上会使您的助手对动态变化的灵活性降低。 instructions 您可以通过调用 来更新 Assistant，但是有一个更好的地方可以传递动态值，我们将在运行时看到这些值。 client.beta.assistants.update 请注意，如果您在创建运行时多次传递  ，则助手的 将被运行的 覆盖。它们并不相辅相成，因此请根据您的需求进行选择：静态指令的助理级别或动态指令的运行级别。 instructions instructions instructions 对于模型，我选择了 模型，以便我们可以在本系列的第 2 部分中添加函数调用。如果您想节省几分钱，同时在我们实现工具时让自己感到纯粹的挫败感，您可以使用  。 gpt-4-turbo-preview 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 ) 请注意，目前只能使用角色 添加消息。我相信 OpenAI 计划在未来的版本中改变这一点，因为这是相当有限的。 user 现在，我们可以获取线程中的最后一条消息：   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 还是不太聪明。比霍比特人更接近鹦鹉🦜。在下一部分中，真正的乐趣开始了。 创建并轮询运行 创建运行时，需要定期检索 对象以检查运行的状态。这就是所谓的轮询，这很糟糕。您需要进行民意调查才能确定您的代理下一步应该做什么。 OpenAI 计划添加对流式传输的支持，以使这一切变得更简单。同时，我将在下一节中向您展示如何设置轮询。 Run 请注意以下方法名称上的  ，这是 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 将 方法添加到我们的 Agent 类中： _poll_run   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.")  🥵 唷，这么多……我们来拆包吧。   接收 对象作为参数并提取当前的 Run   。所有可用状态都可以在 OpenAI 中找到。我们将只使用一些适合我们当前目的的。 _poll_run Run status 文档 现在，我们运行 while 循环来检查完成状态，同时处理一些错误情况。 Assistant API 的实际计费有点模糊，因此为了安全起见，我选择在 2 分钟后取消运行。 即使 OpenAI 在 10 分钟后取消运行时也会出现 状态。如果跑步时间超过 2 分钟，那么您可能就遇到了问题。 expired 由于我也不想每隔几毫秒轮询一次，因此我通过仅每 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/

Hire Me

Read My Stories

該音頻是用故事的原始語言製作的！

如何使用 Python 中的 OpenAI 助手构建代理 - 第 1 部分：对话式

About Author

註釋

標籤

这篇文章刊登在

Related Stories

Telegram：加密岛通往大陆的桥梁

比特币 UTXO 模型，为独特的生态系统提供动力

如何将您的工作流程提高 10 倍：17 个必备应用程序

点击赚钱：Telegram 可能会在 Solana 之前吸引下一个 100 亿加密用户

Telegram：加密岛通往大陆的桥梁

比特币 UTXO 模型，为独特的生态系统提供动力

如何将您的工作流程提高 10 倍：17 个必备应用程序

点击赚钱：Telegram 可能会在 Solana 之前吸引下一个 100 亿加密用户

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps