1. Composio 介紹
...
創建 Composio 帳號：進入 Composio ，進行註冊及登入。
APP 驗證：打開命令提示字元（cmd） *若未執行此綁定步驟，後續達哥將無法協助執行 action
輸入
pip install composio_core composio_openai輸入
composio add [APP] -e entity [ID]進行綁定APP：請到 Composio All Tools 網頁查看
ID：任意取一個代表該 APP 的某一個 Account，如：AIDE
取得 Composio API Key：進入 Composio Settings 的 API Keys 即可取得 API Key。
取得 DaVinci API Key：進入 DaVinci Personal Settings，取得 API Key。
下載
此 zip 檔，解壓縮過後利用瀏覽器開啟 index.html 。View file name web.zip 填入 Plugin Configuration：根據各項目填寫相關資訊，使用此 Plugin Generator 即可馬上透過 No-Code 的方式產生 Plugin。
選擇 APP：選取所需串接的 APP。
獲取 Plugin：提交後，會自動生成一個 JSON 檔，將該檔案上傳至達哥平台。
執行
後即可與達哥進行對話。View file name app.py
...
| 程式碼區塊 | ||
|---|---|---|
| ||
from flask import Flask, render_template, request
from flask_cors import CORS
app = Flask(__name__)
CORS(app)
if __name__ == '__main__':
app.run(debug=True, port=5001) |
3.
...
初始化與綁定服務
3.1 初始化
...
Composio 工具集
通過 Composio API Key 和 Entity ID 來初始化 Composio 工具集。這使我們能夠連接指定的外部應用程序，如 Gmail、Youtube、GitHub 等 。
| 程式碼區塊 | ||
|---|---|---|
| ||
from composio_openai import OpenAIComposioToolSet openaicomposio_clienttoolset = OpenAI( base_url="<https://prod.dvcbot.net/api/assts/v1>", apiComposioToolSet(api_key=DavinciComposio_API_KEY, entity_id=Entity_ID) |
3.2
...
綁定應用
將需要綁定的應用（如Gmail、YouTube、GitHub 等）映射到工具集。
| 程式碼區塊 | ||
|---|---|---|
| ||
fromdef run_composio_openai import(Composio_API_KEY, Entity_ID, App_list): # Initialize ComposioToolSet composio_toolset = ComposioToolSet(api_key=Composio_API_KEY, entity_id=Entity_ID) |
3.3 綁定應用
將需要綁定的應用（如Gmail、YouTube、GitHub 等）映射到工具集。
| 程式碼區塊 | ||
|---|---|---|
| ||
from composio_openai import App # Check if the app is valid App_list_attr = [] for app in App_list: # print(app) app_enum = getattr(App, app.upper(), None) if app_enum is None: raise ValueError(f'Invalid name') else: App_list_attr.append(app_enum) tool = composio_toolset.get_tools(apps=App_list_attr) # tool = composio_toolset.get_tools(apps=[App.GMAIL]) # return app all action to DaVinci return tool[0] |
4. 執行對話和工具調用
將達哥中的用戶訊息作為任務傳入 OpenAI 助手，利用助手協助從上一部的應用當中取得對應 action 並執行。此程式片段會執行達哥回傳的 action (比如發送 E-mail, …)。
| 程式碼區塊 | ||
|---|---|---|
| ||
assistant_instruction = "You are a super intelligent personal assistant, you can help me find what to do."
assistant = openai_client.beta.assistants.create(
name="Personal Assistant",
instructions=assistant_instruction,
model="aide-gpt-4o",
tools=tool,
metadata={"backend_id": "default"}
)
# 創建對話執行線程
thread = openai_client.beta.threads.create()
message = openai_client.beta.threads.messages.create(thread_id=thread.id, role="user", content=my_task)
# 執行助手操作
run = openai_client.beta.threads.runs.create(thread_id=thread.id, assistant_id=assistant.id)
# 處理助手工具調用
response_after_tool_calls = composio_toolset.wait_and_handle_assistant_tool_calls(
client=openai_client,
run=run,
thread=thread,
) |
5. 檢查狀態並返回結果
根據助手操作的狀態來決定是否進行後續操作，並最終返回結果。
| 程式碼區塊 | ||
|---|---|---|
| ||
while run.status in ["queued", "in_progress"]:
run = openai_client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)
if run.status == "requires_action":
print(run)
elif run.status == "failed":
print(run)
else:
messages = openai_client.beta.threads.messages.list(thread_id=thread.id).data
return messages[0].content[0].text.value |
6. API 路由設定
| 程式碼區塊 | ||
|---|---|---|
| ||
def run_action(response, Composio_API_KEY, Entity_ID):
# Initialize ComposioToolSet
composio_toolset = ComposioToolSet(api_key=Composio_API_KEY, entity_id=Entity_ID)
# convert string to python object
if response['arguments'].find('true') != -1:
res_pre = response['arguments'].replace('true', 'True')
res = eval(res_pre)
elif response['arguments'].find('false') != -1:
res_pre = response['arguments'].replace('false', 'False')
res = eval(res_pre)
else:
res = eval(response['arguments'])
# get app action
app_enum = getattr(Action, response['name'].upper(), None)
print(app_enum)
# execute app action
result = composio_toolset.execute_action(action = app_enum, params = res, entity_id= Entity_ID)
return result |
6. API 路由設定
此程式片段設定 Composio Plugin 可以 request 到的 API。
| 程式碼區塊 | ||
|---|---|---|
| ||
# get app action @app.route('/', methods=['POST']) def home(): data = request.get_json() msgComposio_API_KEY = data['conversationComposio_API_KEY'][-1]['content'] Davinci Entity_ID = data['Entity_ID'] App_list = data['App'] response = run_composio(Composio_API_KEY, Entity_ID, App_list) response_data = {"response": response} return response_data @app.route('/run', methods=['Davinci_API_KEY']'POST']) def run_app(): data = request.get_json() Composio_API_KEY = data['Composio_API_KEY'] Entity_ID = data['Entity_ID'] App_listresponse = data['response']['Appfunction_call'] # print(data) responseres = run_composio(msg, Davinci_API_KEYaction(response, Composio_API_KEY, Entity_ID, App_list) response_data = {"response": responseres} return response_data |
7. 執行應用
最後，啟動 Flask 應用來監聽 POST 請求，並根據用戶需求返回結果。
...