需求：有一個問題列表<q.pickle>，需要逐個提問，然后把回答記錄下來。

筆記中所有的API key已刪除，保證Key的安全。

封面圖片來自：開源大語言模型(LLM)匯總（持續(xù)更新中） (yii666.com)

訊飛星火大模型api調(diào)用

以下是調(diào)用實例，官方文檔還有通過web或者Linux調(diào)用的。

import _thread as thread

import base64

import datetime

import hashlib

import hmac

import json

import time

from urllib.parse import urlparse

import ssl

from datetime import datetime

from time import mktime

from urllib.parse import urlencode

from wsgiref.handlers import format_date_time

import sys

import websocket

import pickle



# file = open('訊飛星火導(dǎo)出未分隔的結(jié)果.txt', 'w')

# def custom_print(text):

#     print(text)

#     file.write(text + '\n')

#     file.flush()  # 立即將內(nèi)容寫入文件

#

# # 重定向sys.stdout到自定義的輸出流

# sys.stdout = custom_print

# # sys.stdout = custom_print



# 把輸出的內(nèi)容重新定向到一個文件

# sys.stdout = open('訊飛星火導(dǎo)出未分隔的結(jié)果.txt', 'w')



class Ws_Param(object):

    # 初始化

    def __init__(self, APPID, APIKey, APISecret, gpt_url):

        self.APPID = APPID

        self.APIKey = APIKey

        self.APISecret = APISecret

        self.host = urlparse(gpt_url).netloc

        self.path = urlparse(gpt_url).path

        self.gpt_url = gpt_url



        # 生成url

    def create_url(self):

        # 生成RFC1123格式的時間戳

        now = datetime.now()

        date = format_date_time(mktime(now.timetuple()))



        # 拼接字符串

        signature_origin = "host: " + self.host + "\n"

        signature_origin += "date: " + date + "\n"

        signature_origin += "GET " + self.path + " HTTP/1.1"



        # 進行hmac-sha256進行加密

        signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),

                                 digestmod=hashlib.sha256).digest()



        signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')



        authorization_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'



        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')



        # 將請求的鑒權(quán)參數(shù)組合為字典

        v = {

            "authorization": authorization,

            "date": date,

            "host": self.host

        }

        # 拼接鑒權(quán)參數(shù)，生成url

        url = self.gpt_url + '?' + urlencode(v)

        # 此處打印出建立連接時候的url,參考本demo的時候可取消上方打印的注釋，比對相同參數(shù)時生成的url與自己代碼生成的url是否一致

        return url



        # 收到websocket錯誤的處理

def on_error(ws, error):

    print("### error:", error)



    # 收到websocket關(guān)閉的處理

def on_close(ws):

    print("### closed ###")



    # 收到websocket連接建立的處理

def on_open(ws):

    thread.start_new_thread(run, (ws,))



def run(ws, *args):

    data = json.dumps(gen_params(appid=ws.appid, question=ws.question))

    ws.send(data)



    # 收到websocket消息的處理

def on_message(ws, message):

    # print(message)

    data = json.loads(message)

    code = data['header']['code']

    if code != 0:

        print(f'請求錯誤: {code}, {data}')

        ws.close()

    else:

        choices = data["payload"]["choices"]

        status = choices["status"]

        content = choices["text"][0]["content"]

        print(content, end='')

        if status == 2:

            ws.close()



def gen_params(appid, question):

"""

通過appid和用戶的提問來生成請參數(shù)

"""

data = {

"header": {

"app_id": appid,

"uid": "1234"

},

"parameter": {

"chat": {

"domain": "general",

"random_threshold": 0.5,

"max_tokens": 2048,

"auditing": "default"

}

},

"payload": {

"message": {

"text": [

{"role": "user", "content": question}

]

}

}

}

return data



def main(appid, api_key, api_secret, gpt_url, question):

wsParam = Ws_Param(appid, api_key, api_secret, gpt_url)

websocket.enableTrace(False)

wsUrl = wsParam.create_url()

ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close, on_open=on_open)

ws.appid = appid

ws.question = question

ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})

return



def ask_question(q):

main(appid="",

api_secret="",

api_key="",

gpt_url="ws://spark-api.xf-yun.com/v1.1/chat",

question= q)

print("@")



with open('D:\Desktop\AI與夜曲編程\chatbot及其余內(nèi)容\自己寫的小Bot\q.pickle', 'rb') as f:

questions = pickle.load(f)

for question in questions:

ask_question(question)

time.sleep(5)



# sys.stdout.close()

print("done")

保存：

import csv

def export_text_to_csv(text, filename):

    paragraphs = text.split('@')

    with open(filename, 'w', newline='', encoding='utf-8') as csvfile:

        writer = csv.writer(csvfile)

        for paragraph in paragraphs:

            writer.writerow([paragraph.strip()])  # 每行一個文本，去除前后空格

    print("成功導(dǎo)出為CSV文件：", filename)

export_text_to_csv(text, 'output.csv')

GPT 3.5-0613 和 GPT4 api調(diào)用

流式調(diào)用（GPT4可用）

import openai



openai.api_key = ""

openai.api_base = ""



response = openai.ChatCompletion.create(

    model='gpt-4',

    messages=[

        {'role': 'user', 'content': "樹上八只鳥，打掉一只，還有幾只鳥"},

    ],

    stream=True

)



for chunk in response:

    print(chunk.choices[0].delta.content, end="", flush=True)

GPT3.5調(diào)用

def chat_with_robot(text):

    completion = openai.ChatCompletion.create(

        model = "gpt-3.5-turbo-0613",

        messages=[

            {'role': 'user', 'content': text},

        ],

        temperature=0)

    return completion

def use_gpt_in_jupyter(question):

    clipbord = []

    output = chat_with_robot(question)

    print('GPT3，', output['usage']['total_tokens'], "token used,",round(output['usage']['total_tokens']/1000*0.002, 5), "人民幣 used >>>>>")

    print(output["choices"][0]["message"]["content"])

    pass

with open('questions.pickle', 'rb') as f:

    questions = pickle.load(f)

questions[::10]
new_answers=[]

timer = 0

for question in questions:

    answer = use_gpt(question)

    new_answers.append(answer)

    timer = timer+1

    # time.sleep(30)

    print(f"第 {timer} 個問題已回答，\n【問題】= {question} \n【答案】= {answer}\n")

    time.sleep(25)
df = pd.DataFrame({'Answers': new_answers})

df.to_excel('new_answers.xlsx', index=False)

商湯大模型API調(diào)用

import time

import requests

import json

import pandas as pd

import pickle

url = 'https://lm_experience.sensetime.com/v1/nlp/chat/completions'
def use_shangtang(question):

    data = {

        "messages": [{"role": "user", "content": question}],

        "temperature": 1,

        "top_p": 0.7,

        "max_new_tokens": 1024,

        "repetition_penalty": 1.05,

        "stream": False,

        "user": "test"

    }   # 咱只能2048個token呢，商湯大模型不行啊

    headers = {

        'Content-Type': 'application/json',

        'Authorization': api_secret_key

    }

    response = requests.post(url, headers=headers, json=data)

    raw_response = json.loads(response.text)

    return raw_response["data"]["choices"][0]["message"]
with open('自己寫的小Bot\q.pickle', 'rb') as f:

    questions = pickle.load(f)

questions[::100]

answers=[]

timer = 0

for question in questions:

    answer = use_shangtang(question)

    answers.append(answer)

    timer = timer+1

    # time.sleep(30)

    print(f"第 {timer} 個問題已回答，\n【問題】= {question} \n【答案】= {answer}\n")

    time.sleep(3)

df = pd.DataFrame({'Answers': answers})

df.to_excel('商湯answers.xlsx', index=False)

print("done")

本文章轉(zhuǎn)載微信公眾號@小陸的空間

上一篇：

萬字解讀AI Agent架構(gòu)體系，API和RPA將成為重點

下一篇：

ChatGPT API 定價（成本）：您需要了解的一切
#你可能也喜歡這些API文章!

我們有何不同？

API服務(wù)商零注冊

多API并行試用

數(shù)據(jù)驅(qū)動選型，提升決策效率

查看全部API→
??

熱門場景實測，選對API

#AI文本生成大模型API

對比大模型API的內(nèi)容創(chuàng)意新穎性、情感共鳴力、商業(yè)轉(zhuǎn)化潛力

25個渠道
一鍵對比試用API 限時免費

#AI深度推理大模型API

對比大模型API的邏輯推理準確性、分析深度、可視化建議合理性

10個渠道
一鍵對比試用API 限時免費