云原生 API 网关 APISIX 入门教程
Moonshot 大模型的API调用
作者: weixin03
2024-12-11
这篇文章将介绍如何通过三种不同的方式调用Moonshot AI提供的Kimi API，包括同步调用、异步调用和SSE（Server-Sent Events）调用。Moonshot AI 提供的 Kimi API 兼容 OpenAI 的接口规范，所以我们可以使用 OpenAI 的 SDK 来调用和使用 Kimi 大模型。但由于OpenAI SDK仅支持同步调用，所以后两种这里使用HTTP方式进行操作。
准备工作
创建虚拟环境
conda create --name=moonshot python=3.8版本及以上安装openai库
conda activate moonshotpip install --upgrade 'openai>=1.0'同步调用
同步调用是一种阻塞式的调用方式，客户端发起请求后，会等待服务器处理请求并返回结果。在这个过程中，客户端不会进行其他操作，直到收到服务器的响应。
在开始调用之前我们需要创建一个Moonshot AI客户端，用来与Kimi API进行交互。
from openai import OpenAI
client = OpenAI(
api_key="MOONSHOT_API_KEY",
base_url="https://api.moonshot.cn/v1"
)api_key用于验证用户身份，这里要将 MOONSHOT_API_KEY 替换为自己从 Kimi开放平台：
https://platform.moonshot.cn/console/api-keys
申请的API Key。
base_url 参数用于指定API客户端库（如OpenAI SDK）与哪个服务器的API端点进行通信。当我们使用OpenAI SDK来调用Moonshot AI的API时，需要将 base_url 从Open AI的API端点”https://api.openai.com/v1″替换为Moonshot AI的API端点”https://api.moonshot.cn/v1″，这样SDK才能正确地将请求发送到Moonshot AI的服务器。
调用代码如下：
def chat_kimi(problem):
system_content = "你是一位语言艺术大师" # 设置系统消息，可以在这里定义聊天机器人的初始状态或规则
user_content = problem # 用户的输入
assistant_content = " " # 初始化聊天机器人的回答为空白
# 发送请求到Moonshot AI的API
response = client.chat.completions.create(
model="moonshot-v1-8k", # <-- 指定使用的模型，根据需要选择合适的模型
messages=[
{"role": "system", "content": system_content},
{"role": "user", "content": user_content},
{"role": "assistant", "content": assistant_content}
],
temperature=0.5, # 控制回答的随机性，其值越趋于1随机性越大
max_tokens=1024, # 最大返回的token数量
)
return response.choices[0].message.content
print(f"Kimi:{chat_kimi(input("user:"))}")
对于非stream，response响应的内容格式：
ChatCompletion(
id='chatcmpl-673f770075b744800cd80585',
choices=[
Choice(
finish_reason='stop',
index=0,
logprobs=None,
message=ChatCompletionMessage(
role='assistant',
content='你好！很高兴和你交流。我是Kimi，来自月之暗面科技有限公司开发的Moonshot AI。有什么可以帮助你的吗？',
))],
created=1732212480,
model='moonshot-v1-8k',
object='chat.completion',
usage=CompletionUsage(completion_tokens=28, prompt_tokens=16, total_tokens=44))所以上述调用我们使用response.choices[0].message.content来返回有效的信息。
同步调用的机制流程图如下：
异步调用
异步调用是一种非阻塞式的调用方式，客户端发起请求后，不会立即等待服务器的响应，而是继续执行其他操作。当服务器处理完请求并返回结果时，客户端再通过回调函数或其他机制来处理响应结果。
为实现异步调用，我们使用HTTP客户端库aiohttp。以下是安装aiohttp库的命令：
pip install aiohttp调用代码如下：
import aiohttp
import asyncio
async def chat_kimi(problem):
api_key = "MOONSHOT_API_KEY"
base_url = "https://api.moonshot.cn/v1"
model = "moonshot-v1-8k"
headers = { #定义一个字典，用于存储HTTP请求的头部信息
"Authorization": f"Bearer {api_key}", #使用Bearer令牌进行身份验证
"Content-Type": "application/json" #告诉服务器发送的数据是JSON格式
}
system_content = " "
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_content},
{"role": "user", "content": problem}
],
"temperature": 0.5,
"max_tokens": 1024
}
async with aiohttp.ClientSession() as session:
async with session.post(f"{base_url}/chat/completions", headers=headers, json=payload) as response:
if response.status == 200:
data = await response.json()
return data
else:
raise Exception(f"Request failed with status {response.status}")
#查询响应内容
print(asyncio.run(chat_kimi("你好")))response响应的内容：
异步调用的机制流程图如下：
SSE调用
SSE是一种允许服务器主动向客户端发送数据的技术。在这种调用方式中，客户端建立一个持久的HTTP连接，服务器可以在任何时候通过这个连接向客户端实时发送数据。
调用代码如下：
import aiohttp
import asyncio
import json
async def sse_request(problem):
api_key = "sk-njKietCYs1p8yG7Q87rcgQlrp5yDTZRurZa5abgtkyde1OfI"
base_url = "https://api.moonshot.cn/v1/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"Accept": "text/event-stream" #告诉服务器客户端期望接收SSE格式的数据
}
system_content = " "
data = {
"model": "moonshot-v1-128k",
"messages": [
{"role": "system", "content": system_content},
{"role": "user", "content": problem}
],
"temperature": 0.8,
"stream": True,
}
async with aiohttp.ClientSession() as session:
async with session.post(base_url, headers=headers, json=data) as response:
if response.status != 200:
raise Exception(await response.text())
async for line in response.content:
line = line.strip()
if not line:
continue # 忽略空行
# 将字节对象转换为字符串
line_str = line.decode('utf-8')
if line_str.startswith("data: "):
line_str = line_str[6:] # 移除 "data: " 前缀
if line_str == "[DONE]":
break # 结束符，结束循环
# 处理数据块
try:
chunk = json.loads(line_str)
choice = chunk.get("choices", [{}])[0]
delta = choice.get("delta", {})
content = delta.get("content")
if content:
print(content, end='')
except json.JSONDecodeError:
print("Error decoding JSON:", line_str)asyncio.run(sse_request("讲个冷笑话吧"))
response响应的内容格式：
SSE调用的机制流程图：
OK啊~就到这吧，写不下去了，总结一下。
总结
本文就简单介绍了一下三种调用Kimi API的方式，同步、异步、SSE，每种方式都有其适用的场景和特点：
同步调用：
使用场景：适合顺序执行的任务，或者客户端可以暂停等待结果的场景。
行为特性：简单直观，容易实现和调试。但是，如果服务器响应慢，客户端会处于空闲等待状态，效率较低。
异步调用：
使用场景：适合处理大量并发请求，或者需要提高客户端响应速度的场景。
行为特性：提高了程序的效率和响应速度，但逻辑相对复杂，需要处理异步逻辑和回调
SSE调用：
使用场景：适合需要服务器实时推送数据到客户端的场景，如实时消息更新、股票价格变动等。
行为特性：实现了轻量级的实时通信，只需要使用标准的HTTP协议。但仅支持服务器到客户端的单向通信。
区别
阻塞性：同步调用是阻塞性的，客户端必须等待响应；异步调用是非阻塞性的，客户端在等待响应时可以继续执行其他任务；SSE调用也是非阻塞性的，并且可以持续接收服务器的推送。
实时性：同步调用和异步调用通常用于请求-响应模式，不保证实时性；SSE调用则允许服务器实时推送数据到客户端。
复杂性：同步调用通常最容易理解和实现；异步调用增加了实现的复杂性，需要处理异步逻辑；SSE调用虽然逻辑简单，但需要服务器支持SSE规范。
适用场景：同步调用适合简单的请求-响应模式；异步调用适合需要高并发处理的应用程序；SSE调用适合需要服务器实时更新数据的应用程序。
注意点
API Key安全：请确保自己的API Key安全，不要泄露给他人，以免造成不必要的损失。
请求限制：请遵守Moonshot AI的使用条款和请求频率限制，以免请求被拒绝。
本文章转载微信公众号@LIT科研小分队
🔥