MediaTek DaVinci Assistant API 介紹
MediaTek DaVinci 推出 Assistant API，讓您在達哥平台上開發的 Assistant 可以串接進各式各樣的環境當中，進而達到達哥 Assistant 可以在不同環境、裝置服務您的需求。
如何分享製作好的 Assistant
當我們在達哥上建立好 Assistant 後，我們有兩種方式將 Assistant 分享給其他使用者使用：
Share 該 Assistant 給對方
提供該 Assistant 的『Assistant ID 』與『發自 API Key 』給對方
若想收回該 Assistant 的服務時，僅在自己的 API Key 面板上刪除該 API Key 即可
使用 gradio 進行 preview 測試
當我們在達哥平台上創建完 Assistant 時，若想進行 preview 測試，我們提供與 gradio 串接的 sample code。以下為教學步驟：
取得 User API key：
點選達哥面板左下角
More Actions按鈕，選取Settings
進到
Settings後，點選+ API Key按鈕新增
複製
API Key
取得 Assistant ID：
選取欲 preview 的 Assistant，點選
Setting按鈕
選取
Advancedtab，複製Assistant ID
Demo
Text:
Image:
替換掉對應的
API_KEY與ASSISTANT_ID
在輸入框輸入 image url 即可
如果您想要使用本機影像，您可以使用下列 Python 程式碼將它轉換成 base64，以便將其傳遞至 API。 或者您可以使用線上工具將影像檔轉成 base64。
import base64 from mimetypes import guess_type # Function to encode a local image into data URL def local_image_to_data_url(image_path): # Guess the MIME type of the image based on the file extension mime_type, _ = guess_type(image_path) if mime_type is None: mime_type = 'application/octet-stream' # Default MIME type if none is found # Read and encode the image file with open(image_path, "rb") as image_file: base64_encoded_data = base64.b64encode(image_file.read()).decode('utf-8') # Construct the data URL return f"data:{mime_type};base64,{base64_encoded_data}" # Example usage image_path = '<path_to_image>' data_url = local_image_to_data_url(image_path) print("Data URL:", data_url)
得到 base64 字串後，在輸入框輸入以下格式即可
"data:image/jpeg;base64,<your_image_data>"
MediaTek DaVinci Assistant API 使用教學
Curl
Text
要先拿到 Assistant ID & User API key
expot API KEY，ASSISTANT_ID 以及你的 input
export ASSISTANT_ID="YOUR ASSISTANT ID" export API_KEY="YOUR API KEY" export INPUT_MSG="YOUR MESSAGE TO ASSISTANT"
執行以下腳本 (請先確保環境有安裝 jq 套件)
mac: brew install jq
ubuntu: apt-get install jq
cent os: yum install jq
BASE_URL="https://prod.dvcbot.net/api/assts/v1" # create thread AUTH_HEADER="Authorization: Bearer ${API_KEY}" THREAD_URL="${BASE_URL}/threads" THREAD_ID=`curl -s --location "${THREAD_URL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" \ --data '{}' | jq .id | tr -d '"'` # add msg to thread CREATE_MSG_DATA=$(< <(cat <<EOF { "role": "user", "content": "$INPUT_MSG" } EOF )) MSG_URL="${BASE_URL}/threads/${THREAD_ID}/messages" curl -s --location "${MSG_URL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" \ --data "${CREATE_MSG_DATA}" > /dev/null # run the assistant within thread CREATE_RUN_DATA=$(< <(cat <<EOF { "assistant_id": "$ASSISTANT_ID", "additional_instructions": "The current time is: `date '+%Y-%m-%d %H:%M:%S'`" } EOF )) RUN_URL="${BASE_URL}/threads/${THREAD_ID}/runs" RUN_ID=`curl -s --location "${RUN_URL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" \ --data "${CREATE_RUN_DATA}" | jq .id | tr -d '"'` # get run result RUN_STAUS="" while [[ $RUN_STAUS != "completed" ]] do RESP=`curl -s --location --request GET "${RUN_URL}/$RUN_ID" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}"` RUN_STAUS=`echo $RESP| jq .status | tr -d '"'`; REQUIRED_ACTION=`echo $RESP| jq .required_action` while [[ $RUN_STAUS = "requires_action" ]] && [[ ! -z "$REQUIRED_ACTION" ]] do TOOL_OUTPUTS='[]' LEN=$( echo "$REQUIRED_ACTION" | jq '.submit_tool_outputs.tool_calls | length' ) for (( i=0; i<$LEN; i++ )) do FUNC_NAME=`echo "$REQUIRED_ACTION" | jq ".submit_tool_outputs.tool_calls[$i].function.name" | tr -d '"'` ARGS=`echo "$REQUIRED_ACTION" | jq ".submit_tool_outputs.tool_calls[$i].function.arguments"` ARGS=${ARGS//\\\"/\"} ARGS=${ARGS#"\""} ARGS=${ARGS%"\""} PLUGINAPI_URL="${BASE_URL}/pluginapi?tid=${THREAD_ID}&aid=${ASSISTANT_ID}&pid=${FUNC_NAME}" OUTPUT=`curl -s --location "${PLUGINAPI_URL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" \ --data "${ARGS}"` OUTPUT="${OUTPUT:0:8000}" OUTPUT=${OUTPUT//\"/\\\"} CALL_ID=`echo "$REQUIRED_ACTION" | jq ".submit_tool_outputs.tool_calls[$i].id" | tr -d '"'` TOOL_OUTPUT=$(< <(cat <<EOF { "tool_call_id": "$CALL_ID", "output": "$OUTPUT" } EOF )) TOOL_OUTPUTS=$(jq --argjson obj "$TOOL_OUTPUT" '. += [$obj]' <<< "$TOOL_OUTPUTS") done SUBMIT_TOOL_OUTPUT_RUN_RUL="${BASE_URL}/threads/${THREAD_ID}/runs/${RUN_ID}/submit_tool_outputs" TOOL_OUTPUTS_DATA=$(< <(cat <<EOF { "tool_outputs": $TOOL_OUTPUTS } EOF )) curl -s --location "${SUBMIT_TOOL_OUTPUT_RUN_RUL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" \ --data "${TOOL_OUTPUTS_DATA}" > /dev/null RESP=`curl -s --location --request GET "${RUN_URL}/$RUN_ID" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}"` RUN_STAUS=`echo $RESP| jq .status | tr -d '"'`; sleep 1 done sleep 1 done #list msg RESPONSE_MSG=`curl -s --location --request GET "${MSG_URL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" | jq .data[0].content[].text.value` echo "you: "$INPUT_MSG echo "" echo "davinci bot: "$RESPONSE_MSG即可看到結果如下
you: "your message here" davinci bot: "response from assistant"
Image
要先拿到 Assistant ID & User API key
expot API KEY，ASSISTANT_ID 以及你的 input
export ASSISTANT_ID="YOUR ASSISTANT ID" export API_KEY="YOUR API KEY" export IMAGE_URL="YOUR IMAGE URL HEHE"
IMAGE_URL 格式參考上方 Gradio image 範例
執行以下腳本 (請先確保環境有安裝 jq 套件)
mac: brew install jq
ubuntu: apt-get install jq
cent os: yum install jq
BASE_URL="https://prod.dvcbot.net/api/assts/v1" # create thread AUTH_HEADER="Authorization: Bearer ${API_KEY}" THREAD_URL="${BASE_URL}/threads" THREAD_ID=`curl -s --location "${THREAD_URL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" \ --data '{}' | jq .id | tr -d '"'` # add msg to thread CREATE_MSG_DATA=$(< <(cat <<EOF { "role": "user", "content": [ { "type": "image_url", "image_url": { "url": "$IMAGE_URL" } } ] } EOF )) MSG_URL="${BASE_URL}/threads/${THREAD_ID}/messages" curl -s --location "${MSG_URL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" \ --data "${CREATE_MSG_DATA}" > /dev/null # run the assistant within thread CREATE_RUN_DATA=$(< <(cat <<EOF { "assistant_id": "$ASSISTANT_ID", "additional_instructions": "The current time is: `date '+%Y-%m-%d %H:%M:%S'`" } EOF )) RUN_URL="${BASE_URL}/threads/${THREAD_ID}/runs" RUN_ID=`curl -s --location "${RUN_URL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" \ --data "${CREATE_RUN_DATA}" | jq .id | tr -d '"'` # get run result RUN_STAUS="" while [[ $RUN_STAUS != "completed" ]] do RESP=`curl -s --location --request GET "${RUN_URL}/$RUN_ID" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}"` RUN_STAUS=`echo $RESP| jq .status | tr -d '"'`; REQUIRED_ACTION=`echo $RESP| jq .required_action` while [[ $RUN_STAUS = "requires_action" ]] && [[ ! -z "$REQUIRED_ACTION" ]] do TOOL_OUTPUTS='[]' LEN=$( echo "$REQUIRED_ACTION" | jq '.submit_tool_outputs.tool_calls | length' ) for (( i=0; i<$LEN; i++ )) do FUNC_NAME=`echo "$REQUIRED_ACTION" | jq ".submit_tool_outputs.tool_calls[$i].function.name" | tr -d '"'` ARGS=`echo "$REQUIRED_ACTION" | jq ".submit_tool_outputs.tool_calls[$i].function.arguments"` ARGS=${ARGS//\\\"/\"} ARGS=${ARGS#"\""} ARGS=${ARGS%"\""} PLUGINAPI_URL="${BASE_URL}/pluginapi?tid=${THREAD_ID}&aid=${ASSISTANT_ID}&pid=${FUNC_NAME}" OUTPUT=`curl -s --location "${PLUGINAPI_URL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" \ --data "${ARGS}"` OUTPUT="${OUTPUT:0:8000}" OUTPUT=${OUTPUT//\"/\\\"} CALL_ID=`echo "$REQUIRED_ACTION" | jq ".submit_tool_outputs.tool_calls[$i].id" | tr -d '"'` TOOL_OUTPUT=$(< <(cat <<EOF { "tool_call_id": "$CALL_ID", "output": "$OUTPUT" } EOF )) TOOL_OUTPUTS=$(jq --argjson obj "$TOOL_OUTPUT" '. += [$obj]' <<< "$TOOL_OUTPUTS") done SUBMIT_TOOL_OUTPUT_RUN_RUL="${BASE_URL}/threads/${THREAD_ID}/runs/${RUN_ID}/submit_tool_outputs" TOOL_OUTPUTS_DATA=$(< <(cat <<EOF { "tool_outputs": $TOOL_OUTPUTS } EOF )) curl -s --location "${SUBMIT_TOOL_OUTPUT_RUN_RUL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" \ --data "${TOOL_OUTPUTS_DATA}" > /dev/null RESP=`curl -s --location --request GET "${RUN_URL}/$RUN_ID" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}"` RUN_STAUS=`echo $RESP| jq .status | tr -d '"'`; sleep 1 done sleep 1 done #list msg RESPONSE_MSG=`curl -s --location --request GET "${MSG_URL}" \ --header 'OpenAI-Beta: assistants=v2' \ --header 'Content-Type: application/json' \ --header "${AUTH_HEADER}" | jq .data[0].content[].text.value` echo "" echo "davinci bot: "$RESPONSE_MSG即可看到結果如下
davinci bot: "response from assistant"
Python
Text or image as Input
import json
from openai import OpenAI
from datetime import datetime
ASSISTANT_API = 'https://prod.dvcbot.net/api/assts/v1'
API_KEY = ''
client = OpenAI(
base_url=ASSISTANT_API,
api_key=API_KEY,
)
ASSISTANT_ID = ''
# 定義多個訊息
messages = [
{"type": "text", "text": "tell me about the image"},
{"type": "image_url", "image_url": {"url": "https://xxx.xxx.xxx.jpg"}},
{"type": "text", "text": "What do you think about this image?"}
]
# 建立 thread
thread = client.beta.threads.create(messages=[])
# 連續發送訊息
for message in messages:
client.beta.threads.messages.create(thread_id=thread.id, role='user', content=[message])
# 執行 assistant
run = client.beta.threads.runs.create_and_poll(thread_id=thread.id, assistant_id=ASSISTANT_ID, additional_instructions=f"\nThe current time is: {datetime.now()}")
while run.status == 'requires_action' and run.required_action:
outputs = []
for call in run.required_action.submit_tool_outputs.tool_calls:
resp = client._client.post(ASSISTANT_API + '/pluginapi', params={"tid": thread.id, "aid": ASSISTANT_ID, "pid": call.function.name}, headers={"Authorization": "Bearer " + API_KEY}, json=json.loads(call.function.arguments))
outputs.append({"tool_call_id": call.id, "output": resp.text[:8000]})
run = client.beta.threads.runs.submit_tool_outputs_and_poll(run_id=run.id, thread_id=thread.id, tool_outputs=outputs)
if run.status == 'failed' and run.last_error:
print(run.last_error.model_dump_json())
msgs = client.beta.threads.messages.list(thread_id=thread.id, order='desc')
client.beta.threads.delete(thread_id=thread.id)
print(msgs.data[0].content[0].text.value)
Text & image as input (Streaming)
import asyncio
import json
import httpx
from openai import AsyncOpenAI
ASSISTANT_API = "https://prod.dvcbot.net/api/assts/v1"
API_KEY = ""
ASSISTANT_ID = ""
USER_PROMPT = "從一數到一千"
async def main():
http_client = httpx.AsyncClient(verify=False)
client = AsyncOpenAI(base_url=ASSISTANT_API, api_key=API_KEY, http_client=http_client)
thread = await client.beta.threads.create()
user_prompt = USER_PROMPT
await client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content=user_prompt,
)
stream = await client.beta.threads.runs.create(
assistant_id=ASSISTANT_ID,
thread_id=thread.id,
stream=True,
)
requires_action_run_id = ""
async for event in stream:
if event.event == "thread.message.delta":
print(event)
elif event.event == "thread.run.requires_action":
requires_action_run_id = event.data.id
if requires_action_run_id != "":
run = await client.beta.threads.runs.retrieve(requires_action_run_id, thread_id=thread.id)
outputs = []
for call in run.required_action.submit_tool_outputs.tool_calls:
print(f"call plugin {call.function.name} with args: {call.function.arguments}")
resp = await client._client.post(
ASSISTANT_API + "/pluginapi",
params={"tid": thread.id, "aid": ASSISTANT_ID, "pid": call.function.name},
headers={"Authorization": "Bearer " + API_KEY},
json=json.loads(call.function.arguments),
timeout=30,
)
result = resp.text[:8000]
print(f"plugin {call.function.name} result {result}")
outputs.append({"tool_call_id": call.id, "output": result})
stream = await client.beta.threads.runs.submit_tool_outputs(
run_id=run.id,
stream=True,
thread_id=thread.id,
tool_outputs=outputs,
)
async for event in stream:
print(event)
await client.beta.threads.delete(thread_id=thread.id)
if __name__ == "__main__":
asyncio.run(main())
Voice mode
(Coming Soon)