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科研小分隊