如何快速實現REST API集成以優化業務流程
上傳文件的方法如下：
def create_file(file_path):
file = client.files.create(file=open(file_path, "rb"), purpose="assistants")
return fileAssistant[2] 和 File[3] 更多的 API 可以看官方的 API 文檔。
接下來是創建一個 Thread，可以單獨創建，也可以和 Message 一起創建，這里我們連同 Message 一起創建，示例代碼如下：
def create_thread(prompt):
thread = client.beta.threads.create(
messages=[
{
"role": "user",
"content": prompt,
"file_ids": [file.id],
}
]
)
return threadThread 更多的 API 可以看官方的 API 文檔[4]。
然后是創建 Run，創建 Run 時需要指定 Assistant 和 Thread，示例代碼如下：
def run_assistant(thread, assistant):
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
)
return run下面是 Run 的狀態流轉圖：
當創建完了 Run 后，我們需要根據 Run 的 ID 來獲取 Run，查詢其狀態，這是一個重復的過程，下面是查詢 Run 的方法：
def retrieve_run(thread, run):
run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)
return runRun 更多的 API 可以看官方的 API 文檔[5]。
當 Run 運行完成后，我們可以通過獲取 Run 的步驟來查看執行的過程，示例代碼如下：
def list_run_steps(thread, run):
run_steps = client.beta.threads.runs.steps.list(
thread_id=thread.id,
run_id=run.id,
)
return run_stepsRun Step 也有對應的狀態，狀態流轉圖如下所示：
Run Step 狀態比 Run 的狀態要簡單一些，狀態的流轉條件跟 Run 的一樣，這里就不再贅述了。
當 Run 運行完成后，我們還需要獲取 message 的結果，示例代碼如下：
def list_messages(thread):
thread_messages = client.beta.threads.messages.list(thread_id=thread.id)
return thread_messagesMessage 更多的 API 可以看官方的 API 文檔[6]。
下面我們來通過幾個示例來演示下 Assistant API 的具體功能。
我們使用代碼解釋器來解一道方程式，示例代碼如下：
from time import sleep
def code_interpreter():
assistant = create_assistants(
instructions="你是一個數學導師。寫代碼來回答數學問題。",
tools=[{"type": "code_interpreter"}]
)
thread = create_thread(prompt="我需要計算這個方程式的解：3x + 11 = 14。你能幫我嗎？")
run = run_assistant(thread, assistant)
while True:
run = retrieve_run(thread=thread, run=run)
if run.status == "completed":
break
sleep(1)
messages = list_messages(thread)
print(messages.data[0].content[0].text.value)
print(f"messages: {messages.json()}")
run_steps = list_run_steps(thread=thread, run=run)
print(f"run_steps: {run_steps.json()}")你是一個數學導師。寫代碼來回答數學問題。{"type": "code_interpreter"}。運行結果如下：
方程式3x + 11 = 14的解為 x = 1。我們再來看這個 Run 中產生的所有 Messages 信息：
{
"data": [
{
"id": "msg_7cyjjNTgjXOtWlnLOFIeMKW4",
"object": "thread.message",
"role": "assistant",
"content": [
{
"text": {
"annotations": [],
"value": "方程式3x + 11 = 14的解為x = 1。"
},
"type": "text"
}
]
},
{
"id": "msg_xVpdPAd4VOO6Ve5bJ5XoOoiw",
"object": "thread.message",
"role": "user",
"content": [
{
"text": {
"annotations": [],
"value": "我需要計算這個方程式的解：3x + 11 = 14。你能幫我嗎？"
},
"type": "text"
}
]
}
],
"object": "list",
"has_more": false
}總共只有 2 條消息，一條是 user 輸入的問題，另一條是 assistant 返回的結果，中間并沒有工具的消息。我們再來看這個 Run 中的所有 Run Step 信息：
{
"data": [
{
"object": "thread.run.step",
"status": "completed",
"step_details": {
"message_creation": { "message_id": "msg_7cyjjNTgjXOtWlnLOFIeMKW4" },
"type": "message_creation"
},
"type": "message_creation"
},
{
"object": "thread.run.step",
"status": "completed",
"step_details": {
"tool_calls": [
{
"id": "call_mTRfGO52jA6oPLLMACKr5HD5",
"code_interpreter": {
"input": "from sympy import symbols, Eq, solve\r\n\r\n# Define the variable\r\nx = symbols('x')\r\n\r\n# Define the equation\r\nequation = Eq(3*x + 11, 14)\r\n\r\n# Solve the equation\r\nsolution = solve(equation, x)\r\nsolution",
"outputs": [{ "logs": "[1]", "type": "logs" }]
},
"type": "code_interpreter"
}
],
"type": "tool_calls"
},
"type": "tool_calls"
}
],
"object": "list",
"has_more": false
}這個 Run 有 2 個步驟，一個是創建消息，另外一個是代碼解釋器的執行，其中代碼解釋器中執行過程中產生的 input 信息并不會顯示到最終的結果中，只是 LLM 的一個思考過程，類似 LangChain 的 Agent 里面的 debug 信息。
下面我們再使用知識檢索工具來演示一下功能，知識檢索工具需要上傳一個文件，文件通過 API 上傳后，OpenAI 后端會自動分割文檔、embedding、存儲向量，并提供根據用戶問題檢索文檔相關內容的功能，這些都是自動完成的，用戶只需要上傳文檔即可。我們就隨便找一個 pdf 文件來做演示，下面是騰訊云搜的產品文檔：
下面是知識檢索的示例代碼：
def knownledge_retrieve():
file = create_file("tengxunyun.pdf")
assistant = create_assistants(
instructions="你是一個客戶支持機器人，請用你的專業知識回答客戶的問題。",
tools=[{"type": "retrieval"}],
file_ids=[file.id],
)
thread = create_thread(prompt="騰訊云云搜是什么")
run = run_assistant(thread, assistant)
while True:
run = retrieve_run(thread=thread, run=run)
if run.status == "completed":
break
sleep(1)
messages = list_messages(thread)
print(messages.data[0].content[0].text.value)
print(f"messages: {messages.json()}")
run_steps = list_run_steps(thread=thread, run=run)
print(f"run_steps: {run_steps.json()}"){"type": "retrieval"}。運行結果如下：
騰訊云云搜是騰訊云的一站式搜索托管服務平臺，提供數據處理、檢索串識別、搜索結果獲取與排序，搜索數據運營等一整套搜索相關服務。
該平臺繼承了騰訊 SOSO 在搜索引擎領域多年的技術財富，在搜索架構、海量數據存儲和計算、智能排序、用戶意圖識別、搜索質量運營等方面有很深的技術沉淀。
騰訊云云搜負責了騰訊主要產品的搜索業務，包括微信朋友圈、手機 QQ、騰訊視頻、QQ 音樂、應用寶、騰訊地圖、QQ 空間等。
它提供數據處理、用戶檢索串智能識別、排序可定制、高級功能、運營支持等功能，以幫助開發者優化搜索服務，并提供豐富的運營數據查詢功能，
包括檢索量、文檔新增量、檢索耗時、檢索失敗率、熱榜等。開發者可以通過騰訊云云搜讓自己的搜索更具個性化，更匹配應用的需求。【1?source】可以看到答案確實是從文檔中查找而來，內容基本一致，在結尾處還有一個引用【1?source】，這個是 Message 的注釋內容，關于 Message 的注釋功能這里不過多介紹，后面有機會再寫文章說明，關于 Message 注釋功能可以看這里[7]。
我們再來看下知識檢索的 Run Step 信息：
{
"data": [
{
"type": "message_creation"
// ......
},
{
"object": "thread.run.step",
"status": "completed",
"step_details": {
"tool_calls": [
{
"id": "call_19YklOZJq1HDP8WfmMydcVeq",
"retrieval": {},
"type": "retrieval"
}
],
"type": "tool_calls"
},
"thread_id": "thread_ejr0YGodJ2NmG7dFf1hZMPUL",
"type": "tool_calls"
}
]
}在檢索工具的步驟中，只是返回了工具的類型，但檢索的內容并沒有放在步驟中，也就是說檢索工具并沒有產生內部推理過程的信息。
最后我們再來看下自定義工具的示例，我們沿用上一篇文章用到的查詢天氣工具get_current_weather，我們先定義工具集 tools：
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "獲取某個地方當前的天氣情況",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市名，比如：北京, 上海",
},
},
"required": ["location"],
},
},
}
]接著我們使用 Assistant API 來調用自定義工具，示例代碼如下：
def get_current_weather(location):
"""Get the current weather in a given location"""
if "北京" in location.lower():
return json.dumps({"location": location, "temperature": "10°"})
elif "上海" in location.lower():
return json.dumps({"location": location, "temperature": "15°"})
else:
return json.dumps({"location": location, "temperature": "20°"})
def function_calling():
assistant = create_assistants(
instructions="你是一個天氣機器人，使用提供的工具來回答問題。",
tools=tools,
)
thread = create_thread(prompt="今天北京、上海和成都的天氣怎么樣？")
available_functions = {
"get_current_weather": get_current_weather,
}
run = run_assistant(thread, assistant)
# 下面是處理 Run 并提交工具的返回結果
# 中間這部分代碼在下面演示
messages = list_messages(thread)
print(messages.data[0].content[0].text.value)
print(f"messages: {messages}")
run_steps = list_run_steps(thread=thread, run=run)
print(f"run_steps: {run_steps}")available_functions來做函數的動態調用get_current_weather方法使用一些簡單的邏輯來模擬天氣查詢的功能：下面是處理 Run 狀態并提交工具返回結果的方法，代碼如下：
while True:
run = retrieve_run(thread=thread, run=run)
if run.status == "completed":
break
if run.status == "requires_action":
tool_calls = run.required_action.submit_tool_outputs.tool_calls
tool_infos = []
for tool_call in tool_calls:
function_name = tool_call.function.name
function_to_call = available_functions[function_name]
function_args = json.loads(tool_call.function.arguments)
function_response = function_to_call(
location=function_args.get("location"),
)
tool_infos.append(
{"call_id": tool_call.id, "function_response": function_response}
)
submit_tool_outputs(thread=thread, run=run, tool_infos=tool_infos)
sleep(1)get_current_weather方法，因此我們新建了一個數組 tool_infos 來保存工具返回的結果拿到工具返回的結果后，我們通過 Run API 來提交這些結果：
def submit_tool_outputs(thread, run, tool_infos):
tool_outputs = []
for tool_info in tool_infos:
tool_outputs.append(
{
"tool_call_id": tool_info["call_id"],
"output": tool_info["function_response"],
}
)
client.beta.threads.runs.submit_tool_outputs(
thread_id=thread.id,
run_id=run.id,
tool_outputs=tool_outputs,
)運行結果如下：
今天北京的溫度是 10℃，上海的溫度是 15℃，成都的溫度是 20℃。在 Assistant API 的使用過程中，我們發現現在獲取 Run 的狀態都是通過輪詢的方式，這可能會導致更多的 token 損耗和性能問題，OpenAI 官方介紹在未來會對這一機制進行改進，將其替換成通知模式，另外還有其他改進計劃如下：
以上就是 Assistant API 的基本原理和功能介紹，和 GPTs 相比兩者各有優勢，GPTs 更適合沒有編程經驗的用戶使用，而 Assistant API 則適合有開發經驗的開發人員或團隊使用，而且可以使用自定義工具來打造更為強大的功能，但也有一些缺點，就是無法使用 DALL-E 3 畫圖工具和上傳圖片（這也意味著無法做圖片識別），這些功能相信在未來會逐步支持。如果文中有錯誤或者不足之處，還請在評論區留言。
關注我，一起學習各種人工智能和 AIGC 新技術，歡迎交流，如果你有什么想問想說的，歡迎在評論區留言。
?這里:?https://platform.openai.com/docs/assistants/how-it-works/managing-threads-and-messages[1]
這里: https://platform.openai.com/docs/assistants/tools/supported-files[2]
Assistant: https://platform.openai.com/docs/api-reference/assistants[3]
File: https://platform.openai.com/docs/api-reference/files[4]
官方的 API 文檔: https://platform.openai.com/docs/api-reference/threads[5]
官方的 API 文檔: https://platform.openai.com/docs/api-reference/runs[6]
官方的 API 文檔: https://platform.openai.com/docs/api-reference/messages[7]
文章轉自微信公眾號@極客與黑客之路
