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。Plugin。(圖待補)
選擇 APP:選取所需串接的 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>", ComposioToolSet(api_key=DavinciComposio_API_KEY, entity_id=Entity_ID) |
3.2
...
綁定應用
將需要綁定的應用(如Gmail、YouTube、GitHub 等)映射到工具集。
程式碼區塊 | ||
---|---|---|
| ||
fromdef run_composio_openai import ComposioToolSet(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 並執行。文字逮捕
程式碼區塊 | ||
---|---|---|
| ||
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.valuedef 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 路由設定
程式碼區塊 | ||
---|---|---|
| ||
# 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'] response# print(data) res = run_composioaction(msgresponse, Davinci_API_KEY, Composio_API_KEY, Entity_ID, App_list) response_data = {"response": responseres} return response_data |
7. 執行應用
最後,啟動 Flask 應用來監聽 POST 請求,並根據用戶需求返回結果。
...