Day 28:受限唯讀 Agent 的決策迴圈：整合檢索、呼叫圖與 Tool 呼叫

2026 iThome 鐵人賽

DAY 28

AI Engineering

30 天打造 Codebase Intelligence Agent：從程式碼檢索、結構化索引到變更影響分析實戰系列第 28 篇

18th鐵人賽

__zoe

團隊nutc imac

2026-10-12 20:07:37

116 瀏覽

分享至

在過去四週中，我們獨立完成了 AST 語法索引、SQLite 符號庫、RRF 混合檢索、本地 LLM Reranker、跨檔案呼叫圖（Call Graph）以及 Git Diff 變更分析。

但如果只把這些功能做成零散的 CLI 指令，使用者仍然需要自己當「膠水」：先下搜尋指令找檔案，再手動拿 Symbol 去跑影響分析，最後自己把結果拼裝起來。

今天我們把所有組件收攏整合，打造一個專為程式碼庫服務的受限唯讀 Agent（Guarded ReadOnly Agent）。

為什麼必須是「受限唯讀（Guarded ReadOnly）」？

程式碼庫智慧體（Codebase Agent）最危險的陷阱就是給予過多自由度（例如允許執行任意 shell 指令或寫入檔案）。

在本專案中，我們貫徹三個安全鐵律：

嚴格唯讀（Read-Only Sandbox）：工具箱只提供純查詢介面（搜尋、取片段、查呼叫圖），嚴禁檔案修改與未授權命令執行。
確定性工具分派（Deterministic Tool Dispatching）：模型只能透過標準 JSON Schema 表達意圖，所有參數在執行前經過嚴格型別校驗與邊界過濾。
有界迭代（Bounded ReAct Loop）：限制最多思考與工具調用輪數（預設 4 輪），防止模型在複雜問題中陷入發散式的無窮迴圈。

Agent 工具箱架構與 Schema 設計

我們為 Agent 提供四個高內聚的唯讀工具：

search_code(query, limit)：執行 Hybrid Search + Reranker 找出最相關的程式碼區塊。
get_symbol_info(path, symbol)：精確取得該 Symbol 的行號範圍、文件與原始碼。
get_callers(path, symbol)：查詢誰直接或間接呼叫了這個函式。
analyze_impact(path, symbol, max_depth)：執行反向 BFS 衝擊傳播，產出受波及呼叫鏈。

核心實作：`app/agent_loop.py`

在 app/agent_loop.py 中實作 Tool 註冊表、執行器以及基於本地模型（gemma4:e4b）的 ReAct 決策迴圈：

# app/agent_loop.py
import json
import re
import urllib.request
import sqlite3
from typing import Dict, Any, List, Callable
from dataclasses import dataclass

@dataclass
class ToolDefinition:
    name: str
    description: str
    parameters: Dict[str, Any]
    handler: Callable[..., Any]

class GuardedAgent:
    def __init__(self, conn: sqlite3.Connection, model: str = "gemma4:e4b", host: str = "http://localhost:11434"):
        self.conn = conn
        self.model = model
        self.url = f"{host}/api/chat"
        self.tools: Dict[str, ToolDefinition] = {}

    def register_tool(self, name: str, description: str, parameters: Dict[str, Any], handler: Callable):
        self.tools[name] = ToolDefinition(name, description, parameters, handler)

    def _call_ollama(self, messages: List[Dict[str, str]]) -> str:
        payload = {
            "model": self.model,
            "messages": messages,
            "stream": False,
            "options": {"temperature": 0.0}
        }
        req = urllib.request.Request(
            self.url,
            data=json.dumps(payload).encode("utf-8"),
            headers={"Content-Type": "application/json"}
        )
        with urllib.request.urlopen(req, timeout=30) as resp:
            data = json.loads(resp.read().decode("utf-8"))
            return data["message"]["content"]

    def _build_system_prompt(self) -> str:
        tool_descs = []
        for t in self.tools.values():
            tool_descs.append(
                f"- Tool: `{t.name}`\n  Description: {t.description}\n  Params: {json.dumps(t.parameters)}"
            )

        return (
            "You are a cautious Codebase Intelligence Agent operating in a READ-ONLY environment.\n"
            "Your task is to answer user questions about the codebase strictly supported by evidence.\n\n"
            "Available Tools:\n"
            f"{chr(10).join(tool_descs)}\n\n"
            "Execution Protocol:\n"
            "To call a tool, output ONLY a JSON object in this format:\n"
            "{\n"
            '  "action": "call_tool",\n'
            '  "tool": "<tool_name>",\n'
            '  "arguments": { ... }\n'
            "}\n\n"
            "When you have collected enough evidence to answer, output:\n"
            "{\n"
            '  "action": "final_answer",\n'
            '  "answer": "<your thorough answer with file paths and line numbers>"\n'
            "}\n"
            "Do NOT wrap output in markdown codeblocks if possible. Only emit valid JSON."
        )

    def run(self, user_query: str, max_steps: int = 4) -> str:
        """執行受限決策迴圈 (Bounded Decision Loop)"""
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": user_query}
        ]

        for step in range(max_steps):
            raw_reply = self._call_ollama(messages)
            decision = self._parse_json_action(raw_reply)

            if not decision:
                # 容錯回退：若輸出非結構化格式，直接以文字作為回答
                return raw_reply

            action = decision.get("action")

            if action == "final_answer":
                return decision.get("answer", "")

            if action == "call_tool":
                tool_name = decision.get("tool")
                args = decision.get("arguments", {})

                if tool_name not in self.tools:
                    tool_result = f"Error: Tool '{tool_name}' not found."
                else:
                    try:
                        # 執行註冊的唯讀處理器
                        tool_result = self.tools[tool_name].handler(**args)
                    except Exception as e:
                        tool_result = f"Error executing tool: {str(e)}"

                # 將執行結果回灌給模型
                messages.append({"role": "assistant", "content": json.dumps(decision)})
                messages.append({
                    "role": "user",
                    "content": f"[Tool Result from {tool_name}]:\n{json.dumps(tool_result, ensure_ascii=False)}"
                })

        return "Agent stopped: Reached maximum investigation steps without final answer."

    def _parse_json_action(self, text: str) -> Dict[str, Any]:
        """從回應文字中提取 JSON 動作，抵禦 Markdown 污染"""
        cleaned = re.sub(r"^```(?:json)?", "", text.strip()).rstrip("`").strip()
        try:
            return json.loads(cleaned)
        except json.JSONDecodeError:
            pass

        match = re.search(r"\{.*\}", cleaned, re.DOTALL)
        if match:
            try:
                return json.loads(match.group(0))
            except json.JSONDecodeError:
                pass
        return {}

組裝工具鏈：綁定檢索與衝擊分析

在 app/agent_service.py 中建立預設的 Agent 工廠函式，綁定底層能力：

# app/agent_service.py
import sqlite3
from app.agent_loop import GuardedAgent
from app.impact_analyzer import ImpactAnalyzer

def create_codebase_agent(conn: sqlite3.Connection, search_engine) -> GuardedAgent:
    agent = GuardedAgent(conn)

    # 1. 搜尋工具
    agent.register_tool(
        name="search_code",
        description="Search codebase using hybrid keyword and semantic retrieval.",
        parameters={"query": "string", "limit": "integer (optional)"},
        handler=lambda query, limit=5: search_engine.search(query, limit=limit)
    )

    # 2. 符號原始碼查詢工具
    def _get_symbol_info(path: str, symbol: str):
        cur = conn.cursor()
        cur.execute(
            "SELECT path, name, start_line, end_line, content FROM vectors WHERE path = ? AND name = ?",
            (path, symbol)
        )
        row = cur.fetchone()
        if row:
            return dict(row)
        return {"error": "Symbol not found"}

    agent.register_tool(
        name="get_symbol_info",
        description="Get exact code snippet and line numbers of a symbol.",
        parameters={"path": "string", "symbol": "string"},
        handler=_get_symbol_info
    )

    # 3. 變更影響分析工具
    impact_analyzer = ImpactAnalyzer(conn)
    agent.register_tool(
        name="analyze_impact",
        description="Find all downstream callers impacted if a function is changed.",
        parameters={"path": "string", "symbol": "string", "max_depth": "integer (optional)"},
        handler=lambda path, symbol, max_depth=3: [
            {"path": n.path, "symbol": n.symbol, "depth": n.depth, "line": n.trigger_line}
            for n in impact_analyzer.trace_impact(path, symbol, max_depth=max_depth)
        ]
    )

    return agent

實作單元測試：`tests/unit/test_agent_loop.py`

驗證 Agent 的決策循環、Tool 調用與 JSON 容錯解析（不依賴真實 Ollama 模型）：

# tests/unit/test_agent_loop.py
import sqlite3
from app.agent_loop import GuardedAgent

def test_json_action_parsing_clean_and_wrapped():
    conn = sqlite3.connect(":memory:")
    agent = GuardedAgent(conn)

    # 1. 純文字 JSON
    p1 = agent._parse_json_action('{"action": "final_answer", "answer": "Done"}')
    assert p1.get("action") == "final_answer"
    assert p1.get("answer") == "Done"

    # 2. Markdown 污染
    p2 = agent._parse_json_action('```json\n{"action": "call_tool", "tool": "search_code"}\n```')
    assert p2.get("action") == "call_tool"
    assert p2.get("tool") == "search_code"

def test_agent_tool_dispatching():
    conn = sqlite3.connect(":memory:")
    agent = GuardedAgent(conn)

    called_args = []
    agent.register_tool(
        name="mock_search",
        description="mock",
        parameters={"q": "string"},
        handler=lambda q: called_args.append(q) or [{"file": "test.py"}]
    )

    # 模擬兩輪模型對話：第一輪呼叫工具，第二輪給出答案
    responses = [
        '{"action": "call_tool", "tool": "mock_search", "arguments": {"q": "find_init"}}',
        '{"action": "final_answer", "answer": "Found in test.py"}'
    ]

    agent._call_ollama = lambda msgs: responses.pop(0)

    final = agent.run("Find init")
    assert called_args == ["find_init"]
    assert final == "Found in test.py"

執行測試確認綠燈：

uv run pytest tests/unit/test_agent_loop.py -v

tests/unit/test_agent_loop.py::test_json_action_parsing_clean_and_wrapped PASSED [ 50%]
tests/unit/test_agent_loop.py::test_agent_tool_dispatching PASSED              [100%]
============================== 2 passed in 0.04s ==============================

實際在 `mobileai-local-rag` 執行 Agent 查詢

整合進 CLI 後，向 Agent 提出需要「檢索 + 影響分析」複合能力的問題：

uv run python -m app.cli agent-ask "如果修改 Qdrant 的 collection name 取得邏輯，會影響哪些函式？"

終端機執行過程追蹤

[Agent Loop Step 1]
Decision: Call Tool -> search_code(query="collection name")
Result: Found `get_collection_name` in src/rag_common.py:18

[Agent Loop Step 2]
Decision: Call Tool -> analyze_impact(path="src/rag_common.py", symbol="get_collection_name")
Result: 3 downstream nodes (build_index, init_session, chat_loop)

[Agent Loop Step 3]
Decision: Final Answer Emitted.

============================================================
                     Agent 綜合回答報告
============================================================
修改 Qdrant collection 名稱的定義 (位於 `src/rag_common.py:18` 的 `get_collection_name`)
將會影響以下下游進入點與呼叫鏈：

1. 【直接影響 - Depth 1】
   - `src/build_index.py::build_index` (行號 18)：建立向量庫時依賴該常數名稱初始化 Collection。
   - `src/rag_chat.py::init_session` (行號 32)：對話工作階段啟動時用於連線至指定 Collection。

2. 【間接影響 - Depth 2】
   - `src/rag_chat.py::chat_loop` (行號 75)：進入對話迴圈前呼叫了 `init_session()`。

建議在修改後執行 `tests/unit/test_build_index.py` 與 `tests/unit/test_rag_chat.py` 驗證相容性。
============================================================

Agent 不再只是回答一段程式碼文字，而是自己排程調用檢索器鎖定目標，再調用呼叫圖追蹤下游，給出有證據支持的工程分析。