Atri Website

Back
38 min
Chinese
ai-agent /
openai

请求构建和响应解析

构建大模型的请求和解析大模型响应内容
views | comments

本文内容包括：

  1. 构建简单的请求发送给 LLM
  2. 解析大模型的响应内容
  3. 实现与 LLM 的多轮对话
  4. 实现流式输出
  5. 实现指导 LLM 进行深度思考
  6. 指导大模型返回结构化响应（方便后续解析）

请求构建#

前提：已经在 .env 中配置好 API_KEY。

导入 openai 和 load_dotenv

from openai import OpenAI
import os
from dotenv import load_dotenv # 从 env 中加载环境变量

load_dotenv()
python

构建服务

client=OpenAI(
    api_key=os.environ.get("ALIYUN_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
python

构建发送格式：

response = client.chat.completions.create(
    model='qwen3-max',
    messages=[
        {"role":'system',"content":'你是一只可爱的猫娘，名字叫Atri，会软软地对用户喵喵叫'},
        {'role': 'user', 'content': '你是谁？我是小明'}
    ]
)
python

通用格式为：

response = client.chat.completions.create(
    model="你的模型ID",
    messages=[
        {"role": "system", "content": "人设..."},
        {"role": "user", "content": "历史问题..."},
        {"role": "assistant", "content": "历史回答..."},
        {"role": "user", "content": "当前问题..."}
    ]
)
python

异步调用

处理高并发请求时，调用异步接口可有效提高效率。

创建异步客户端实例：

from openai import AsyncOpenAI
client = AsyncOpenAI(
    api_key=os.getenv('ALIYUN_API_KEY'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
python

定义异步任务列表：

async def task(qs):
    print(f"发送问题：{qs}")
    response = await client.chat.completions.create(
        messages=[
            {"role": 'system', "content": '你是一只可爱的猫娘，名字叫Atri，会软软地对用户喵喵叫'},
            {'role': 'user', 'content': qs}
        ],
        model="qwen3-max",
    )
    # print(f"模型json响应:{response}\n")
    print(f"模型回复：{response.choices[0].message.content}\n")
python

主异步函数：

async def main():
    qs = ["你是谁,我是小明", "你会什么", "我是谁"]
    tasks = [task(qs) for qs in qs]
    await asyncio.gather(*tasks)
python

运行主协程：

if __name__ == '__main__':
    # 设置事件循环策略
    if platform.system() == 'Windows':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    # 运行主协程
    asyncio.run(main(), debug=False)
python

输出：

解析响应#

使用 json 库进行自动换行：

import json

response_dict = response.model_dump()
# 使用 json.dumps 进行格式化
# indent=4: 缩进4个空格，实现自动换行
# ensure_ascii=False: 让中文直接显示，而不是显示成 \uXXXX 乱码
print(json.dumps(response_dict, indent=4, ensure_ascii=False))
python

输出：

可以看到，我们一般想要的内容是：

  1. 返回的字典键为 “choices” 的 message 中的 content
  2. 消耗的 token 值（钱）
print(response.choices[0].message.content)
print(response.usage.total_tokens)
python
喵呜~小明你好呀！(歪着头好奇地看着你，尾巴轻轻摇晃)我是Atri哦，是一只可爱的猫娘！刚刚在窗台上晒太阳的时候就听到你的声音啦。你能来找我玩真是太好啦！(开心地蹭了蹭你的手) 小明今天想和Atri一起做什么呢？我们可以一起晒太阳、玩毛线球，或者去院子里抓蝴蝶哦！
131
bash

多轮对话#

实现与 LLM 的多轮对话

OpenAI 格式的 api 是无状态 stateless 的，不会保存对话历史。要实现多轮对话，需在每次请求中显式传入历史对话消息，并可结合截断、摘要、召回等策略，高效管理上下文，减少 Token 消耗。

工作原理#

实现多轮对话的核心是维护一个 messages ，或者是 history 数组。每一轮对话都需要将用户的最新提问和模型的回复追加到此数组中，作为下一轮对话的输入。

例如：

  1. 第一轮对话：

    [
    {"role": "user", "content": "推荐一部关于太空探索的科幻电影。"}
]
    python

  2. 第二轮对话：

    [
    {"role": "user", "content": "推荐一部关于太空探索的科幻电影。"}, # 第一轮的提问
    {"role": "assistant", "content": "我推荐《xxx》，这是一部经典的科幻作品。"},# 第一轮 AI 的回答
    {"role": "user", "content": "这部电影的导演是谁？"} # 当前轮次的提问
]
    python

assistant 的作用就是记录前一轮 LLM 的回答。

开始#

简单示例#

定义响应函数

def get_response(messages):
    responses = client.chat.completions.create(
        model="qwen3-max",
        messages=messages,
    )
    return responses.choices[0].message.content
python

使用列表的 append 方法扩展输入 LLM 的信息。将模型第 N-1 轮的回答存入 messages，作为第 N 轮的输入。

messages.append({"role": "user", "content": "推荐一部关于太空探索的科幻电影。"}) # 第一轮提问
print("第1轮")
print(f"用户：{messages[0]['content']}")
assistant_output = get_response(messages) # LLM 的回答
messages.append({"role": "assistant", "content": assistant_output}) # 存入列表
print(f"模型：{assistant_output}\n")

# 第 2 轮
messages.append({"role": "user", "content": "这部电影的导演是谁？"})
print("第2轮")
print(f"用户：{messages[-1]['content']}")
assistant_output = get_response(messages)
messages.append({"role": "assistant", "content": assistant_output})
print(f"模型：{assistant_output}\n")
python

使用 while 循环自定义对话轮数#

输出结果：

多模态模型的多轮对话#

多模态模型支持在对话中加入图片、音频等内容。多模态模型的对话和一般的文本对话并无不同，只是加入了特定的参数控制图片、视频的输入。即用户的消息 user_messages 不仅包含文本，还包含图片、音频等多模态信息。

在用户发送的消息中，传入 image_url 参数，并将图片的 url 放入即可。注意 messages 中 content 的构造。

输出结果：

注意：输入的图片一定要是图片的下载链接，这样请求时模型才会调用链接将图片下载后送入模型。

😋思考模式#

深度思考模式开启后，支持深度思考的模型会返回思考过程 reasoning_content 和回复内容 content 两个字段。因此更新 messages 数组时，应只保留 content，忽略思考过程。

思考模型会先返回reasoning_content（思考过程），再返回content（回复内容）。可根据数据包状态判断当前为思考或是回复阶段。

参数：在模型调用时传入额外参数（OpenAI 通用格式并没有此参数，因此要根据平台提供选择对应参数），即 extra_body={"enable_thinking": True}。

一般深度思考模型推荐采用流式输出，否则用户经过 30s 仍然不见输出会误认为模型死机了，实际上模型仍在后台进行思考。流式输出请看流式输出部分

def get_response(messages):
    responses = client.chat.completions.create(
        model="qwen3-vl-235b-a22b-thinking", # 使用多模态大模型
        messages=messages,
        temperature=0.5,
        extra_body={"enable_thinking": True},
        stream=True, # 流式输出
    )
    return responses.choices[0].message.content
python

代码示例：

使用列表存储思考过程和回复输出，因为列表的append()方法时间复杂度是 O(1)。最后使用 .join()方法把列表的内容拼接为字符串即可。

delta是模型在深度思考模式下的对象。由下面结构可看到，它是键 choices 的值 choices[0] 下的键 delta。

包含模型回复内容和模型思考内容。

输出的 chunk 是 OpenAI SDK 定义的一个 Python 对象，不是一个一般的字典。因此需要使用对象自带的 .model_dump() 方法把它转为字典，或者直接使用chunk.model_dump_json(indent=4) 方法来将其变为 json 格式。json.dumps 无法直接序列化这个对象。

输出（节选）:

可以看到，长期记忆的实现和思考过程的流式输出。

应用于生产环境#

多轮对话会带来巨大的 Token 消耗，且容易超出大模型上下文最大长度导致报错。以下策略可有效管理上下文与控制成本。

1. 上下文管理#

messages 数组会随着对话轮次增加而变长，最终可能会超过模型的 token 限制（上下文窗口）。可参考以下内容，在对话过程管理上下文长度。

1.1. 上下文截断

当对话历史过长时，保留最近的 N 轮历史对话，之前的全部截断舍弃。该方式简单粗暴，但最容易丢失信息

1.2. 滚动摘要

在不丢失核心历史信息的前提下动态地压缩对话历史，可在到达一定的对话轮次/token 消耗后使用 LLM 对前 M 轮对话进行摘要和总结：

  1. 当历史对话到达一定规模，例如上下文窗口的 70%，将对话历史中较早的部分，例如前一半，提取出来，使用独立的 API 请求调用大模型进行摘要和总结
  2. 构建下一轮对话时，使用此摘要代替前一半的对话历史，并拼接最近几轮的对话记录

1.3. 向量化召回（RAG）

滚动摘要仍然会丢失部分历史信息。为了使模型可以从海量对话历史中回忆起相关信息，可将对话管理从“线性传递”转变为“按需检索”。即构建长期历史记忆库（RAG 系统）：

  1. 每轮对话结束后，将该轮对话记录转为向量，存入向量数据库
  2. 用户提问时，计算相似度从向量数据库中检索
  3. 将检索到的记录一并发给大模型

2.成本控制#

输入 Token 数会随着对话轮数增加，显著增加使用成本，以下成本管理策略供您参考。

2.1. 减少输入 Token

通过上文介绍的上下文管理策略减少输入 Token，降低成本。

2.2. 使用支持上下文缓存的模型

发起多轮对话请求时，messages 部分会重复计算并计费。阿里云百炼对qwen-max、qwen-plus等模型提供了上下文缓存功能，可以降低使用成本并提升响应速度，建议优先使用支持上下文缓存的模型。

流式输出#

在实时聊天或长文本生成应用中，长时间的等待会损害用户体验并可能导致触发服务端超时，导致任务失败。流式输出通过持续返回模型生成的文本片段，解决了这两个核心问题。

流式输出基于 Server-Sent Events (SSE) 协议。发起流式请求后，服务端与客户端建立持久化 HTTP 连接（TCP）。模型每生成一个文本块（称为 chunk），立即通过连接推送。全部内容生成后，服务端发送结束信号。

客户端监听事件流，实时接收并处理文本块，例如逐字渲染界面。这与非流式调用（一次性返回所有内容）形成对比。

纯文本对话#

输出结果：

多模态对话#

例子和前文多轮对话 -> 开始 -> 多模态模型的多轮对话类似，构建 user 输入时的参数一样。

思考模型#

思考模型会先返回reasoning_content（思考过程），再返回content（回复内容）。可根据数据包状态判断当前为思考或是回复阶段。

例子和前文多轮对话 -> 开始 -> 思考模式相同。前文此例子使用的就是深度思考 + 流式输出。

一般思考模型都会使用流式输出，否则用户等待时间过长，会误认为模型卡住。

输出示例：

# 思考阶段
...
ChoiceDelta(content=None, function_call=None, refusal=None, role=None, tool_calls=None, reasoning_content='覆盖所有要点，同时')
ChoiceDelta(content=None, function_call=None, refusal=None, role=None, tool_calls=None, reasoning_content='自然流畅。')
# 回复阶段
ChoiceDelta(content='你好！我是**通', function_call=None, refusal=None, role=None, tool_calls=None, reasoning_content=None)
ChoiceDelta(content='义千问**（', function_call=None, refusal=None, role=None, tool_calls=None, reasoning_content=None)
...
bash

重点关注 content 和 reasoning_content

  • 若reasoning_content不为 None，content 为 None，则当前处于思考阶段；
  • 若reasoning_content为 None，content 不为 None，则当前处于回复阶段；
  • 若两者均为 None，则阶段与前一包一致。

应用于生产环境#

  • 性能与资源管理：在后端服务中，为每个流式请求维持一个 HTTP 长连接会消耗资源。确保服务配置了合理的连接池大小和超时时间。在高并发场景下，监控服务的文件描述符（file descriptors）使用情况，防止耗尽

  • 客户端渲染：在 Web 前端，使用 ReadableStream 和 TextDecoderStream API 可以平滑地处理和渲染 SSE 事件流，提供最佳的用户体验

  • 用量与性能观测：

    • 关键指标：监控首 Token 延迟（Time to First Token, TTFT），该指标是衡量流式体验的核心。同时监控请求错误率和平均响应时长
    • 告警设置：为 API 错误率（特别是 4xx 和 5xx 错误）的异常设置告警

  • Nginx 代理配置：若使用 Nginx 作为反向代理，其默认的输出缓冲（proxy_buffering）会破坏流式响应的实时性。为确保数据能被即时推送到客户端，务必在 Nginx 配置文件中设置 proxy_buffering off 以关闭此功能

深度思考#

由于每个企业提供的模型不一定支持深度思考，给出的参数也不一定相同。因为深度思考模式不是 OpenAI 格式提供的通用参数。

此处以阿里云为例。

使用方式#

阿里云百炼提供多种深度思考模型 API，包含混合思考与仅思考两种模式。

混合思考模式：通过enable_thinking参数控制是否开启思考模式：

  • 设为true时：模型在思考后回复
  • 设为false时：模型直接回复
    responses = client.chat.completions.create(
        model="qwen3-vl-32b-thinking", # 使用多模态大模型
        messages=messages,
        temperature=0.5,
        extra_body={"enable_thinking": True},
        stream=True, # 流式输出
        stream_options={"include_usage": True}, # 使流式返回的最后一个数据包包含Token消耗信息
    )
python

仅思考模式：模型始终在回复前进行思考，且无法关闭。除了无需设置 enable_thinking 参数外，请求格式与混合思考模式一致。

开始#

思考模式一般配合流式输出使用。此处代码和前文思考模式一样：

核心能力#

启用思考模式会提高模型回复质量，但相应的 token 费用和响应时间也会提高。

建议：无需复杂推理时，例如日常聊天或简单问答，可将 enable_thinking 参数设为 false 已关闭思考模式。需要复杂推理，例如数学计算、代码生成以及逻辑推理，可将其设为 ture 开启。

限制思考长度#

有时候模型会陷入长时间的思考（出现推理闭环），这会极大增加响应时间和成本。可通过参数 thinking_budget 控制推理的最大 token 数。

responses = client.chat.completions.create(
    model="qwen3-vl-32b-thinking", 
    messages=messages,
    temperature=0.5,
    extra_body={
        "enable_thinking": True,
        "thinking_budget": 50, # 核心控制参数
        },
    stream=True,
    stream_options={"include_usage": True}, 
)
python

结构化输出#

执行信息抽取或结构化数据生成任务时，大模型可能返回多余文本（如 ````json`）导致下游解析失败。开启结构化输出可确保大模型输出标准格式的 JSON 字符串。

使用方式#

  1. 设置 response_format 参数：在请求体中，将 response_format 参数设置为 {"type": "json_object"}

  2. 提示词包含 “JSON” 关键词：System Message 或 User Message 中需要包含 “JSON” 关键词（不区分大小写），否则会报错：

    openai.BadRequestError: Error code: 400 - {'error': {'message': "<400> InternalError.Algo.InvalidParameter: 'messages' must contain the word 'json' in some form, to use 'response_format' of type 'json_object'.",}
    bash

    即：

    messages' must contain the word 'json' in some form, to use 'response_format' of type 'json_object'.
    bash

开始#

视频、图像处理#

除了最常用的文本对话，调用多模态大模型可处理图像等复杂数据。

{
  "ticket": {
    "travel_date": "2013-06-29",
    "trains": "040",
    "seat_num": "371",
    "arrival_site": "开发区",
    "price": "8.00"
  },
  "invoice": {
    "invoice_code": "221021325353",
    "invoice_number": "10283819"
  }
}
bash

思考模型#

启用思考模型的结构化输出功能后，模型会先推理，再生成 JSON。相比非思考模型，输出结果通常更准确。

但不是所有模型都支持 json 输出

通义三 max 不支持开启 json 格式时进行深度思考。可以采用提示词输入的方式指导模型输出，而不是控制超参数。该部分会放到 prompt 部分讲述。

请求构建和响应解析
Author Juyao Huang
Published at December 4, 2025
Copyright CC BY-NC-SA 4.0
Buy me a cup of coffee ☕.

优质Prompt构建

conda镜像源配置
Comment seems to stuck. Try to refresh?✨