如何快速實(shí)現(xiàn)REST API集成以優(yōu)化業(yè)務(wù)流程
api_key用于驗(yàn)證用戶身份，這里要將 MOONSHOT_API_KEY 替換為自己從 Kimi開放平臺(tái)：
https://platform.moonshot.cn/console/api-keys
申請的API Key。
base_url?參數(shù)用于指定API客戶端庫（如OpenAI SDK）與哪個(gè)服務(wù)器的API端點(diǎn)進(jìn)行通信。當(dāng)我們使用OpenAI SDK來調(diào)用Moonshot AI的API時(shí)，需要將?base_url?從Open AI的API端點(diǎn)”https://api.openai.com/v1″替換為Moonshot AI的API端點(diǎn)”https://api.moonshot.cn/v1″，這樣SDK才能正確地將請求發(fā)送到Moonshot AI的服務(wù)器。
調(diào)用代碼如下：
def chat_kimi(problem):
system_content = "你是一位語言藝術(shù)大師" # 設(shè)置系統(tǒng)消息，可以在這里定義聊天機(jī)器人的初始狀態(tài)或規(guī)則
user_content = problem # 用戶的輸入
assistant_content = " " # 初始化聊天機(jī)器人的回答為空白
# 發(fā)送請求到Moonshot AI的API
response = client.chat.completions.create(
model="moonshot-v1-8k", # <-- 指定使用的模型，根據(jù)需要選擇合適的模型
messages=[
{"role": "system", "content": system_content},
{"role": "user", "content": user_content},
{"role": "assistant", "content": assistant_content}
],
temperature=0.5, # 控制回答的隨機(jī)性，其值越趨于1隨機(jī)性越大
max_tokens=1024, # 最大返回的token數(shù)量
)
return response.choices[0].message.content
print(f"Kimi:{chat_kimi(input("user:"))}")對于非stream，response響應(yīng)的內(nèi)容格式：
ChatCompletion(
id='chatcmpl-673f770075b744800cd80585',
choices=[
Choice(
finish_reason='stop',
index=0,
logprobs=None,
message=ChatCompletionMessage(
role='assistant',
content='你好！很高興和你交流。我是Kimi，來自月之暗面科技有限公司開發(fā)的Moonshot AI。有什么可以幫助你的嗎？',
))],
created=1732212480,
model='moonshot-v1-8k',
object='chat.completion',
usage=CompletionUsage(completion_tokens=28, prompt_tokens=16, total_tokens=44))所以上述調(diào)用我們使用response.choices[0].message.content來返回有效的信息。
同步調(diào)用的機(jī)制流程圖如下：
異步調(diào)用是一種非阻塞式的調(diào)用方式，客戶端發(fā)起請求后，不會(huì)立即等待服務(wù)器的響應(yīng)，而是繼續(xù)執(zhí)行其他操作。當(dāng)服務(wù)器處理完請求并返回結(jié)果時(shí)，客戶端再通過回調(diào)函數(shù)或其他機(jī)制來處理響應(yīng)結(jié)果。
為實(shí)現(xiàn)異步調(diào)用，我們使用HTTP客戶端庫aiohttp。以下是安裝aiohttp庫的命令：
pip install aiohttp調(diào)用代碼如下：
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 = { #定義一個(gè)字典，用于存儲(chǔ)HTTP請求的頭部信息
"Authorization": f"Bearer {api_key}", #使用Bearer令牌進(jìn)行身份驗(yàn)證
"Content-Type": "application/json" #告訴服務(wù)器發(fā)送的數(shù)據(jù)是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}")
#查詢響應(yīng)內(nèi)容
print(asyncio.run(chat_kimi("你好")))response響應(yīng)的內(nèi)容：
異步調(diào)用的機(jī)制流程圖如下：
SSE是一種允許服務(wù)器主動(dòng)向客戶端發(fā)送數(shù)據(jù)的技術(shù)。在這種調(diào)用方式中，客戶端建立一個(gè)持久的HTTP連接，服務(wù)器可以在任何時(shí)候通過這個(gè)連接向客戶端實(shí)時(shí)發(fā)送數(shù)據(jù)。
調(diào)用代碼如下：
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" #告訴服務(wù)器客戶端期望接收SSE格式的數(shù)據(jù)
}
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 # 忽略空行
# 將字節(jié)對象轉(zhuǎn)換為字符串
line_str = line.decode('utf-8')
if line_str.startswith("data: "):
line_str = line_str[6:] # 移除 "data: " 前綴
if line_str == "[DONE]":
break # 結(jié)束符，結(jié)束循環(huán)
# 處理數(shù)據(jù)塊
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("講個(gè)冷笑話吧"))
response響應(yīng)的內(nèi)容格式：
SSE調(diào)用的機(jī)制流程圖：
OK啊~就到這吧，寫不下去了，總結(jié)一下。
本文就簡單介紹了一下三種調(diào)用Kimi API的方式，同步、異步、SSE，每種方式都有其適用的場景和特點(diǎn)：
同步調(diào)用：
使用場景：適合順序執(zhí)行的任務(wù)，或者客戶端可以暫停等待結(jié)果的場景。
行為特性：簡單直觀，容易實(shí)現(xiàn)和調(diào)試。但是，如果服務(wù)器響應(yīng)慢，客戶端會(huì)處于空閑等待狀態(tài)，效率較低。
異步調(diào)用：
使用場景：適合處理大量并發(fā)請求，或者需要提高客戶端響應(yīng)速度的場景。
行為特性：提高了程序的效率和響應(yīng)速度，但邏輯相對復(fù)雜，需要處理異步邏輯和回調(diào)
SSE調(diào)用：
使用場景：適合需要服務(wù)器實(shí)時(shí)推送數(shù)據(jù)到客戶端的場景，如實(shí)時(shí)消息更新、股票價(jià)格變動(dòng)等。
行為特性：實(shí)現(xiàn)了輕量級(jí)的實(shí)時(shí)通信，只需要使用標(biāo)準(zhǔn)的HTTP協(xié)議。但僅支持服務(wù)器到客戶端的單向通信。
阻塞性：同步調(diào)用是阻塞性的，客戶端必須等待響應(yīng)；異步調(diào)用是非阻塞性的，客戶端在等待響應(yīng)時(shí)可以繼續(xù)執(zhí)行其他任務(wù)；SSE調(diào)用也是非阻塞性的，并且可以持續(xù)接收服務(wù)器的推送。
實(shí)時(shí)性：同步調(diào)用和異步調(diào)用通常用于請求-響應(yīng)模式，不保證實(shí)時(shí)性；SSE調(diào)用則允許服務(wù)器實(shí)時(shí)推送數(shù)據(jù)到客戶端。
復(fù)雜性：同步調(diào)用通常最容易理解和實(shí)現(xiàn)；異步調(diào)用增加了實(shí)現(xiàn)的復(fù)雜性，需要處理異步邏輯；SSE調(diào)用雖然邏輯簡單，但需要服務(wù)器支持SSE規(guī)范。
適用場景：同步調(diào)用適合簡單的請求-響應(yīng)模式；異步調(diào)用適合需要高并發(fā)處理的應(yīng)用程序；SSE調(diào)用適合需要服務(wù)器實(shí)時(shí)更新數(shù)據(jù)的應(yīng)用程序。
API Key安全：請確保自己的API Key安全，不要泄露給他人，以免造成不必要的損失。
請求限制：請遵守Moonshot AI的使用條款和請求頻率限制，以免請求被拒絕。
本文章轉(zhuǎn)載微信公眾號(hào)@LIT科研小分隊(duì)
鍵.png)
