momentry_core/scripts/TOOL_CALLING_README.md

Momentry Tool Calling Module 使用說明

目錄

• 概述
• 安裝
• 快速開始
• 工具說明
• 進階用法
• API 參考
• 常見問題

---

概述

Tool Calling Module 是一個基於 Ollama API 的工具調用模組，支援順序執行多個工具來完成複雜任務。

核心功能

• ✅ 支援 PostgreSQL 資料庫查詢
• ✅ 支援 Qdrant 向量搜尋
• ✅ 支援 Bash 命令執行
• ✅ 支援 HTTP API 調用
• ✅ 防止工具重複調用
• ✅ 自動參數正規化

---

安裝

依賴套件

pip install requests psycopg2-binary

檔案位置

/Users/accusys/momentry_core/scripts/
├─ tool_caller.py      # 核心模組
└─ test_tool_caller.py # 測試腳本

---

快速開始

基本用法

from tool_caller import OllamaToolCaller

# 1. 建立 Tool Caller
caller = OllamaToolCaller(
    base_url="http://localhost:11434",
    model="llama3.1:8b",
    max_tool_calls=5
)

# 2. 註冊預設工具
caller.register_default_tools()

# 3. 執行查詢
result = caller.run("How many videos are in the database?")
print(result)
# 輸出: "There are 23 videos in the database."

中文查詢

result = caller.run("查詢資料庫中有多少影片")
print(result)
# 輸出: "有 23 個視頻存放在資料庫中。"

---

工具說明

1. query_postgres - PostgreSQL 查詢

執行 SQL 查詢語句。

# 直接執行
result = caller.registry.execute("query_postgres", {
    "query": "SELECT COUNT(*) FROM videos"
})
print(result.data)
# {'rows': [{'count': 23}], 'row_count': 1}

參數:

參數	類型	必填	說明
query	string	✅	SQL 查詢語句

支援的 SQL 操作:

• SELECT (查詢)
• INSERT (新增)
• UPDATE (更新)
• DELETE (刪除)

2. search_qdrant - 向量搜尋

在 Qdrant 向量資料庫中搜尋相似內容。

result = caller.registry.execute("search_qdrant", {
    "collection": "momentry_rule1",
    "query_text": "sunset beach",
    "limit": 10
})
print(result.data)
# {'matches': [...], 'match_count': 3}

參數:

參數	類型	必填	預設值	說明
collection	string	❌	momentry_rule1	Qdrant collection 名稱
query_text	string	✅	-	搜尋文字 (會自動轉為向量)
limit	integer	❌	10	最大結果數

3. execute_bash - Bash 執行

執行系統 Bash 命令。

result = caller.registry.execute("execute_bash", {
    "command": "df -h",
    "timeout": 10
})
print(result.data)
# {'stdout': 'Filesystem...', 'stderr': '', 'returncode': 0}

參數:

參數	類型	必填	預設值	說明
command	string	✅	-	Bash 命令
timeout	integer	❌	30	超時秒數

安全限制:

以下命令會被阻止:

• rm -rf /
• mkfs
• dd if=
• > /dev/

4. call_api - HTTP API 調用

調用外部 HTTP API。

result = caller.registry.execute("call_api", {
    "url": "https://api.example.com/data",
    "method": "GET",
    "headers": {"Authorization": "Bearer token123"}
})
print(result.data)
# {'status_code': 200, 'headers': {...}, 'body': '{...}'}

參數:

參數	類型	必填	預設值	說明
url	string	✅	-	API 端點 URL
method	string	❌	GET	HTTP 方法 (GET/POST/PUT/DELETE)
headers	object	❌	{}	HTTP 標頭
data	object	❌	null	請求體 (POST/PUT)

---

進階用法

1. 自訂工具

from tool_caller import OllamaToolCaller, ToolResult

caller = OllamaToolCaller()

# 定義工具執行器
def my_custom_tool(args):
    # 自訂邏輯
    result = do_something(args["param1"], args["param2"])
    return ToolResult(success=True, data=result)

# 註冊工具
caller.registry.register(
    name="my_custom_tool",
    description="My custom tool description",
    parameters={
        "type": "object",
        "properties": {
            "param1": {"type": "string"},
            "param2": {"type": "integer"}
        },
        "required": ["param1"]
    },
    executor=my_custom_tool
)

2. 自訂連線字串

# 使用自訂 PostgreSQL 連線
caller.register_default_tools(
    db_url="postgres://user:pass@host:5432/dbname"
)

# 使用自訂 Qdrant 連線
caller.register_default_tools(
    qdrant_url="http://qdrant-server:6333"
)

3. 多輪對話

caller = OllamaToolCaller(max_tool_calls=5)

# 第一次查詢
result1 = caller.run("查詢資料庫中有多少影片")
print(f"第一次: {result1}")

# 第二次查詢 (會自動清理歷史)
result2 = caller.run("查詢使用者數量")
print(f"第二次: {result2}")

4. 追蹤工具調用歷史

caller = OllamaToolCaller()
caller.register_default_tools()

result = caller.run("查詢資料庫中有多少影片")
print(f"結果: {result}")
print(f"調用歷史: {caller._tool_call_history}")
# ['query_postgres:{"query": "SELECT COUNT(*) FROM videos"}']

---

API 參考

OllamaToolCaller

建構函數

OllamaToolCaller(
    base_url: str = "http://localhost:11434",
    model: str = "llama3.1:8b",
    max_iterations: int = 10,
    max_tool_calls: int = 5
)

參數	類型	預設值	說明
base_url	string	http://localhost:11434	Ollama API 位址
model	string	llama3.1:8b	模型名稱
max_iterations	int	10	最大迭代次數
max_tool_calls	int	5	最大工具調用次數

方法

run(user_query: str) -> str

執行工具調用迴圈並返回最終結果。

chat(messages: List[Dict]) -> Dict

發送聊天請求到 Ollama API。

execute_tool_call(tool_call: Dict) -> ToolResult

執行單個工具調用。

register_default_tools(db_url=None, qdrant_url=None)

註冊預設工具。

---

ToolResult

@dataclass
class ToolResult:
    success: bool      # 是否成功
    data: Any          # 結果資料
    error: str = None  # 錯誤訊息
    execution_time_ms: float = 0  # 執行時間 (毫秒)

---

ToolRegistry

方法

register(name, description, parameters, executor)

註冊新工具。

get_definitions() -> List[Dict]

取得所有工具定義。

has_tool(name: str) -> bool

檢查工具是否存在。

execute(name: str, arguments: Dict) -> ToolResult

執行工具。

---

常見問題

Q1: 工具調用失敗怎麼辦?

檢查以下幾點:

1. Ollama 服務是否運行: curl http://localhost:11434/api/tags
2. 模型是否已下載: ollama list
3. 工具連線是否正確 (PostgreSQL/Qdrant)

Q2: 如何除錯工具調用?

# 啟用詳細日誌
caller = OllamaToolCaller()
caller.register_default_tools()

# 追蹤調用歷史
result = caller.run("查詢資料庫")
print(f"調用歷史: {caller._tool_call_history}")

Q3: 如何處理大量資料?

# 限制 Bash 輸出大小
result = caller.registry.execute("execute_bash", {
    "command": "ls -la | head -100",
    "timeout": 30
})

# 限制 PostgreSQL 結果
result = caller.registry.execute("query_postgres", {
    "query": "SELECT * FROM videos LIMIT 100"
})

Q4: 如何自訂安全規則?

修改 execute_bash 中的 blocked 列表:

blocked = [
    "rm -rf /",
    "mkfs",
    "dd if=",
    "> /dev/",
    "your_custom_dangerous_command"
]

---

範例腳本

查詢資料庫

from tool_caller import OllamaToolCaller

caller = OllamaToolCaller()
caller.register_default_tools()

# 查詢影片數量
result = caller.run("How many videos are in the database?")
print(result)

# 查詢特定資料
result = caller.run("查詢所有狀態為 completed 的影片")
print(result)

系統監控

# 檢查磁碟使用量
result = caller.run("Check the current disk usage")
print(result)

# 檢查記憶體使用量
result = caller.run("Check the current memory usage")
print(result)

# 檢查處理程序
result = caller.run("Show running processes")
print(result)

向量搜尋

# 搜尋相似影片
result = caller.run("Find videos similar to sunset beach")
print(result)

# 搜尋特定人物
result = caller.run("Find videos with John in them")
print(result)

---

版本資訊

• 版本: 1.0.0
• 更新日期: 2026-07-26
• 作者: Momentry Core Team