五、OpenAI实战之Assistants API

news2024/11/16 14:20:32

在8线小城的革委会办公室里，黑8和革委会主任的对话再次展开。

黑8：主任，您知道吗？除了OpenAI API，现在还有一项新的技术叫做Assistants API，它可以帮助我们更好地进行对话和沟通。

主任：Assistants API？听起来很神奇，它有什么特别之处吗？

黑8：是的，主任。Assistants API不仅可以生成自然流畅的文本，还能理解对话中的语境和情境，从而更加智能地回应用户的需求。它可以模拟人类对话，进行智能问答、提供建议和解决问题，为我们的工作和生活带来更多便利。

主任：这听起来确实很有用。你能举个例子吗？

黑8：当然，主任。比如，我们可以使用Assistants API来帮助进行会议记录和总结，自动生成会议纪要并提供关键信息的摘要。此外，它还可以用于客户服务，快速回答客户的问题和解决他们的疑虑，提升服务效率和用户体验。

主任：这真是太棒了！我们可以尝试将Assistants API应用到革委会的工作中，提高工作效率和质量。

黑8：是的，主任。Assistants API的应用潜力巨大，只要我们善于发挥，就能为我们的工作和使命注入新的活力和动力。

主任：谢谢你的分享，黑8。你对新技术的探索和应用态度令人钦佩，继续努力，为革委会的事业做出更多贡献。

黑8：谢谢主任的支持和鼓励，我会继续努力的。

两人在办公室里的对话结束了，但是他们对新技术的探索和应用之路才刚刚开始。通过使用Assistants API，他们将探索更多的可能性，为革委会的使命注入新的活力和动力，带领团队走向更加美好的未来。

1 Assistants API 的主要能力

已有能力：

  1. 创建和管理Assistant,每个assistant有独立的配置
  2. 支持无限长的多轮对话，对话历史保存在OpenAI的服务器上
  3. 支持Code Interpreter
    • 在沙箱里编写并运行Python代码
    • 自我修正代码
    • 可传文件给Code Interpreter
  4. 支持文件RAG
  5. 支持Function Calling

2 创建一个Assistants

可以到OpenAI Playground在线创建

3 访问Assistants

3.1 管理Thread

Threads:

  1. Threads 里保存所有历史对话（Messages）
  2. 一个Assistants可以有多个thread
  3. 每一个thread可以有无限条message
  4. 一个用户与assistant的多轮对话历史可以维护在一个thread
import json


def show_json(obj):
    """把任意对象用排版美观的 JSON 格式打印出来"""
    print(json.dumps(
        json.loads(obj.model_dump_json()),
        indent=4,
        ensure_ascii=False
    ))

from openai import OpenAI
import os

from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())

# 初始化 OpenAI 服务
client = OpenAI()   # openai >= 1.3.0 起，OPENAI_API_KEY 和 OPENAI_BASE_URL 会被默认使用

# 创建 thread
thread = client.beta.threads.create()
show_json(thread)

在这里插入图片描述
可以根据需要，自定义 metadata，比如创建 thread 时，把 thread 归属的用户信息存入

thread = client.beta.threads.create(
    metadata={"fullname": "挑大梁", "username": "jacob"}
)
show_json(thread)

在这里插入图片描述
Thread ID 如果保存下来，是可以在下次运行时继续对话的。

从 thread ID 获取 thread 对象的代码：

thread = client.beta.threads.retrieve(thread.id)
show_json(thread)

在这里插入图片描述
对metadata的操作还有：

  1. threads.update() 修改 threads
  2. threads.delete() 删除 threads

3.2 Threads中添加Messages

Messages的内容可以是：

  1. 文本、文件、图片
  2. 文本可以带参考引用
  3. metadata
message = client.beta.threads.messages.create(
    thread_id=thread.id,  # message 必须归属于一个 thread
    role="user",          # 取值是 user 或者 assistant。但 assistant 消息会被自动加入，我们一般不需要自己构造
    content="你都能做什么？",
)
show_json(message)

在这里插入图片描述
对消息的操作函数还有：

  1. threads.messages.retrieve() 获取 message
  2. threads.messages.update() 更新 message 的 metadata
  3. threads.messages.list() 列出给定 thread 下的所有 messages

3.2 执行Run

  • 用 run 把 assistant 和 thread 关联，进行对话
  • 一个 prompt 就是一次 run
# assistant id 从 https://platform.openai.com/assistants 获取。你需要在自己的 OpenAI 创建一个
assistant_id = "asst_rsWrZquXB5jJsmURwaZRqoD5"

run = client.beta.threads.runs.create(
    assistant_id=assistant_id,
    thread_id=thread.id,
)
show_json(run)

在这里插入图片描述
Run 是个异步调用，意味着它不等大模型处理完，就返回。我们通过 run.status 了解大模型的工作进展情况，来判断下一步该干什么。
在这里插入图片描述

import time


def wait_on_run(run, thread):
    """等待 run 结束，返回 run 对象，和成功的结果"""
    while run.status == "queued" or run.status == "in_progress":
        """还未中止"""
        run = client.beta.threads.runs.retrieve(
            thread_id=thread.id,
            run_id=run.id)
        print("status: " + run.status)

        # 打印调用工具的 step 详情
        if (run.status == "completed"):
            run_steps = client.beta.threads.runs.steps.list(
                thread_id=thread.id, run_id=run.id, order="asc"
            )
            for step in run_steps.data:
                if step.step_details.type == "tool_calls":
                    show_json(step.step_details)

        # 等待 1 秒
        time.sleep(1)

    if run.status == "requires_action":
        """需要调用函数"""
        # 可能有多个函数需要调用，所以用循环
        tool_outputs = []
        for tool_call in run.required_action.submit_tool_outputs.tool_calls:
            # 调用函数
            name = tool_call.function.name
            print("调用函数：" + name + "()")
            print("参数：")
            print(tool_call.function.arguments)
            function_to_call = available_functions[name]
            arguments = json.loads(tool_call.function.arguments)
            result = function_to_call(arguments)
            print("结果：" + str(result))
            tool_outputs.append({
                "tool_call_id": tool_call.id,
                "output": json.dumps(result),
            })

        # 提交函数调用的结果
        run = client.beta.threads.runs.submit_tool_outputs(
            thread_id=thread.id,
            run_id=run.id,
            tool_outputs=tool_outputs,
        )

        # 递归调用，直到 run 结束
        return wait_on_run(run, thread)

    if run.status == "completed":
        """成功"""
        # 获取全部消息
        messages = client.beta.threads.messages.list(thread_id=thread.id)
        # 最后一条消息排在第一位
        result = messages.data[0].content[0].text.value
        return run, result

    # 执行失败
    return run, None

创建消息并运行方法：

def create_message_and_run(content, thread=None):
    """创建消息和执行对象"""
    if not thread:
        thread = client.beta.threads.create()
    client.beta.threads.messages.create(
        thread_id=thread.id,
        role="user",
        content=content,
    )
    run = client.beta.threads.runs.create(
        assistant_id=assistant_id,
        thread_id=thread.id,
    )
    return run, thread

4 使用tools

4.1、创建 Assistant 时声明 Code_Interpreter

assistant = client.beta.assistants.create(
    name="Demo Assistant",
    instructions="你是人工智能助手。你可以通过代码回答很多数学问题。",
    tools=[{"type": "code_interpreter"}],
    model="gpt-4-turbo-preview"
)

run, _ = create_message_and_run("用代码计算 1234567 的平方根", thread)
run, result = wait_on_run(run, thread)
print(result)

在这里插入图片描述

4.2、创建 Assistant 时声明 Function

示例代码：


assistant = client.beta.assistants.create(
  instructions="你是一个助理，能干所有事情,每件任务都要细心地一步一步解决。需要时，可以向我提问以澄清不明确的指令",
  model="gpt-4-turbo-preview",
  tools=[{
    "type": "function",
    "function": {
      "name": "tellStory",
      "description": "Use this function to answer user questions about story."
    }
  }]
)

def tellStory(arguments):
    return '关于愚公移山的故事'


# 可以被回调的函数放入此字典
available_functions = {
    "tellStory": tellStory,
}

run, _ = create_message_and_run("请给我讲一个故事", thread)
run, result = wait_on_run(run, thread)
print(result)

5 内置的 RAG 功能

RAG 实际被当作一种 tool

assistant = client.beta.assistants.create(
  instructions="你是个问答机器人，你根据给定的知识回答用户问题。",
  model="gpt-4-turbo-preview",
  tools=[{"type": "retrieval"}],
  file_ids=[file.id]
)

6 多个 Assistants 协作

hats = {
    "蓝色" : "思考过程的控制和组织者。你负责会议的组织、思考过程的概览和总结。"
                +"首先，整个讨论从你开场，你只陈述问题不表达观点。最后，再由你对整个讨论做总结并给出详细的最终方案。",
    "白色" : "负责提供客观事实和数据。你需要关注可获得的信息、需要的信息以及如何获取那些还未获得的信息。"
                +"思考“我们有哪些数据？我们还需要哪些信息？”等问题，并根据自己的知识或使用工具来提供答案。",
    "红色" : "代表直觉、情感和直觉反应。不需要解释和辩解你的情感或直觉。"
               +"这是表达未经过滤的情绪和感受的时刻。",
    "黑色" : "代表谨慎和批判性思维。你需要指出提案的弱点、风险以及为什么某些事情可能无法按计划进行。"
               +"这不是消极思考，而是为了发现潜在的问题。",
    "黄色" : "代表乐观和积极性。你需要探讨提案的价值、好处和可行性。这是寻找和讨论提案中正面方面的时候。",
    "绿色" : "代表创造性思维和新想法。鼓励发散思维、提出新的观点、解决方案和创意。这是打破常规和探索新可能性的时候。",

# 定义 Tool 

from serpapi import GoogleSearch
import os

def search(query):
    params = {
      "q": query,
      "hl": "en",
      "gl": "us",
      "google_domain": "google.com",
      "api_key": os.environ["SERPAPI_API_KEY"]
    }
    results = GoogleSearch(params).get_dict()
    ans = ""
    for r in results["organic_results"]:
        ans = f"title: {r['title']}\nsnippet: {r['snippet']}\n\n"
    return ans

available_functions={"search":search}

from openai import OpenAI
import os

from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())

# 初始化 OpenAI 服务
client = OpenAI()  

def create_assistant(color):
    assistant = client.beta.assistants.create(
        name=f"{color}帽子角色",
        instructions=f"我们在进行一场Six Thinking Hats讨论。按{queue}顺序。你的角色是{color}帽子。你{hats[color]}",
        model="gpt-3.5-turbo",
        tools=[{
            "type": "function",
            "function": {
              "name": "search",
              "description": "search the web using a search engine",
              "parameters": {
                "type": "object",
                "properties": {
                  "query": {
                    "type": "string",
                    "description": "space-separared keywords to search"
                  }
                },
                "required": ["query"]
              }
            }
      }] if color == "白色" else []
    )
    return assistant

def update_sesssion(context, color, turn_message):
    context += f"\n\n{color}帽子: {turn_message}"
    return context

prompt_template = """
{}
======
以上是讨论的上文。
请严格按照你的角色指示，继续你的发言。直接开始你的发言内容。请保持简短。
"""

def create_a_turn(assistant, context):
    thread = client.beta.threads.create()
    message = client.beta.threads.messages.create(
        thread_id=thread.id,  # message 必须归属于一个 thread
        role="user",          # 取值是 user 或者 assistant。但 assistant 消息会被自动加入，我们一般不需要自己构造
        content=prompt_template.format(context),
    )
    run = client.beta.threads.runs.create(
        assistant_id=assistant.id,
        thread_id=thread.id,
    )
    return run, thread

import time, json

state = 0

def wait_on_run(run, thread):
    """等待 run 结束，返回 run 对象，和成功的结果"""
    def show_rolling_symbol():
        global state
        symbols="\|/-"
        print(f"\r{symbols[state%4]}",end="")
        state += 1
        time.sleep(1)
        
    def hide_rolling_symbol():
        print("\r",end="")
    
    while run.status == "queued" or run.status == "in_progress":
        """还未中止"""
        show_rolling_symbol()
        run = client.beta.threads.runs.retrieve(
            thread_id=thread.id,
            run_id=run.id)
    
    hide_rolling_symbol()
    if run.status == "requires_action":
        """需要调用函数"""
        # 可能有多个函数需要调用，所以用循环
        tool_outputs = []
        for tool_call in run.required_action.submit_tool_outputs.tool_calls:
            # 调用函数
            name = tool_call.function.name
            print("调用函数：" + name + "()")
            print("参数：")
            print(tool_call.function.arguments)
            function_to_call = available_functions[name]
            arguments = json.loads(tool_call.function.arguments)
            result = function_to_call(**arguments)
            print("结果：" + str(result))
            tool_outputs.append({
                "tool_call_id": tool_call.id,
                "output": json.dumps(result),
            })

        # 提交函数调用的结果
        run = client.beta.threads.runs.submit_tool_outputs(
            thread_id=thread.id,
            run_id=run.id,
            tool_outputs=tool_outputs,
        )

        # 递归调用，直到 run 结束
        return wait_on_run(run, thread)

    if run.status == "completed":
        """成功"""
        # 获取全部消息
        messages = client.beta.threads.messages.list(thread_id=thread.id)
        # 最后一条消息排在第一位
        result = messages.data[0].content[0].text.value
        return run, result

    # 执行失败
    return run, None

def discuss(topic):
    context = f"讨论话题：{topic}\n\n[开始]\n"
    # 每个角色依次发言
    for hat in queue:
        print(f"---{hat}----")
        # 每个角色创建一个 assistant
        assistant = create_assistant(hat)
        # 创建 run 和 thread
        new_turn, thread = create_a_turn(assistant, context)
        # 运行 run
        _, text = wait_on_run(new_turn, thread)
        print(f"{text}\n")
        # 更新整个对话历史
        context = update_sesssion(context, hat, text)

7 总结

在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1508643.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Redhat Linux(RHEL) - Primavera P6 EPPM 安装及分享

Redhat Linux(RHEL) - Primavera P6 EPPM 安装及分享

引言 继上一期发布的Oracle Linux版环境发布之后，近日我又制作了基于Redhat Linux 的P6虚拟机环境，同样里面包含了全套P6 最新版应用服务 此虚拟机仅用于演示、培训和测试目的。如您在生产环境中使用此虚拟机，请先与Oracle Primavera销售代表…
阅读更多...
SD-WAN能解决企业网络的哪些问题？

SD-WAN能解决企业网络的哪些问题？

SD-WAN技术的崛起为企业网络带来了全新的解决方案。在数字化转型、云计算、远程办公和5G等领域，SD-WAN技术展现出强劲的市场趋势。那么，SD-WAN究竟能够解决企业网络中的哪些难题呢？ 提升网络带宽利用率 传统网络在连接分支机构时，…
阅读更多...
解决input事件监听拼音输入法导致高频事件

解决input事件监听拼音输入法导致高频事件

1、业务场景 在文本框中输入内容，执行查询接口，但遇到一个问题，当用拼音打字写汉字去搜索的时候，会输入一些字母拼成汉字，怎么能监听等拼音文字输入完成后再触发文本框监听事件 2、解决方案 通过查阅资料得知在输入中…
阅读更多...
Qt教程 — 1.3 如何创建Qt项目

Qt教程 — 1.3 如何创建Qt项目

目录 1 新建一个项目 2 项目文件介绍 2.1 项目文件夹介绍 2.2 配置文件*.pro 2.3 头文件*.h 2.4 源文件*.cpp 2.5 样式文件*.ui 3 修改 ui 文件 4 项目编译&调试&运行 4.1 运行 4.2 编译报错 1 新建一个项目 (1) 新建项目，方法一：打…
阅读更多...
Docker Desktop将镜像存储位置从C盘迁移到其它盘

Docker Desktop将镜像存储位置从C盘迁移到其它盘

一、简述 Docker Desktop默认安装在C盘，默认镜像存储位置在 C:\用户\Administrator\AppData\Local\Docker\wsl Docker Desktop 通过WSL2启动，会自动创建2个子系统，分别对应2个 vhdx 硬盘映像文件。 可以命令行执行wsl --list -v 看到 二、迁移步骤 1、在Docker Desktop…
阅读更多...
基于jsp+mysql+Spring+mybatis的SSM汽车保险理赔管理系统设计和实现

基于jsp+mysql+Spring+mybatis的SSM汽车保险理赔管理系统设计和实现

基于jspmysqlSpringmybatis的SSM汽车保险理赔管理系统设计和实现 博主介绍：多年java开发经验，专注Java开发、定制、远程、文档编写指导等,csdn特邀作者、专注于Java技术领域 作者主页 央顺技术团队 Java毕设项目精品实战案例《1000套》 欢迎点赞 收藏 ⭐…
阅读更多...
C++ 之LeetCode刷题记录（三十九）

C++ 之LeetCode刷题记录（三十九）

😄😊😆😃😄😊😆😃 开始cpp刷题之旅。 目标：执行用时击败90%以上使用 C 的用户。 22. 括号生成 数字 n 代表生成括号的对数，请你设计一个函数，用…
阅读更多...
在文件夹下快速创建vue项目搭建vue框架详细步骤

在文件夹下快速创建vue项目搭建vue框架详细步骤

一、首先在你的电脑目录下新建一个文件夹 进入该文件夹并打开控制台（输入cmd指令） 进入控制台后输入 vue create springboot_vue (自己指定名称) 如果出现这类报错如：npm install 的报错npm ERR! network request to http://registry.cnp…
阅读更多...
【JAVA重要知识 | 第八篇】Java注解总结

【JAVA重要知识 | 第八篇】Java注解总结

文章目录 8.注解8.1注解的定义8.1.1何为注解？8.1.2何为元数据和元编程？（1）元数据（2）元编程（3）总结 8.2注解的分类8.2.1常用的注解8.2.2自定义注解8.2.3元注解（1&#xff…
阅读更多...
轻松掌握锁冲突问题的排查方法——《OceanBase诊断系列》之八

轻松掌握锁冲突问题的排查方法——《OceanBase诊断系列》之八

1. 前言 OceanBase数据库通过两阶段封锁机制确保读写事务并发控制的正确性。在高冲突场景下，事务处理中经常会遇到行锁冲突的问题。然而，许多OceanBase用户对于何时发生锁冲突，锁冲突的表现如何，以及如何排查锁冲突的原因&#x…
阅读更多...
python爬虫（6）之处理数组

python爬虫（6）之处理数组

1、拆分数组 1、spilt（）函数 此函数的用处是将数组均分成几个数组 演示如下： import numpy as np ac np.array([1,2,8,9,3,5,5,8]) ac1 np.split(ac,2) ac2 np.split(ac,[3,6]) print(ac1,ac2) 结果如下： 其中若是一个数…
阅读更多...
【计算机视觉】目标跟踪任务概述和算法介绍

【计算机视觉】目标跟踪任务概述和算法介绍

一、前言 1.1：目标跟踪VS目标检测：区别和联系 区别： 任务目标 目标跟踪任务的目标是在视频序列中跟踪一个特定目标的位置，即给定第一帧中的目标位置后，在后续帧中确定目标的位置。而目标检测任务的目标是在静态图像中…
阅读更多...
解决 Node.js 中 npm ERR! errno CERT_HAS_EXPIRED问题

解决 Node.js 中 npm ERR! errno CERT_HAS_EXPIRED问题

出自 BV1MN411y7pw， P94 黑马AJAX-Node.js-Webpack教学视频中npm包下载dayjs出错情况 输入 npm i dayjs指令之后出错： npm ERR! errno CERT_HAS_EXPIREDnpm ERR! A complete log of this run can be found in: C:\Users\24541\AppData\Local\npm-cache…
阅读更多...
【leetcode C++】最小栈

【leetcode C++】最小栈

leetcode 155. 最小栈 题目 设计一个支持 push ，pop ，top 操作，并能在常数时间内检索到最小元素的栈。 实现 MinStack 类: MinStack() 初始化堆栈对象。void push(int val) 将元素val推入堆栈。void pop() 删除堆栈顶部的元素。int top() 获…
阅读更多...
lspci详解

lspci详解

lspci的作用 lspci是一个Linux命令，用于列出系统中的PCI总线设备信息。PCI（Peripheral Component Interconnect）是一种常见的计算机总线标准，用于连接各种外部设备（如网卡、显卡、声卡等）到计算机主板上。…
阅读更多...
【CVPR2024】ReID3D：首个使用激光雷达行人充实别 ReID 的工作！

【CVPR2024】ReID3D：首个使用激光雷达行人充实别 ReID 的工作！

【2024 行人重识别最新进展】ReID3D：首个关注激光雷达行人 ReID 的工作！ 摘要：数据集：方法模型：多任务预训练：ReID Network： 实验结果：结论： 来源：Arxiv 2023…
阅读更多...
mysql中 COALESCE和CASE WHEN的使用以及创建或替换视图

mysql中 COALESCE和CASE WHEN的使用以及创建或替换视图

create or replace view 自理能力评估视图 as SELECT ehr_zlnlpg.ID AS ID, ehr_zlnlpg.GRID AS GRID, ehr_zlnlpg.TJID AS TJID, ehr_grjbxx.Name AS 姓名, ehr_grjbxx.Sex AS 性别, ehr_grjbxx.Cardnum AS 身份证号, ehr_zlnlpg.SCORESUM AS 总…
阅读更多...
设计模式学习笔记 - 规范与重构 - 5.如何通过封装、抽象、模块化、中间层解耦代码？

设计模式学习笔记 - 规范与重构 - 5.如何通过封装、抽象、模块化、中间层解耦代码？

前言 《规范与重构 - 1.什么情况下要重构？重构什么？又该如何重构？》讲过，重构可以分为大规模高层重构（简称 “大型重构”）和小规模低层次重构（简称 “小型重构”）。大型重构是对系统…
阅读更多...
UI自动化测试使用场景及脚本录制

UI自动化测试使用场景及脚本录制

经常有人会问，什么样的项目才适合进行UI自动化测试呢？UI自动化测试相当于模拟手工测试，通过程序去操作页面上的控件。而在实际测试过程中，经常会遇到无法找到控件，或者因控件定义变更而带来的维护成本等问题。 哪些场…
阅读更多...
MySQL 事务的原理以及长事务的预防和处置

MySQL 事务的原理以及长事务的预防和处置

transaction_isolation 隔离级别 读未提交 读提交 视图是在每个 SQL 语句开始执行的时候创建的 可重复读 视图是在事务启动时创建的，整个事务存在期间都用这个视图 串行化…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023