使用 Kimi API 的流式输出功能 —— Streaming

from openai import OpenAI
 
client = OpenAI(
    api_key = "MOONSHOT_API_KEY", # 在这里将 MOONSHOT_API_KEY 替换为你从 Kimi 开放平台申请的 API Key
    base_url = "https://api.moonshot.cn/v1",
)
 
stream = client.chat.completions.create(
    model = "kimi-k2.5",
    messages = [
        {"role": "system", "content": "你是 Kimi，由 Moonshot AI 提供的人工智能助手，你更擅长中文和英文的对话。你会为用户提供安全，有帮助，准确的回答。同时，你会拒绝一切涉及恐怖主义，种族歧视，黄色暴力等问题的回答。Moonshot AI 为专有名词，不可翻译成其他语言。"},
        {"role": "user", "content": "你好，我叫李雷，1+1等于多少？"}
    ],
    stream=True, # <-- 注意这里，我们通过设置 stream=True 开启流式输出模式
)
 
# 当启用流式输出模式（stream=True），SDK 返回的内容也发生了变化，我们不再直接访问返回值中的 choice
# 而是通过 for 循环逐个访问返回值中每个单独的块（chunk）
 
for chunk in stream:
	# 在这里，每个 chunk 的结构都与之前的 completion 相似，但 message 字段被替换成了 delta 字段
	delta = chunk.choices[0].delta # <-- message 字段被替换成了 delta 字段
 
	if delta.content:
		# 我们在打印内容时，由于是流式输出，为了保证句子的连贯性，我们不人为地添加
		# 换行符，因此通过设置 end="" 来取消 print 自带的换行符。
		print(delta.content, end="")

import os
import httpx
from openai import OpenAI
 
client = OpenAI(
    api_key = "MOONSHOT_API_KEY", # 在这里将 MOONSHOT_API_KEY 替换为你从 Kimi 开放平台申请的 API Key
    base_url = "https://api.moonshot.cn/v1",
)
 
stream = client.chat.completions.create(
    model = "kimi-k2.5",
    messages = [
        {"role": "system", "content": "你是 Kimi，由 Moonshot AI 提供的人工智能助手，你更擅长中文和英文的对话。你会为用户提供安全，有帮助，准确的回答。同时，你会拒绝一切涉及恐怖主义，种族歧视，黄色暴力等问题的回答。Moonshot AI 为专有名词，不可翻译成其他语言。"},
        {"role": "user", "content": "你好，我叫李雷，1+1等于多少？"}
    ],
    stream=True, # <-- 注意这里，我们通过设置 stream=True 开启流式输出模式
)
 
 
def estimate_token_count(input: str) -> int:
    """
    在这里实现你的 Tokens 计算逻辑，或是直接调用我们的 Tokens 计算接口计算 Tokens
 
    https://api.moonshot.cn/v1/tokenizers/estimate-token-count
    """
    header = {
        "Authorization": f"Bearer {os.environ['MOONSHOT_API_KEY']}",
    }
    data = {
        "model": "kimi-k2.5",
        "messages": [
            {"role": "user", "content": input},
        ]
    }
    r = httpx.post("https://api.moonshot.cn/v1/tokenizers/estimate-token-count", headers=header, json=data)
    r.raise_for_status()
    return r.json()["data"]["total_tokens"]
 
 
completion = []
for chunk in stream:
	delta = chunk.choices[0].delta
	if delta.content:
		completion.append(delta.content)
 
 
print("completion_tokens:", estimate_token_count("".join(completion)))

import httpx # 我们使用 httpx 库来执行我们的 HTTP 请求
 
 
data = {
	"model": "kimi-k2.5",
	"messages": [
		# 具体的 messages
	],
	"stream": True,
}
 
 
# 使用 httpx 向 Kimi 大模型发出 chat 请求，并获得响应 r
r = httpx.post("https://api.moonshot.cn/v1/chat/completions", json=data)
if r.status_code != 200:
	raise Exception(r.text)
 
 
data: str
 
# 在这里，我们使用了 iter_lines 方法来逐行读取响应体
for line in r.iter_lines():
	# 去除每一行收尾的空格，以便更好地处理数据块
	line = line.strip()
 
	# 接下来我们要处理三种不同的情况：
	#   1. 如果当前行是空行，则表明前一个数据块已接收完毕（即前文提到的，通过两个换行符结束数据块传输），我们可以对该数据块进行反序列化，并打印出对应的 content 内容；
	#   2. 如果当前行为非空行，且以 data: 开头，则表明这是一个数据块传输的开始，我们去除 data: 前缀后，首先判断是否是结束符 [DONE]，如果不是，将数据内容保存到 data 变量；
	#   3. 如果当前行为非空行，但不以 data: 开头，则表明当前行仍然归属上一个正在传输的数据块，我们将当前行的内容追加到 data 变量尾部；
 
	if len(line) == 0:
		chunk = json.loads(data)
 
		# 这里的处理逻辑可以替换成你的业务逻辑，打印仅是为了展示处理流程
		choice = chunk["choices"][0]
		usage = choice.get("usage")
		if usage:
			print("total_tokens:", usage["total_tokens"])
		delta = choice["delta"]
		role = delta.get("role")
		if role:
			print("role:", role)
		content = delta.get("content")
		if content:
			print(content, end="")
 
		data = "" # 重置 data
	elif line.startswith("data: "):
		data = line.lstrip("data: ")
 
		# 当数据块内容为 [DONE] 时，则表明所有数据块已发送完毕，可断开网络连接
		if data == "[DONE]":
			break
	else:
		data = data + "\n" + line # 我们仍然在追加内容时，为其添加一个换行符，因为这可能是该数据块有意将数据分行展示

import httpx # 我们使用 httpx 库来执行我们的 HTTP 请求
 
 
data = {
	"model": "kimi-k2.5",
	"messages": [
		# 具体的 messages
	],
	"stream": True,
	"n": 2, # <-- 注意这里，我们要求 Kimi 大模型输出 2 个回复
}
 
 
# 使用 httpx 向 Kimi 大模型发出 chat 请求，并获得响应 r
r = httpx.post("https://api.moonshot.cn/v1/chat/completions", json=data)
if r.status_code != 200:
	raise Exception(r.text)
 
 
data: str
 
# 在这里，我们预先构建一个 List，用于存放不同的回复消息，由于我们设置了 n=2，因此我们将 List 初始化为 2 个元素
messages = [{}, {}]
 
# 在这里，我们使用了 iter_lines 方法来逐行读取响应体
for line in r.iter_lines():
	# 去除每一行收尾的空格，以便更好地处理数据块
	line = line.strip()
 
	# 接下来我们要处理三种不同的情况：
	#   1. 如果当前行是空行，则表明前一个数据块已接收完毕（即前文提到的，通过两个换行符结束数据块传输），我们可以对该数据块进行反序列化，并打印出对应的 content 内容；
	#   2. 如果当前行为非空行，且以 data: 开头，则表明这是一个数据块传输的开始，我们去除 data: 前缀后，首先判断是否是结束符 [DONE]，如果不是，将数据内容保存到 data 变量；
	#   3. 如果当前行为非空行，但不以 data: 开头，则表明当前行仍然归属上一个正在传输的数据块，我们将当前行的内容追加到 data 变量尾部；
 
	if len(line) == 0:
		chunk = json.loads(data)
 
		# 通过循环获取每个数据块中所有的 choice，并获取 index 对应的 message 对象
		for choice in chunk["choices"]:
			index = choice["index"]
			message = messages[index]
			usage = choice.get("usage")
			if usage:
				message["usage"] = usage
			delta = choice["delta"]
			role = delta.get("role")
			if role:
				message["role"] = role
			content = delta.get("content")
			if content:
				message["content"] = message["content"] + content
 
			data = "" # 重置 data
	elif line.startswith("data: "):
		data = line.lstrip("data: ")
 
		# 当数据块内容为 [DONE] 时，则表明所有数据块已发送完毕，可断开网络连接
		if data == "[DONE]":
			break
	else:
		data = data + "\n" + line # 我们仍然在追加内容时，为其添加一个换行符，因为这可能是该数据块有意将数据分行展示
 
 
# 在组装完所有 messages 后，我们分别打印其内容
for index, message in enumerate(messages):
	print("index:", index)
	print("message:", json.dumps(message, ensure_ascii=False))