OpenAl大模型开发与应用实践

Posted on 2025-03-13 Edited on 2025-03-15 In 人工智能 , 大模型 Views: Word count in article: 38k Reading time ≈ 35 mins.

1、Models API

1.1 List Models

列出当前可用的模型，并提供每个模型的基本信息，如所有者和可用性。

import os
from openai import OpenAI
client = OpenAI()

# 需要在环境变量中先设置OPENAI_API_KEY
models = client.models.list()
print(models)

SyncPage[Model](data=[Model(id='gpt-4o-2024-11-20', created=1739331543, object='model', owned_by='system'), Model(id='gpt-4o-mini-audio-preview-2024-12-17', created=1734115920, object='model', owned_by='system'), Model(id='dall-e-3', created=1698785189, object='model', owned_by='system'), Model(id='dall-e-2', created=1698798177, object='model', owned_by='system'), Model(id='gpt-4o-audio-preview-2024-10-01', created=1727389042, object='model', owned_by='system'), Model(id='gpt-4o-audio-preview', created=1727460443, object='model', owned_by='system'), Model(id='o1-mini-2024-09-12', created=1725648979, object='model', owned_by='system'), Model(id='o1-mini', created=1725649008, object='model', owned_by='system'), Model(id='omni-moderation-latest', created=1731689265, object='model', owned_by='system'), Model(id='gpt-4o-mini', created=1721172741, object='model', owned_by='system'), Model(id='gpt-4o-mini-audio-preview', created=1734387424, object='model', owned_by='system'), Model(id='omni-moderation-2024-09-26', created=1732734466, object='model', owned_by='system'), Model(id='babbage-002', created=1692634615, object='model', owned_by='system'), Model(id='tts-1-hd-1106', created=1699053533, object='model', owned_by='system'), Model(id='text-embedding-3-large', created=1705953180, object='model', owned_by='system'), Model(id='gpt-4o-2024-05-13', created=1715368132, object='model', owned_by='system'), Model(id='tts-1-hd', created=1699046015, object='model', owned_by='system'), Model(id='o1-preview', created=1725648897, object='model', owned_by='system'), Model(id='o1-preview-2024-09-12', created=1725648865, object='model', owned_by='system'), Model(id='gpt-3.5-turbo-instruct-0914', created=1694122472, object='model', owned_by='system'), Model(id='gpt-4o-mini-search-preview', created=1741391161, object='model', owned_by='system'), Model(id='tts-1-1106', created=1699053241, object='model', owned_by='system'), Model(id='davinci-002', created=1692634301, object='model', owned_by='system'), Model(id='gpt-3.5-turbo-1106', created=1698959748, object='model', owned_by='system'), Model(id='gpt-4o-search-preview', created=1741388720, object='model', owned_by='system'), Model(id='gpt-3.5-turbo-instruct', created=1692901427, object='model', owned_by='system'), Model(id='gpt-4o-mini-search-preview-2025-03-11', created=1741390858, object='model', owned_by='system'), Model(id='gpt-3.5-turbo-0125', created=1706048358, object='model', owned_by='system'), Model(id='gpt-4o-2024-08-06', created=1722814719, object='model', owned_by='system'), Model(id='gpt-3.5-turbo', created=1677610602, object='model', owned_by='openai'), Model(id='gpt-3.5-turbo-16k', created=1683758102, object='model', owned_by='openai-internal'), Model(id='gpt-4o', created=1715367049, object='model', owned_by='system'), Model(id='text-embedding-3-small', created=1705948997, object='model', owned_by='system'), Model(id='text-embedding-ada-002', created=1671217299, object='model', owned_by='openai-internal'), Model(id='gpt-4.5-preview', created=1740623059, object='model', owned_by='system'), Model(id='gpt-4.5-preview-2025-02-27', created=1740623304, object='model', owned_by='system'), Model(id='gpt-4o-search-preview-2025-03-11', created=1741388170, object='model', owned_by='system'), Model(id='tts-1', created=1681940951, object='model', owned_by='openai-internal'), Model(id='gpt-4o-mini-2024-07-18', created=1721172717, object='model', owned_by='system'), Model(id='whisper-1', created=1677532384, object='model', owned_by='openai-internal')], object='list')

models.data: 目前OpenAI提供的大语言模型列表，列表中的每一项都对应着一个模型实例。
以GPT-3.5-Turbo模型为例，解释说明各项参数：
- created: 这是模型创建的时间戳，单位为 Unix 时间戳（自1970年1月1日（00:00:00 GMT）以后的秒数）。
- id: 这是模型的唯一标识符。在这个例子中，模型的 ID 是 “text-davinci-003”。
- object: 这个字段表示的是当前对象的类型，在这个例子中，对象是 “model”，说明这个 JSON 对象是一个模型。
- owned_by: 这个字段表示的是模型的所有者，在这个例子中，模型的所有者是 “openai-internal”。

1	models.data

获取模型ID列表

models.data[0].id # 值为"gpt-4o-2024-11-20"

model_list = [model.id for model in models.data]
print(model_list)

1.2 Retrieve Model

根据前面查询到当前支持的模型ID列表，获取指定模型实例，如gpt-3.5-turbo。

from openai import OpenAI
client = OpenAI()

# 将模型 ID 传入 retrieve 接口
gpt_3 = client.models.retrieve("gpt-3.5-turbo")
print(gpt_3)

2、Completions API

使用Completions API实现各类文本生成任务，主要请求参数说明：
- model （string，必填）：要使用的模型的 ID。可以参考 模型端点兼容性表。
- prompt （string or array，必填，Defaults to ）：生成补全的提示，编码为字符串、字符串数组、token数组或token数组数组。注意，这是模型在训练过程中看到的文档分隔符，所以如果没有指定提示符，模型将像从新文档的开头一样生成。
- stream （boolean，选填，默认 false）：当它设置为 true 时，API 会以 SSE（ Server Side Event ）方式返回内容，即会不断地输出内容直到完成响应，流通过 data: [DONE] 消息终止。
- max_tokens （integer，选填，默认是 16）：补全时要生成的最大 token 数。提示 max_tokens 的 token 计数不能超过模型的上下文长度。大多数模型的上下文长度为 2048 个token（最新模型除外，它支持 4096）
- temperature （number，选填，默认是1）：使用哪个采样温度，在 0和2之间。较高的值，如0.8会使输出更随机，而较低的值，如0.2会使其更加集中和确定性。通常建议修改这个（temperature ）或 top_p 但两者不能同时存在，二选一。
- n （integer，选填，默认为 1）：每个 prompt 生成的补全次数。注意：由于此参数会生成许多补全，因此它会快速消耗token配额。小心使用，并确保对 max_tokens 和 stop 进行合理的设置。

生成英文文本：

import os
from openai import OpenAI

client = OpenAI()

data = client.completions.create(
  model="gpt-3.5-turbo-instruct",
  prompt="Say this is a test",
  max_tokens=7,
  temperature=0
)

print(data)

1 2	text = data.choices[0].text print(text) # 值为"None"

### 【新增】/v1/completions 调用也即将过期； 更新为 /v1/chat/completions ，使用gpt-4o-mini模型

from openai import OpenAI

client = OpenAI()

data = client.chat.completions.create(
    model="gpt-4o-mini",
    max_tokens=7,
temperature=0,
    messages=[
        {"role": "user", "content": "Say this is a test"}
    ]
)
print(data)
# 内容输出数据结构，和/v1/completions 接口有变化，相应调整
text = data.choices[0].message.content
print(text)

生成中文文本：

data = client.completions.create(
  model="gpt-3.5-turbo-instruct",
  prompt="讲10个给程序员听得笑话",
  max_tokens=1000,
  temperature=0.5
)

text = data.choices[0].text
print(text) # 值为"None"

### 【新增】/v1/completions 调用也即将过期； 更新为 /v1/chat/completions ，使用gpt-4o-mini模型

from openai import OpenAI

client = OpenAI()

data = client.chat.completions.create(
    model="gpt-4o-mini",
    max_tokens=1000,
    temperature=0.5,
    messages=[
        {"role": "user", "content": "讲10个给程序员听得笑话"}
    ]
)
print(data)
# 内容输出数据结构，和/v1/completions 接口有变化，相应调整
text = data.choices[0].message.content
print(text)

生成Python代码，并执行和验证：

data = client.completions.create(
  model="gpt-3.5-turbo-instruct",
  prompt="生成可执行的快速排序 Python 代码",
  max_tokens=1000,
  temperature=0
)

text = data.choices[0].text
print(text) # 值为"None"

### 【新增】/v1/completions 调用也即将过期； 更新为 /v1/chat/completions ，使用gpt-4o-mini模型

from openai import OpenAI

client = OpenAI()

data = client.chat.completions.create(
    model="gpt-4o-mini",
    max_tokens=1000,
    temperature=0,
    messages=[
        {"role": "user", "content": "生成可执行的快速排序 Python 代码，只输出代码块内容，函数名quick_sort"}
    ]
)
print("\n【完整内容】：\n",data)
# 内容输出数据结构，和/v1/completions 接口有变化，相应调整
text = data.choices[0].message.content
print("\n【文本内容】：\n",text)

1
2
3

text = data.choices[0].message.content
text = text.replace("`python", "").replace("`", "").strip() # 去除标记并去除首尾空白
print("\n【清理后的文本内容】：\n", text)

# Jupyter Notebook 中执行生成的代码
# 我现在用 Completion API 生成了 Python 代码，并以字符串形式存放在 text 中，如下所示：
text = data.choices[0].text
print(text)

def quick_sort(arr):
    if len(arr) <= 1:
        return arr
    pivot = arr[0]
    left = [x for x in arr[1:] if x <= pivot]
    right = [x for x in arr[1:] if x > pivot]
    return quick_sort(left) + [pivot] + quick_sort(right)
  
# `exec` 函数会执行传入的字符串作为 Python 代码。
# 在这个例子中，我们使用 `exec` 来定义了一个 `quick_sort` 函数，然后你就可以调用这个函数了。
# 请注意，`exec` 可以执行任何 Python 代码，因此在使用它的时候一定要小心，特别是当你执行的代码来自不可信的来源时。
exec(text)

# 现在你可以调用这个函数了
print(quick_sort([12,3,6,8,10,1,2,1]))

3、Chat Completions API

使用Chat Completions API实现对话任务，聊天补全(Chat Completions API)以消息列表作为输入，并返回模型生成的消息作为输出。尽管聊天格式旨在使多轮对话变得简单，但它同样适用于没有任何对话的单轮任务。主要请求参数说明：
- model （string，必填）：要使用的模型ID。有关哪些模型适用于Chat API的详细信息。
- messages （array，必填）：迄今为止描述对话的消息列表。
  - role （string，必填）：发送此消息的角色。system 、user 或 assistant 之一（一般用 user 发送用户问题，system 发送给模型提示信息）。
  - content （string，必填）：消息的内容。
  - name （string，选填）：此消息的发送者姓名。可以包含 a-z、A-Z、0-9 和下划线，最大长度为 64 个字符。
- stream （boolean，选填，是否按流的方式发送内容）：当它设置为 true 时，API 会以 SSE（ Server Side Event ）方式返回内容。SSE 本质上是一个长链接，会持续不断地输出内容直到完成响应。如果不是做实时聊天，默认false即可。
- max_tokens （integer，选填）：在聊天补全中生成的最大 tokens 数。输入token和生成的token的总长度受模型上下文长度的限制。
- temperature （number，选填，默认是 1）。采样温度，在 0和 2 之间。较高的值，如0.8会使输出更随机，而较低的值，如0.2会使其更加集中和确定性。通常建议修改这个（temperature ）或者 top_p ，但两者不能同时存在，二选一。

开启聊天模式

from openai import OpenAI
client = OpenAI()

# 使用 messages 记录迄今为止对话的消息列表
messages=[
    {
        "role": "user", 
        "content": "Hello!"
    }
]

data = client.chat.completions.create(
  model="gpt-3.5-turbo",
  messages = messages
)
print(data)

# 从返回的数据中获取生成的消息
new_message = data.choices[0].message
# 打印 new_message
print(new_message)

1
2
3

# 将消息追加到 messages 列表中
messages.append(new_message)
print(messages)

type(new_message) # 值为"openai.types.chat.chat_completion_message.ChatCompletionMessage"
# 发现new_message类型不对，需要做转化再加入messages中
messages.pop()
new_message = data.choices[0].message
new_message_dict = {"role": new_message.role, "content": new_message.content}
type(new_message_dict) # 值为"dict"
print(new_message_dict) # 值为"{'role': 'assistant', 'content': '你好！有什么我可以帮助你的吗？\n'}"
# 将消息追加到 messages 列表中
messages.append(new_message_dict)
print(messages)

新一轮对话

new_chat = {
    "role": "user",
    "content": "1.讲一个程序员才听得懂的冷笑话；2.今天是几号？3.明天星期几？"
}

messages.append(new_chat)

from pprint import pprint

pprint(messages)

data = client.chat.completions.create(
  model="gpt-3.5-turbo",
  messages=messages
)

new_message = data.choices[0].message
# 打印 new_messages 
print(new_message)

1 2	# 打印 new_messages 内容 print(new_message.content)

使用多种身份聊天对话

目前role参数支持3类身份： system, user assistant：

# 构造聊天记录
messages=[
    {"role": "system", "content": "你是一个乐于助人的体育界专家。"},
    {"role": "user", "content": "2008年奥运会是在哪里举行的？"},
]

import openai

data = client.chat.completions.create(
  model="gpt-3.5-turbo",
  messages=messages
)

message = data.choices[0].message.content
print(message) # 2008年夏季奥运会在北京举行

# 添加 GPT 返回结果到聊天记录
messages.append({"role": "assistant", "content": message})
messages

1
2
3

# 第二轮对话
messages.append({"role": "user", "content": "1.金牌最多的是哪个国家？2.奖牌最多的是哪个国家？"})
messages

data = client.chat.completions.create(
  model="gpt-3.5-turbo",
  messages=messages
)
message = data.choices[0].message.content
print(message)

data = client.chat.completions.create(
  model="gpt-3.5-turbo",
  messages=[{'role': 'user', 'content': '1.金牌最多的是哪个国家？2.奖牌最多的是哪个国家？'}]
)
data.choices[0].message.content

4、使用tiktoken计算token数量

tiktoken是OpenAI开发的一种BPE分词器。
给定一段文本字符串（例如，"tiktoken is great!"）和一种编码方式（例如，"cl100k_base"），分词器可以将文本字符串切分成一系列的token（例如，["t", "ik", "token", " is", " great", "!"]）。
将文本字符串切分成token非常有用，因为GPT模型看到的文本就是以token的形式呈现的。知道一段文本字符串中有多少个token可以告诉你（a）这个字符串是否对于文本模型来说太长了而无法处理，以及（b）一个OpenAI API调用的费用是多少（因为使用量是按照token计价的）。
编码方式规定了如何将文本转换成token。不同的模型使用不同的编码方式。tiktoken支持OpenAI模型使用的三种编码方式：

你可以使用 tiktoken.encoding_for_model() 获取一个模型的编码方式，如下所示：

1 2	encoding = tiktoken.encoding_for_model('gpt-3.5-turbo') # 注意，p50k_base 与 r50k_base 有很大的重叠，对于非代码应用，它们通常会产生相同的token。

不同语言的分词器库，在英语中，token的长度通常在一个字符到一个单词之间变化（例如，"t" 或 " great"），尽管在某些语言中，token可以比一个字符短或比一个单词长。空格通常与单词的开头一起分组（例如，" is" 而不是 "is " 或 " "+"is"）。你可以快速在 OpenAI分词器检查一段字符串如何被分词。

对于 cl100k_base 和 p50k_base 编码方式：
- Python: tiktoken
- .NET / C#: SharpToken, TiktokenSharp
- Java: jtokkit
对于 r50k_base (gpt2) 编码方式，许多语言都提供了分词器：
- Python: tiktoken (或者另选 GPT2TokenizerFast)
- JavaScript: gpt-3-encoder
- .NET / C#: GPT Tokenizer
- Java: gpt2-tokenizer-java
- PHP: GPT-3-Encoder-PHP

1 2	# 安装 tiktoken !pip install --upgrade tiktoken

import tiktoken

# 使用tiktoken.get_encoding()按名称加载编码
# 第一次运行时，它将需要互联网连接进行下载。后续运行不需要互联网连接
encoding = tiktoken.get_encoding("cl100k_base")
# 使用tiktoken.encoding_for_model()函数可以自动加载给定模型名称的正确编码
encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")

encoding.encode("tiktoken is great!") # [83, 1609, 5963, 374, 2294, 0]

# 通过计算.encode()返回的列表的长度来统计token数量
def num_tokens_from_string(string: str, encoding_name: str) -> int:
    """返回文本字符串中的Token数量"""
    encoding = tiktoken.get_encoding(encoding_name)
    num_tokens = len(encoding.encode(string))
    return num_tokens
  
num_tokens_from_string("tiktoken is great!", "cl100k_base") # 6

# .decode()将一个token整数列表转换为字符串
encoding.decode([83, 1609, 5963, 374, 2294, 0]) # 'tiktoken is great!'
encoding.decode([0,1,2,3]) # '!"#$'

# 注意：尽管.decode()可以应用于单个token，但对于不在 utf-8 边界上的token来说，解码可能会有损失或错误
# 对于单个token，.decode_single_token_bytes() 安全地将单个整数token转换为其表示的字节
# 在字符串前面的b表示这些字符串是字节字符串
[encoding.decode_single_token_bytes(token) for token in [83, 1609, 5963, 374, 2294, 0]] # [b't', b'ik', b'token', b' is', b' great', b'!']

不同的编码方式在分割单词、处理空格和非英文字符方面存在差异。通过上述方法，我们可以比较几个示例字符串在不同的编码方式下的表现。

def compare_encodings(example_string: str) -> None:
    """Prints a comparison of three string encodings."""
    # print the example string
    print(f'\nExample string: "{example_string}"')
    # for each encoding, print the # of tokens, the token integers, and the token bytes
    for encoding_name in ["gpt2", "p50k_base", "cl100k_base"]:
        encoding = tiktoken.get_encoding(encoding_name)
        token_integers = encoding.encode(example_string)
        num_tokens = len(token_integers)
        token_bytes = [encoding.decode_single_token_bytes(token) for token in token_integers]
        print()
        print(f"{encoding_name}: {num_tokens} tokens")
        print(f"token integers: {token_integers}")
        print(f"token bytes: {token_bytes}")

1	compare_encodings("antidisestablishmentarianism")

1	compare_encodings("2 + 2 = 4")

1	compare_encodings("お誕生日おめでとう")

ChatGPT模型，如gpt-3.5-turbo和gpt-4，与旧的完成模型一样使用token，但由于其基于消息的格式，很难准确计算对话中将使用多少个token。下面是一个示例函数，用于计算传递给gpt-3.5-turbo或gpt-4的消息中的token数量。请注意，从消息中计算token的确切方式可能因模型而异。请将下面函数中的计数视为估计值，并非永恒保证。特别地，在使用可选功能输入(input)的请求上方会消耗额外的token。

# 定义函数 num_tokens_from_messages，该函数返回由一组消息所使用的token数。
def num_tokens_from_messages(messages, model="gpt-3.5-turbo"):
    """Return the number of tokens used by a list of messages."""
    # 尝试获取模型的编码
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        # 如果模型没有找到，使用 cl100k_base 编码并给出警告
        print("Warning: model not found. Using cl100k_base encoding.")
        encoding = tiktoken.get_encoding("cl100k_base")
    # 针对不同的模型设置token数量
    if model in {
        "gpt-3.5-turbo-0613",
        "gpt-3.5-turbo-16k-0613",
        "gpt-4-0314",
        "gpt-4-32k-0314",
        "gpt-4-0613",
        "gpt-4-32k-0613",
        }:
        tokens_per_message = 3
        tokens_per_name = 1
    elif model == "gpt-3.5-turbo-0301":
        tokens_per_message = 4  # 每条消息遵循 {role/name}\n{content}\n 格式
        tokens_per_name = -1  # 如果有名字，角色会被省略
    elif "gpt-3.5-turbo" in model:
        # 对于 gpt-3.5-turbo 模型可能会有更新，此处返回假设为 gpt-3.5-turbo-0613 的token数量，并给出警告
        print("Warning: gpt-3.5-turbo may update over time. Returning num tokens assuming gpt-3.5-turbo-0613.")
        return num_tokens_from_messages(messages, model="gpt-3.5-turbo-0613")
    elif "gpt-4" in model:
        # 对于 gpt-4 模型可能会有更新，此处返回假设为 gpt-4-0613 的token数量，并给出警告
        print("Warning: gpt-4 may update over time. Returning num tokens assuming gpt-4-0613.")
        return num_tokens_from_messages(messages, model="gpt-4-0613")
    else:
        # 对于没有实现的模型，抛出未实现错误
        raise NotImplementedError(
            f"""num_tokens_from_messages() is not implemented for model {model}. See https://github.com/openai/openai-python/blob/main/chatml.md for information on how messages are converted to tokens."""
        )
    num_tokens = 0
    # 计算每条消息的token数
    for message in messages:
        num_tokens += tokens_per_message
        for key, value in message.items():
            num_tokens += len(encoding.encode(value))
            if key == "name":
                num_tokens += tokens_per_name
    num_tokens += 3  # 每条回复都以助手为首
    return num_tokens

# 让我们验证上面的函数是否与OpenAI API的响应匹配

# OpenAI Python SDK v1.0 更新后的使用方式
from openai import OpenAI
client = OpenAI()


example_messages = [
    {
        "role": "system",
        "content": "You are a helpful, pattern-following assistant that translates corporate jargon into plain English.",
    },
    {
        "role": "system",
        "name": "example_user",
        "content": "New synergies will help drive top-line growth.",
    },
    {
        "role": "system",
        "name": "example_assistant",
        "content": "Things working well together will increase revenue.",
    },
    {
        "role": "system",
        "name": "example_user",
        "content": "Let's circle back when we have more bandwidth to touch base on opportunities for increased leverage.",
    },
    {
        "role": "system",
        "name": "example_assistant",
        "content": "Let's talk later when we're less busy about how to do better.",
    },
    {
        "role": "user",
        "content": "This late pivot means we don't have time to boil the ocean for the client deliverable.",
    },
]

for model in [
    "gpt-3.5-turbo-0613",
    "gpt-3.5-turbo",
    "gpt-4-0613",
    "gpt-4",
    ]:
    print(model)
    # example token count from the function defined above
    print(f"{num_tokens_from_messages(example_messages, model)} prompt tokens counted by num_tokens_from_messages().")
    # example token count from the OpenAI API
    # OpenAI Python SDK v1.0 更新后的使用方式
    completion = client.chat.completions.create(
        model=model,
        messages=example_messages,
        temperature=0,
        max_tokens=1,  # we're only counting input tokens here, so let's not waste tokens on the output
    )
    print(f'{completion.usage.prompt_tokens} prompt tokens counted by the OpenAI API.')
    print()

虽然 GPT-3.5 和 GPT-4 使用了不同的训练集，支持不同的上下文长度，但是他们使用了相同的 Embedding 模型（相同的编码方式），因此分词后，编码完都是 129 个 tokens。

5、Function Calling介绍与实战

2023年6月20日，OpenAI 官方在 Chat Completions API 原有的三种不同角色设定（System, Assistant, User）基础上，新增了 Function Calling 功能。https://openai.com/index/function-calling-and-other-api-updates/
functions 是 Chat Completion API 中的可选参数，用于提供函数定义。其目的是使 GPT 模型能够生成符合所提供定义的函数参数。请注意，API不会实际执行任何函数调用。开发人员需要使用GPT 模型输出来执行函数调用。
如果提供了functions参数，默认情况下，GPT 模型将决定在何时适当地使用其中一个函数。
可以通过将function_call参数设置为{"name": "<insert-function-name>"}来强制 API 使用指定函数。
同时，也支持通过将function_call参数设置为"none"来强制API不使用任何函数。

如果使用了某个函数，则响应中的输出将包含"finish_reason": "function_call"，以及一个具有该函数名称和生成的函数参数的function_call对象。

1 2	# 安装依赖包 !pip install scipy tenacity tiktoken termcolor openai requests

import json
import requests
import os
from tenacity import retry, wait_random_exponential, stop_after_attempt
from termcolor import colored

GPT_MODEL = "gpt-3.5-turbo"

# 定义工具函数
# 使用了retry库，指定在请求失败时的重试策略。
# 这里设定的是指数等待（wait_random_exponential），时间间隔的最大值为40秒，并且最多重试3次（stop_after_attempt(3)）。
# 定义一个函数chat_completion_request，主要用于发送 聊天补全 请求到OpenAI服务器

#### 【新增】
# 智谱模型 Function Calling
# 参考： https://www.bigmodel.cn/dev/howuse/functioncall

# deepseek-chat 模型 Function Calling
# 注意：当前版本 deepseek-chat 模型 Function Calling 功能效果不稳定，会出现循环调用、空回复的情况。
# 参考： https://api-docs.deepseek.com/zh-cn/guides/function_calling

### 使用 智谱模型 Function Calling 和 deepseek-chat 模型 Function Calling 注意替换模型名称，调用地址，和API KEY
@retry(wait=wait_random_exponential(multiplier=1, max=40), stop=stop_after_attempt(3))
def chat_completion_request(messages, functions=None, function_call=None, model=GPT_MODEL):

    # 设定请求的header信息，包括 API_KEY
    headers = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + os.getenv("OPENAI_API_KEY"),
    }

    # 设定请求的JSON数据，包括GPT 模型名和要进行补全的消息
    json_data = {"model": model, "messages": messages}

    # 如果传入了functions，将其加入到json_data中
    if functions is not None:
        json_data.update({"functions": functions})

    # 如果传入了function_call，将其加入到json_data中
    if function_call is not None:
        json_data.update({"function_call": function_call})

    # 尝试发送POST请求到OpenAI服务器的chat/completions接口
    try:
        response = requests.post(
            "https://api.openai.com/v1/chat/completions",
            headers=headers,
            json=json_data,
        )
        # 返回服务器的响应
        return response

    # 如果发送请求或处理响应时出现异常，打印异常信息并返回
    except Exception as e:
        print("Unable to generate ChatCompletion response")
        print(f"Exception: {e}")
        return e

# 定义一个函数pretty_print_conversation，用于打印消息对话内容
def pretty_print_conversation(messages):

    # 为不同角色设置不同的颜色
    role_to_color = {
        "system": "red",
        "user": "green",
        "assistant": "blue",
        "function": "magenta",
    }

    # 遍历消息列表
    for message in messages:

        # 如果消息的角色是"system"，则用红色打印“content”
        if message["role"] == "system":
            print(colored(f"system: {message['content']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"user"，则用绿色打印“content”
        elif message["role"] == "user":
            print(colored(f"user: {message['content']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"assistant"，并且消息中包含"function_call"，则用蓝色打印"function_call"
        elif message["role"] == "assistant" and message.get("function_call"):
            print(colored(f"assistant[function_call]: {message['function_call']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"assistant"，但是消息中不包含"function_call"，则用蓝色打印“content”
        elif message["role"] == "assistant" and not message.get("function_call"):
            print(colored(f"assistant[content]: {message['content']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"function"，则用品红色打印“function”
        elif message["role"] == "function":
            print(colored(f"function ({message['name']}): {message['content']}\n", role_to_color[message["role"]]))

如何使用functions参数

这段代码定义了两个可以在程序中调用的函数，分别是获取当前天气和获取未来N天的天气预报。每个函数(function)都有其名称、描述和需要的参数（包括参数的类型、描述等信息）。我们将把这些传递给 Chat Completions API，以生成符合规范的函数。

# 定义一个名为functions的列表，其中包含两个字典，这两个字典分别定义了两个功能的相关参数

# 第一个字典定义了一个名为"get_current_weather"的功能
functions = [
    {
        "name": "get_current_weather",  # 功能的名称
        "description": "Get the current weather",  # 功能的描述
        "parameters": {  # 定义该功能需要的参数
            "type": "object",
            "properties": {  # 参数的属性
                "location": {  # 地点参数
                    "type": "string",  # 参数类型为字符串
                    "description": "The city and state, e.g. San Francisco, CA",  # 参数的描述
                },
                "format": {  # 温度单位参数
                    "type": "string",  # 参数类型为字符串
                    "enum": ["celsius", "fahrenheit"],  # 参数的取值范围
                    "description": "The temperature unit to use. Infer this from the users location.",  # 参数的描述
                },
            },
            "required": ["location", "format"],  # 该功能需要的必要参数
        },
    },
    # 第二个字典定义了一个名为"get_n_day_weather_forecast"的功能
    {
        "name": "get_n_day_weather_forecast",  # 功能的名称
        "description": "Get an N-day weather forecast",  # 功能的描述
        "parameters": {  # 定义该功能需要的参数
            "type": "object",
            "properties": {  # 参数的属性
                "location": {  # 地点参数
                    "type": "string",  # 参数类型为字符串
                    "description": "The city and state, e.g. San Francisco, CA",  # 参数的描述
                },
                "format": {  # 温度单位参数
                    "type": "string",  # 参数类型为字符串
                    "enum": ["celsius", "fahrenheit"],  # 参数的取值范围
                    "description": "The temperature unit to use. Infer this from the users location.",  # 参数的描述
                },
                "num_days": {  # 预测天数参数
                    "type": "integer",  # 参数类型为整数
                    "description": "The number of days to forecast",  # 参数的描述
                }
            },
            "required": ["location", "format", "num_days"]  # 该功能需要的必要参数
        },
    },
]

这段代码首先定义了一个messages列表用来存储聊天的消息，然后向列表中添加了系统和用户的消息。然后，它使用了之前定义的chat_completion_request函数发送一个请求，传入的参数包括消息列表和函数列表。在接收到响应后，它从JSON响应中解析出助手的消息，并将其添加到消息列表中。最后，它打印出 GPT 模型回复的消息。

# 定义一个空列表messages，用于存储聊天的内容
messages = []

# 使用append方法向messages列表添加一条系统角色的消息
messages.append({
    "role": "system",  # 消息的角色是"system"
    "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."  # 消息的内容
})

# 向messages列表添加一条用户角色的消息
messages.append({
    "role": "user",  # 消息的角色是"user"
    "content": "What's the weather like today"  # 用户询问今天的天气情况
})

# 使用定义的chat_completion_request函数发起一个请求，传入messages和functions作为参数
chat_response = chat_completion_request(
    messages, functions=functions
)

# 解析返回的JSON数据，获取助手的回复消息
assistant_message = chat_response.json()["choices"][0]["message"]

# 将助手的回复消息添加到messages列表中
messages.append(assistant_message)

pretty_print_conversation(messages)

从结果可以看出我们需要提供更详细的信息，以便于 GPT 模型为我们生成适当的函数和对应参数。

使用GPT模型生成函数和对应参数

下面这段代码先向messages列表中添加了用户的位置信息；然后再次使用了chat_completion_request函数发起请求，只是这次传入的消息列表已经包括了用户的新消息；在获取到响应后，它同样从JSON响应中解析出助手的消息，并将其添加到消息列表中。最后，打印出助手的新的回复消息。

# 向messages列表添加一条用户角色的消息，用户告知他们在苏格兰的格拉斯哥
messages.append({
    "role": "user",  # 消息的角色是"user"
    "content": "I'm in Shanghai, China."  # 用户的消息内容
})

# 再次使用定义的chat_completion_request函数发起一个请求，传入更新后的messages和functions作为参数
chat_response = chat_completion_request(
    messages, functions=functions
)

# 解析返回的JSON数据，获取助手的新的回复消息
assistant_message = chat_response.json()["choices"][0]["message"]

# 将助手的新的回复消息添加到messages列表中
messages.append(assistant_message)

pretty_print_conversation(messages)

这段代码的逻辑大体与上一段代码相同，区别在于这次用户的询问中涉及到未来若干天（x天）的天气预报。在获取到回复后，它同样从JSON响应中解析出助手的消息，并将其添加到消息列表中。然后打印出助手的回复消息。（通过不同的prompt方式，我们可以让它针对我们告诉它的其他功能。）

# 初始化一个空的messages列表
messages = []

# 向messages列表添加一条系统角色的消息，要求不做关于函数参数值的假设，如果用户的请求模糊，应该寻求澄清
messages.append({
    "role": "system",  # 消息的角色是"system"
    "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."
})

# 向messages列表添加一条用户角色的消息，用户询问在未来x天内苏格兰格拉斯哥的天气情况
messages.append({
    "role": "user",  # 消息的角色是"user"
    "content": "what is the weather going to be like in Shanghai, China over the next x days"
})

# 使用定义的chat_completion_request函数发起一个请求，传入messages和functions作为参数
chat_response = chat_completion_request(
    messages, functions=functions
)

# 解析返回的JSON数据，获取助手的回复消息
assistant_message = chat_response.json()["choices"][0]["message"]

# 将助手的回复消息添加到messages列表中
messages.append(assistant_message)

# 打印助手的回复消息
pretty_print_conversation(messages)

(GPT 模型再次要求我们澄清，因为它还没有足够的信息。在这种情况下，它已经知道预测的位置，但需要知道需要多少天的预测)

这段代码的主要目标是将用户指定的天数（5天）添加到消息列表中，然后再次调用chat_completion_request函数发起一个请求。返回的响应中包含了助手对用户的回复，即未来5天的天气预报。这个预报是基于用户指定的地点（上海）和天数（5天）生成的。在代码的最后，它解析出返回的JSON响应中的第一个选项，这就是助手的回复消息。

# 向messages列表添加一条用户角色的消息，用户指定接下来的天数为5天
messages.append({
    "role": "user",  # 消息的角色是"user"
    "content": "5 days"
})

# 使用定义的chat_completion_request函数发起一个请求，传入messages和functions作为参数
chat_response = chat_completion_request(
    messages, functions=functions
)

# 解析返回的JSON数据，获取第一个选项
assistant_message = chat_response.json()["choices"][0]["message"]

# 将助手的回复消息添加到messages列表中
messages.append(assistant_message)

# 打印助手的回复消息
pretty_print_conversation(messages)

强制使用指定函数

我们可以通过使用function_call参数来强制GPT模型使用指定函数，例如get_n_day_weather_forecast。通过这种方式，可以让 GPT 模型学习如何使用该函数。

# 在这个代码单元中，我们强制GPT 模型使用get_n_day_weather_forecast函数
messages = []  # 创建一个空的消息列表

# 添加系统角色的消息
messages.append({
    "role": "system",  # 角色为系统
    "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."
})

# 添加用户角色的消息
messages.append({
    "role": "user",  # 角色为用户
    "content": "Give me a weather report for San Diego, USA."
})

# 使用定义的chat_completion_request函数发起一个请求，传入messages、functions以及特定的function_call作为参数
chat_response = chat_completion_request(
    messages, functions=functions, function_call={"name": "get_n_day_weather_forecast"}
)

# 解析返回的JSON数据，获取第一个选项
assistant_message = chat_response.json()["choices"][0]["message"]

# 将助手的回复消息添加到messages列表中
messages.append(assistant_message)

# 打印助手的回复消息
pretty_print_conversation(messages)

下面这段代码演示了在不强制使用特定函数（get_n_day_weather_forecast）的情况下，GPT 模型可能会选择不同的方式来回应用户的请求。对于给定的用户请求”Give me a weather report for San Diego, USA.”，GPT 模型可能不会调用get_n_day_weather_forecast函数。

# 如果我们不强制GPT 模型使用 get_n_day_weather_forecast，它可能不会使用
messages = []  # 创建一个空的消息列表

# 添加系统角色的消息
messages.append({
    "role": "system",  # 角色为系统
    "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."
})

# 添加用户角色的消息
messages.append({
    "role": "user",  # 角色为用户
    "content": "Give me a weather report for San Diego, USA."
})

# 使用定义的chat_completion_request函数发起一个请求，传入messages和functions作为参数
chat_response = chat_completion_request(
    messages, functions=functions
)

# 解析返回的JSON数据，获取第一个选项
assistant_message = chat_response.json()["choices"][0]["message"]

# 将助手的回复消息添加到messages列表中
messages.append(assistant_message)

# 打印助手的回复消息
pretty_print_conversation(messages)

强制不使用函数

然后，我们创建另一个消息列表，并添加系统和用户的消息。这次用户请求的是加拿大多伦多当前的天气（使用摄氏度）。随后，代码再次调用chat_completion_request函数，但这次在function_call参数中明确指定了”none”，表示GPT 模型在处理此请求时不能调用任何函数。最后，代码解析返回的JSON响应，获取第一个选项的消息，即 GPT 模型的回应。

# 创建另一个空的消息列表
messages = []

# 添加系统角色的消息
messages.append({
    "role": "system",  # 角色为系统
    "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."
})

# 添加用户角色的消息
messages.append({
    "role": "user",  # 角色为用户
    "content": "Give me the current weather (use Celcius) for Toronto, Canada."
})

# 使用定义的chat_completion_request函数发起一个请求，传入messages、functions和function_call作为参数
chat_response = chat_completion_request(
    messages, functions=functions, function_call="none"
)

# 解析返回的JSON数据，获取第一个选项
assistant_message = chat_response.json()["choices"][0]["message"]

# 将助手的回复消息添加到messages列表中
messages.append(assistant_message)

# 打印助手的回复消息
pretty_print_conversation(messages)

执行GPT模型生成的函数

接着，我们将演示如何执行输入为 GPT 模型生成的函数，并利用这一点来实现一个可以帮助我们回答关于数据库的问题的代理。为了简单起见，我们将使用Chinook样本数据库。

首先，让我们定义一些有用的实用函数来从SQLite数据库中提取数据。

import sqlite3

conn = sqlite3.connect("data/chinook.db")
print("Opened database successfully")

首先定义三个函数get_table_names、get_column_names和get_database_info，用于从数据库连接对象中获取数据库的表名、表的列名以及整体数据库的信息。

def get_table_names(conn):
    """返回一个包含所有表名的列表"""
    table_names = []  # 创建一个空的表名列表
    # 执行SQL查询，获取数据库中所有表的名字
    tables = conn.execute("SELECT name FROM sqlite_master WHERE type='table';")
    # 遍历查询结果，并将每个表名添加到列表中
    for table in tables.fetchall():
        table_names.append(table[0])
    return table_names  # 返回表名列表


def get_column_names(conn, table_name):
    """返回一个给定表的所有列名的列表"""
    column_names = []  # 创建一个空的列名列表
    # 执行SQL查询，获取表的所有列的信息
    columns = conn.execute(f"PRAGMA table_info('{table_name}');").fetchall()
    # 遍历查询结果，并将每个列名添加到列表中
    for col in columns:
        column_names.append(col[1])
    return column_names  # 返回列名列表


def get_database_info(conn):
    """返回一个字典列表，每个字典包含一个表的名字和列信息"""
    table_dicts = []  # 创建一个空的字典列表
    # 遍历数据库中的所有表
    for table_name in get_table_names(conn):
        columns_names = get_column_names(conn, table_name)  # 获取当前表的所有列名
        # 将表名和列名信息作为一个字典添加到列表中
        table_dicts.append({"table_name": table_name, "column_names": columns_names})
    return table_dicts  # 返回字典列表

将数据库信息转换为Python字典类型

# 获取数据库信息，并存储为字典列表
database_schema_dict = get_database_info(conn)

# 将数据库信息转换为字符串格式，方便后续使用
database_schema_string = "\n".join(
    [
        f"Table: {table['table_name']}\nColumns: {', '.join(table['column_names'])}"
        for table in database_schema_dict
    ]
)

database_schema_dict

然后，定义一个函数ask_database。目标是让 GPT 模型帮我们构造一个完整的 SQL 查询。

# 定义一个功能列表，其中包含一个功能字典，该字典定义了一个名为"ask_database"的功能，用于回答用户关于音乐的问题
functions = [
    {
        "name": "ask_database",
        "description": "Use this function to answer user questions about music. Output should be a fully formed SQL query.",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": f"""
                            SQL query extracting info to answer the user's question.
                            SQL should be written using this database schema:
                            {database_schema_string}
                            The query should be returned in plain text, not in JSON.
                            """,
                }
            },
            "required": ["query"],
        },
    }
]

执行 SQL 查询。首先，定义两个函数ask_database和execute_function_call：

前者用于实际执行 SQL 查询并返回结果
后者用于根据消息中的功能调用信息来执行相应的功能并获取结果

def ask_database(conn, query):
    """使用 query 来查询 SQLite 数据库的函数。"""
    try:
        results = str(conn.execute(query).fetchall())  # 执行查询，并将结果转换为字符串
    except Exception as e:  # 如果查询失败，捕获异常并返回错误信息
        results = f"query failed with error: {e}"
    return results  # 返回查询结果


def execute_function_call(message):
    """执行函数调用"""
    # 判断功能调用的名称是否为 "ask_database"
    if message["function_call"]["name"] == "ask_database":
        # 如果是，则获取功能调用的参数，这里是 SQL 查询
        query = json.loads(message["function_call"]["arguments"])["query"]
        # 使用 ask_database 函数执行查询，并获取结果
        results = ask_database(conn, query)
    else:
        # 如果功能调用的名称不是 "ask_database"，则返回错误信息
        results = f"Error: function {message['function_call']['name']} does not exist"
    return results  # 返回结果

然后，创建一个消息列表，并向其中添加了一个系统消息和一个用户消息。系统消息的内容是指示对话的目标，用户消息的内容是用户的问题。接着，使用chat_completion_request函数发出聊天请求并获取响应，然后从响应中提取出助手的消息并添加到消息列表中。如果助手的消息中包含功能调用，那么就使用execute_function_call函数执行这个功能调用并获取结果，然后将结果作为一个功能消息添加到消息列表中。最后，使用pretty_print_conversation函数打印出整个对话。

# 创建一个空的消息列表
messages = []

# 向消息列表中添加一个系统角色的消息，内容是 "Answer user questions by generating SQL queries against the Chinook Music Database."
messages.append({"role": "system", "content": "Answer user questions by generating SQL queries against the Chinook Music Database."})

# 向消息列表中添加一个用户角色的消息，内容是 "Hi, who are the top 5 artists by number of tracks?"
messages.append({"role": "user", "content": "Hi, who are the top 5 artists by number of tracks?"})

# 使用 chat_completion_request 函数获取聊天响应
chat_response = chat_completion_request(messages, functions)

# 从聊天响应中获取助手的消息
assistant_message = chat_response.json()["choices"][0]["message"]

# 将助手的消息添加到消息列表中
messages.append(assistant_message)

# 如果助手的消息中有功能调用
if assistant_message.get("function_call"):
    # 使用 execute_function_call 函数执行功能调用，并获取结果
    results = execute_function_call(assistant_message)
    # 将功能的结果作为一个功能角色的消息添加到消息列表中
    messages.append({"role": "function", "name": assistant_message["function_call"]["name"], "content": results})

# 使用 pretty_print_conversation 函数打印对话
pretty_print_conversation(messages)

# 向消息列表中添加一个用户的问题，内容是 "What is the name of the album with the most tracks?"
messages.append({"role": "user", "content": "What is the name of the album with the most tracks?"})

# 使用 chat_completion_request 函数获取聊天响应
chat_response = chat_completion_request(messages, functions)

# 从聊天响应中获取助手的消息
assistant_message = chat_response.json()["choices"][0]["message"]

# 将助手的消息添加到消息列表中
messages.append(assistant_message)

# 如果助手的消息中有功能调用
if assistant_message.get("function_call"):
    # 使用 execute_function_call 函数执行功能调用，并获取结果
    results = execute_function_call(assistant_message)
    # 将功能的结果作为一个功能角色的消息添加到消息列表中
    messages.append({"role": "function", "content": results, "name": assistant_message["function_call"]["name"]})

# 使用 pretty_print_conversation 函数打印对话
pretty_print_conversation(messages)