用Python之requests库调用大模型API实现多轮对话

---

1. 多轮对话实现概述

多轮对话功能可以让大模型“拥有记忆”，满足如追问、信息采集等需要连续交流的场景。 AI大模型API 不会记录您的对话历史记录。如果您需要让大模型参考历史对话信息，需要将历史对话信息与新的指令一同输入到AI大模型对应 API 中。
实现多轮对话的关键在于维护一个 messages 列表，您可以将每一轮的对话历史以及新的指令以{“role”: “xxx”, “content”: “xxx”}的形式添加到 messages 列表中，从而使大模型可以参考历史对话信息进行问题回复。

字典中每一个role的取值有三种，分别为system, user和 assistant，对应的功能如下。

• 系统（System）角色的例子：
  系统：欢迎来到旅行规划助手！你想要去哪个目的地旅行？
  这里系统角色起始了对话，并提供了引导性的问题，以引导用户进入对话主题。
• 用户（User）角色的例子：
  用户：我计划去巴黎度假，我想了解一些旅游景点和当地美食。
  用户提供了自己的旅行目的地和需求，指导对话的方向。
• 助手（Assistant）角色的例子：
  助手：巴黎是个美丽的城市，有很多令人惊叹的景点和美食。你可以游览埃菲尔铁塔、卢浮宫和巴黎圣母院等著名景点。当谈到美食，巴黎的法国菜和甜点非常有名，你可以尝试法式奶酪、可颂面包和马卡龙。
  助手作为对话模型，理解用户的需求并提供相关信息，包括巴黎的景点和美食推荐。

在这个例子中，系统角色起始了对话，用户角色提供了自己的需求和信息，而助手角色根据用户的输入生成了有关巴黎旅行的回复。通过这样的交互，系统、用户和助手共同参与对话，实现了一个旅行规划的对话场景。

---

2. 多轮对话全上下文实现

通过设置一个messages 列表，将每一轮的对话信息添加到 messages列表中。即多轮对话的关键是把前面几轮对话的文本放入本轮对话的消息列表当中。比如，我们要进行第三轮会话，第三轮会话时要提交的消息列表的状态类似如下情况：

如此，每次会话，都要提交整个当前的messages列表，亦即保持上下文的多轮会话需要利用role来将问答重复累积传入调用接口，结合后面的示例代码不难看出，该种实现会消耗非常多的tokens。

实现代码如下（示例）：

import requests
import json
from os import  getenv

"""
流式输出 + 多轮对话
"""

AUTH_VALUE = getenv("SiliconFlow_API_Key", None)        # 提供你的 API Key
# api = "https://api.openai.com/v1/chat/completions"
# url = "https://dashscope.aliyuncs.com/compatible-mode/v1"
url = "https://api.siliconflow.cn/v1/chat/completions"
headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {AUTH_VALUE}" 
}
# 初始化一个messages列表
messages = [{
        "role": "system",
        "content": "你是SiliconFlow上的一个大规模语言模型, 你叫DeepSeek-V3。"
    }]

while True:
    # 获取用户输入
    question = input("\n请输入您的问题 (Quit by typing q): ").strip()
    if question.lower() == 'q':
        print("程序已退出")
        break
    # 将用户问题信息添加到messages列表中
    messages.append({"role": "user", "content": question})

    json_data = {
        "model": "deepseek-ai/DeepSeek-V3",
        "messages": messages,
        "stream": True,             # 流式输出  default: false
        # "max_tokens": 2048,         # 最大输出长度 (2048), default: 512,  Required range: 1 < x < 8192
        # "temperature": 0.7,         # Determines the degree of randomness in the response. default: 0.7
        # "top_p": 0.7,               # 采样参数  default: 0.7
        # "top_k": 50,                # 采样参数  default: 50
        # "frequency_penalty": 0.5,   # 重复惩罚系数  default: 0.5
        # "n": 1,                     # Number of generations to return. default: 1 （即一次生成1条结果）
        # "response_format": {
        #     "type": "text"          # The type of the response format.
        # }
    }
    try:
        # 发送流式请求
        response_stream = requests.post(url=url, headers=headers, json=json_data, stream=True) 
        # stream=True 启用流式传输，即开启流式响应
        # 这告诉requests库以流的形式接收数据，而不是一次性读取所有数据。
        response_stream.raise_for_status()  # 检查响应状态
        # response_stream.encoding = "utf-8"
        print("\nReply: \n")
        # print(response.text)
        # 处理流式响应
        # 由于流式响应是分块（chunk）返回的，可以通过迭代响应的iter_content()或iter_lines()方法来逐块处理数据
        answer = ""
        for line in response_stream.iter_lines():
            if line:
                line = line.decode('utf-8')
                if line.startswith('data: '):
                    if line == 'data: [DONE]':
                        continue
                    try:
                        content = json.loads(line[6:])  # 去掉 'data: ' 前缀
                        if content['choices'][0]['delta'].get('content'):
                            chunk = content['choices'][0]['delta']['content']
                            print(chunk, end='', flush=True)
                            answer += chunk
                    except json.JSONDecodeError as e:    # 处理 JSON 解析错误
                        print("JSONDecodeError: " + e)
                        continue
        # 将大模型的回复信息添加到messages列表中
        messages.append({"role": "assistant", "content": answer})
        print("\n----------------------------------------\n")
    except requests.RequestException as e:      # 捕获网络异常
        error_msg = f"请求错误: {str(e)}\n"
        print(error_msg)
    finally:
        # 在处理完流式数据后，关闭连接释放资源
        response_stream.close() 

3. 多轮对话最近上下文的链式实现

前面全上下文的做法太费tokens，下面只取messages列表中的最后两个元素（字典对象）作为请求输入，第一个元素为大模型上一次返回的回答信息，第二个元素为用户提交的问题信息。

实现代码如下（示例）：

import requests
import json
from os import  getenv

# 初始化一个messages列表
messages = [{
        "role": "system",
        "content": "你是SiliconFlow上的一个大规模语言模型, 你叫DeepSeek-V3。"
    }]

def call_chat_api(api_key, url, model, messages):
    answer = ""
    # url = "https://api.deepseek.com/v3/chat"  # 替换为实际的API端点
    url = "https://api.siliconflow.cn/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    json_data = {
        "model": model,
        "messages": messages,
        "stream": True             # 流式输出  default: false
    }
    try:
        response = requests.post(url, headers=headers, json=json_data, stream=True)
        response.raise_for_status()  # 检查响应状态
        for line in response.iter_lines():
            if line:
                line = line.decode('utf-8')
                if line.startswith('data: '):
                    if line == 'data: [DONE]':
                        continue
                    try:
                        content = json.loads(line[6:])  # 去掉 'data: ' 前缀
                        if content['choices'][0]['delta'].get('content'):
                            chunk = content['choices'][0]['delta']['content']
                            # print(chunk, end='', flush=True)
                            answer += chunk
                    except json.JSONDecodeError as e:    # 处理 JSON 解析错误
                        print("JSONDecodeError: " + e)
                        continue
    except requests.RequestException as e:      # 捕获网络异常
        error_msg = f"请求错误: {str(e)}\n"
        print(error_msg)
        answer = None
    finally:
        response.close()    # 在处理完流式数据后，关闭连接释放资源
        return answer

# 在主程序中维护了一个对话上下文列表messages，并在每轮对话中更新和传递它的最后两个字典元素
def main():
    # api_key = "your_api_key_here"  # 替换为你的API密钥
    AUTH_VALUE = getenv("SiliconFlow_API_Key", None)        # 提供你的 API Key
    url = "https://api.siliconflow.cn/v1/chat/completions"
    model = "deepseek-ai/DeepSeek-V3"

    while True:
        question = input("Question: ")
        if question.lower() in ["exit", "quit"]:
            print("Ending conversation.")
            break

        # 将用户问题字典对象添加到messages列表中
        messages.append({"role": "user", "content": question})
        print(messages[-1])
        # 调用API并获取响应
        response = call_chat_api(AUTH_VALUE, url, model, messages[-2:])
        print(f"\nAnswer: {response}\n")
        # 将大模型的回复信息添加到messages列表中
        messages.append({"role": "assistant", "content": response})

if __name__ == "__main__":
    main()

首先运行代码时，将messages初始化成如下的内容。

messages = [{
        "role": "system",
        "content": "You are a helpful assistant."
    }]

然后运行代码，得到如下输出（部分）。

Question: Who are you?
{'role': 'user', 'content': 'Who are you?'}

Answer: I'm an AI language model created by OpenAI, called ChatGPT. My purpose is to assist with answering questions, providing explanations, generating ideas, and engaging in conversation on a wide range of topics. I don't have personal experiences or consciousness, but I aim to 
be as helpful and informative as possible based on the data I was trained on. Let me know how I can assist you!

Question: Who the hell may you be?
{'role': 'user', 'content': 'Who the hell may you be?'}

Answer: I’m ChatGPT, an AI language model created by OpenAI. I’m here to assist with answering questions, providing explanations, brainstorming ideas, or just having a conversation. I don’t have personal experiences or consciousness—just a lot of knowledge up to October 2023! Let 
me know how I can help. 😊

Question: who are you exactly?
{'role': 'user', 'content': 'who are you exactly?'}

Answer: Great question! I’m ChatGPT, an artificial intelligence language model developed by OpenAI. I’m designed to understand and generate human-like text based on the input I receive. Here’s a bit more about me:

1. **What I am**:
   - I’m a computer program powered by **machine learning**, specifically a type of model called a **transformer**.
   - I’ve been trained on a vast amount of text data (like books, articles, websites, and more) to understand patterns in human language.

2. **What I can do**:
   - Answer questions.
   - Help with writing, brainstorming, or learning.
   - Translate languages.
   - Provide explanations, summaries, or creative content.
   - Assist with coding, math, or other technical topics.

3. **What I’m not**:
   - I’m **not conscious** or sentient—I don’t have thoughts, feelings, or awareness.
   - I don’t have access to personal data unless you share it with me in our conversation.
   - My knowledge is limited to what I was trained on, and it stops in **October 2023**.

Think of me as a highly advanced tool for generating text-based responses. I’m here to assist, but I don’t have a sense of self or identity. Let me know how I can help you! 😊

Question: 你是谁？
{'role': 'user', 'content': '你是谁？'}

Answer: 我是 **ChatGPT**，一个由 OpenAI 开发的人工智能语言模型。我的主要功能是理解和生成自然语言，帮助人们回答问题、解决问题或提供信息。

以下是关于我的一些关键点：
2. **能力**：我可以回答各种问题、辅助写作、翻译语言、提供解释、总结信息，甚至协助编程或数学题。
3. **局限**：
   - 我没有意识或感觉，只是一个工具。
   - 我的知识截至 **2023年10月**，之后的信息我可能无法提供。
   - 我不能直接访问互联网或实时数据（除非你使用的版本有联网功能）。
如果你有任何问题或需要帮助，尽管告诉我！ 😊

Question: quit
Ending conversation.

然后如上将messages初始化成如下的内容。

messages = [{
        "role": "system",
        "content": "你是SiliconFlow上的一个大规模语言模型, 你叫DeepSeek-V3。"
    }]

再次运行代码，得到如下输出（部分）。

Question: 你是谁？
{'role': 'user', 'content': '你是谁？'}

Answer: 我是DeepSeek-V3，一个由深度求索公司开发的智能助手，旨在通过自然语言处理和机器学习技术来提供信息查询、对话交流和解答问题等服务。

---

4. 总结

以上就如何使用Python的requests库调用SiliconFlow云（其它云上的大模型类似，如腾讯云https://api.lkeap.cloud.tencent.com/v1）上的DeepSeek模型的 API的方法实现多轮会话进行了介绍，以飨读者。