
如何快速實(shí)現(xiàn)REST API集成以優(yōu)化業(yè)務(wù)流程
上傳文件的方法如下:
def create_file(file_path):
file = client.files.create(file=open(file_path, "rb"), purpose="assistants")
return file
Assistant[2] 和 File[3] 更多的 API 可以看官方的 API 文檔。
接下來(lái)是創(chuàng)建一個(gè) Thread,可以單獨(dú)創(chuàng)建,也可以和 Message 一起創(chuàng)建,這里我們連同 Message 一起創(chuàng)建,示例代碼如下:
def create_thread(prompt):
thread = client.beta.threads.create(
messages=[
{
"role": "user",
"content": prompt,
"file_ids": [file.id],
}
]
)
return thread
Thread 更多的 API 可以看官方的 API 文檔[4]。
然后是創(chuàng)建 Run,創(chuàng)建 Run 時(shí)需要指定 Assistant 和 Thread,示例代碼如下:
def run_assistant(thread, assistant):
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
)
return run
下面是 Run 的狀態(tài)流轉(zhuǎn)圖:
當(dāng)創(chuàng)建完了 Run 后,我們需要根據(jù) Run 的 ID 來(lái)獲取 Run,查詢其狀態(tài),這是一個(gè)重復(fù)的過(guò)程,下面是查詢 Run 的方法:
def retrieve_run(thread, run):
run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)
return run
Run 更多的 API 可以看官方的 API 文檔[5]。
當(dāng) Run 運(yùn)行完成后,我們可以通過(guò)獲取 Run 的步驟來(lái)查看執(zhí)行的過(guò)程,示例代碼如下:
def list_run_steps(thread, run):
run_steps = client.beta.threads.runs.steps.list(
thread_id=thread.id,
run_id=run.id,
)
return run_steps
Run Step 也有對(duì)應(yīng)的狀態(tài),狀態(tài)流轉(zhuǎn)圖如下所示:
Run Step 狀態(tài)比 Run 的狀態(tài)要簡(jiǎn)單一些,狀態(tài)的流轉(zhuǎn)條件跟 Run 的一樣,這里就不再贅述了。
當(dāng) Run 運(yùn)行完成后,我們還需要獲取 message 的結(jié)果,示例代碼如下:
def list_messages(thread):
thread_messages = client.beta.threads.messages.list(thread_id=thread.id)
return thread_messages
Message 更多的 API 可以看官方的 API 文檔[6]。
下面我們來(lái)通過(guò)幾個(gè)示例來(lái)演示下 Assistant API 的具體功能。
我們使用代碼解釋器來(lái)解一道方程式,示例代碼如下:
from time import sleep
def code_interpreter():
assistant = create_assistants(
instructions="你是一個(gè)數(shù)學(xué)導(dǎo)師。寫代碼來(lái)回答數(shù)學(xué)問(wèn)題。",
tools=[{"type": "code_interpreter"}]
)
thread = create_thread(prompt="我需要計(jì)算這個(gè)方程式的解: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()}")
你是一個(gè)數(shù)學(xué)導(dǎo)師。寫代碼來(lái)回答數(shù)學(xué)問(wèn)題
。{"type": "code_interpreter"}
。運(yùn)行結(jié)果如下:
方程式3x + 11 = 14
的解為 x = 1。
我們?cè)賮?lái)看這個(gè) Run 中產(chǎn)生的所有 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": "我需要計(jì)算這個(gè)方程式的解:3x + 11 = 14
。你能幫我嗎?"
},
"type": "text"
}
]
}
],
"object": "list",
"has_more": false
}
總共只有 2 條消息,一條是 user 輸入的問(wèn)題,另一條是 assistant 返回的結(jié)果,中間并沒(méi)有工具的消息。我們?cè)賮?lái)看這個(gè) 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
}
這個(gè) Run 有 2 個(gè)步驟,一個(gè)是創(chuàng)建消息,另外一個(gè)是代碼解釋器的執(zhí)行,其中代碼解釋器中執(zhí)行過(guò)程中產(chǎn)生的 input 信息并不會(huì)顯示到最終的結(jié)果中,只是 LLM 的一個(gè)思考過(guò)程,類似 LangChain 的 Agent 里面的 debug 信息。
下面我們?cè)偈褂弥R(shí)檢索工具來(lái)演示一下功能,知識(shí)檢索工具需要上傳一個(gè)文件,文件通過(guò) API 上傳后,OpenAI 后端會(huì)自動(dòng)分割文檔、embedding、存儲(chǔ)向量,并提供根據(jù)用戶問(wèn)題檢索文檔相關(guān)內(nèi)容的功能,這些都是自動(dòng)完成的,用戶只需要上傳文檔即可。我們就隨便找一個(gè) pdf 文件來(lái)做演示,下面是騰訊云搜的產(chǎn)品文檔:
下面是知識(shí)檢索的示例代碼:
def knownledge_retrieve():
file = create_file("tengxunyun.pdf")
assistant = create_assistants(
instructions="你是一個(gè)客戶支持機(jī)器人,請(qǐng)用你的專業(yè)知識(shí)回答客戶的問(wèn)題。",
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"}
。運(yùn)行結(jié)果如下:
騰訊云云搜是騰訊云的一站式搜索托管服務(wù)平臺(tái),提供數(shù)據(jù)處理、檢索串識(shí)別、搜索結(jié)果獲取與排序,搜索數(shù)據(jù)運(yùn)營(yíng)等一整套搜索相關(guān)服務(wù)。
該平臺(tái)繼承了騰訊 SOSO 在搜索引擎領(lǐng)域多年的技術(shù)財(cái)富,在搜索架構(gòu)、海量數(shù)據(jù)存儲(chǔ)和計(jì)算、智能排序、用戶意圖識(shí)別、搜索質(zhì)量運(yùn)營(yíng)等方面有很深的技術(shù)沉淀。
騰訊云云搜負(fù)責(zé)了騰訊主要產(chǎn)品的搜索業(yè)務(wù),包括微信朋友圈、手機(jī) QQ、騰訊視頻、QQ 音樂(lè)、應(yīng)用寶、騰訊地圖、QQ 空間等。
它提供數(shù)據(jù)處理、用戶檢索串智能識(shí)別、排序可定制、高級(jí)功能、運(yùn)營(yíng)支持等功能,以幫助開(kāi)發(fā)者優(yōu)化搜索服務(wù),并提供豐富的運(yùn)營(yíng)數(shù)據(jù)查詢功能,
包括檢索量、文檔新增量、檢索耗時(shí)、檢索失敗率、熱榜等。開(kāi)發(fā)者可以通過(guò)騰訊云云搜讓自己的搜索更具個(gè)性化,更匹配應(yīng)用的需求?!??source】
可以看到答案確實(shí)是從文檔中查找而來(lái),內(nèi)容基本一致,在結(jié)尾處還有一個(gè)引用【1?source】
,這個(gè)是 Message 的注釋內(nèi)容,關(guān)于 Message 的注釋功能這里不過(guò)多介紹,后面有機(jī)會(huì)再寫文章說(shuō)明,關(guān)于 Message 注釋功能可以看這里[7]。
我們?cè)賮?lái)看下知識(shí)檢索的 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"
}
]
}
在檢索工具的步驟中,只是返回了工具的類型,但檢索的內(nèi)容并沒(méi)有放在步驟中,也就是說(shuō)檢索工具并沒(méi)有產(chǎn)生內(nèi)部推理過(guò)程的信息。
最后我們?cè)賮?lái)看下自定義工具的示例,我們沿用上一篇文章用到的查詢天氣工具get_current_weather
,我們先定義工具集 tools:
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "獲取某個(gè)地方當(dāng)前的天氣情況",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市名,比如:北京, 上海",
},
},
"required": ["location"],
},
},
}
]
接著我們使用 Assistant API 來(lái)調(diào)用自定義工具,示例代碼如下:
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="你是一個(gè)天氣機(jī)器人,使用提供的工具來(lái)回答問(wèn)題。",
tools=tools,
)
thread = create_thread(prompt="今天北京、上海和成都的天氣怎么樣?")
available_functions = {
"get_current_weather": get_current_weather,
}
run = run_assistant(thread, assistant)
# 下面是處理 Run 并提交工具的返回結(jié)果
# 中間這部分代碼在下面演示
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
來(lái)做函數(shù)的動(dòng)態(tài)調(diào)用get_current_weather
方法使用一些簡(jiǎn)單的邏輯來(lái)模擬天氣查詢的功能:下面是處理 Run 狀態(tài)并提交工具返回結(jié)果的方法,代碼如下:
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
方法,因此我們新建了一個(gè)數(shù)組 tool_infos 來(lái)保存工具返回的結(jié)果拿到工具返回的結(jié)果后,我們通過(guò) Run API 來(lái)提交這些結(jié)果:
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,
)
運(yùn)行結(jié)果如下:
今天北京的溫度是 10℃,上海的溫度是 15℃,成都的溫度是 20℃。
在 Assistant API 的使用過(guò)程中,我們發(fā)現(xiàn)現(xiàn)在獲取 Run 的狀態(tài)都是通過(guò)輪詢的方式,這可能會(huì)導(dǎo)致更多的 token 損耗和性能問(wèn)題,OpenAI 官方介紹在未來(lái)會(huì)對(duì)這一機(jī)制進(jìn)行改進(jìn),將其替換成通知模式,另外還有其他改進(jìn)計(jì)劃如下:
以上就是 Assistant API 的基本原理和功能介紹,和 GPTs 相比兩者各有優(yōu)勢(shì),GPTs 更適合沒(méi)有編程經(jīng)驗(yàn)的用戶使用,而 Assistant API 則適合有開(kāi)發(fā)經(jīng)驗(yàn)的開(kāi)發(fā)人員或團(tuán)隊(duì)使用,而且可以使用自定義工具來(lái)打造更為強(qiáng)大的功能,但也有一些缺點(diǎn),就是無(wú)法使用 DALL-E 3 畫圖工具和上傳圖片(這也意味著無(wú)法做圖片識(shí)別),這些功能相信在未來(lái)會(huì)逐步支持。如果文中有錯(cuò)誤或者不足之處,還請(qǐng)?jiān)谠u(píng)論區(qū)留言。
關(guān)注我,一起學(xué)習(xí)各種人工智能和 AIGC 新技術(shù),歡迎交流,如果你有什么想問(wèn)想說(shuō)的,歡迎在評(píng)論區(qū)留言。
?這里:?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]
文章轉(zhuǎn)自微信公眾號(hào)@極客與黑客之路
對(duì)比大模型API的內(nèi)容創(chuàng)意新穎性、情感共鳴力、商業(yè)轉(zhuǎn)化潛力
一鍵對(duì)比試用API 限時(shí)免費(fèi)