Module 參考

src/ 下各模組的職責、介面與使用方式。

---

Agent 層

src/agent/base_langraph_agent.py

所有 agent 的抽象基底。定義共用的 HarnessState 和 agent 生命週期。

HarnessState（TypedDict）

所有 agent 共享的狀態物件，透過 LangGraph 在節點間傳遞：

class HarnessState(TypedDict):
    user_request: str           # 使用者原始輸入
    session_context: dict       # 多輪對話歷史 {"history": [...]}
    constraints: dict           # 工具權限、policy 限制
    plan: dict                  # Planner 產出的執行計畫
    actor_outputs: list[dict]   # 每個 Executor step 的輸出
    review_results: list[dict]  # Reviewer 的評估結果
    current_step_index: int     # 目前執行到第幾個 step
    retry_count: int            # 已重試次數
    max_retries: int            # 最大重試次數（預設 3）
    llm_provider: str           # LLM provider 名稱
    status: str                 # 當前狀態：planning / acting / reviewing / done
    final_output: str           # 最終輸出結果

BaseLangGraphAgent（ABC）

class BaseLangGraphAgent(ABC):
    def __init__(
        self,
        name: str,
        role: str,
        llm: Any,
        prompt_path: str | None = None,
        tools: list | None = None,
    ) -> None

    @abstractmethod
    def build_graph(self) -> StateGraph: ...

    def run(self, state: HarnessState) -> HarnessState:
        """執行 agent 的 compiled graph，回傳更新後的 state"""

    def read_prompt(self) -> str:
        """載入 prompts/{name}.md，第一次讀取後 cache"""

    def build_messages(self, user_content: str) -> list[BaseMessage]:
        """組合 [SystemMessage(prompt), HumanMessage(user_content)]"""

    def parse_json_response(self, content: str) -> dict[str, Any]:
        """從 LLM 輸出萃取 JSON，支援 markdown code block 剝除"""

---

src/agent/planner_agent.py

將使用者需求拆解為 Executor 可執行的步驟清單。

class PlannerAgent(BaseLangGraphAgent):
    def __init__(
        self,
        llm: Any | None = None,
        llm_provider: LLMProvider = "custom",
    ) -> None

輸出 plan 格式（存入 state["plan"]）：

{
  "task_goal": "string",
  "task_type": "research | coding | writing | automation | data_analysis | other",
  "assumptions": ["string"],
  "risks": ["string"],
  "success_criteria": ["string"],
  "steps": [
    {
      "id": "step_1",
      "description": "string",
      "required_tool": "tool_name | none",
      "expected_output": "string",
      "risk_level": "low | medium | high"
    }
  ],
  "requires_user_confirmation": false
}

---

src/agent/executor_agent.py

依照 plan 的 current_step_index 執行單一步驟。

class ExecutorAgent(BaseLangGraphAgent):
    def __init__(
        self,
        llm: Any | None = None,
        llm_provider: LLMProvider = "custom",
    ) -> None

輸出格式（append 到 state["actor_outputs"]）：

{
  "step_id": "step_1",
  "status": "success | failed | partial",
  "actions_taken": [{"action": "string", "outcome": "string"}],
  "tool_calls": [{"tool_name": "string", "arguments": {}}],
  "tool_outputs": ["string"],
  "artifacts_created": ["string"],
  "errors": ["string"],
  "result_summary": "string"
}

---

src/agent/reviewer_agent.py

檢查所有 Executor 輸出，決定下一步動作。

class ReviewerAgent(BaseLangGraphAgent):
    def __init__(
        self,
        llm: Any | None = None,
        llm_provider: LLMProvider = "custom",
    ) -> None

輸出格式（append 到 state["review_results"]）：

{
  "verdict": "pass | fail | needs_revision",
  "score": 0.85,
  "issues": [
    {
      "type": "missing_output | format_error | logic_error | safety | quality",
      "severity": "low | medium | high",
      "description": "string",
      "suggested_fix": "string"
    }
  ],
  "next_action": "finalize | retry_actor | replan | ask_user",
  "question": "string（ask_user 時填寫）"
}

---

Graph 層

src/graph/harness_graph.py

串接三個 agent 的 LangGraph 主圖，處理條件路由與重試邏輯。

class HarnessGraph:
    def __init__(
        self,
        llm_provider: LLMProvider = "custom",
        max_retries: int = 3,
        status_callback: Callable[[str, str], None] | None = None,
        model_name: str | None = None,
    ) -> None

    def run(
        self,
        user_request: str,
        session_context: dict[str, Any] | None = None,
        constraints: dict[str, Any] | None = None,
        thread_id: str | None = None,
    ) -> HarnessState:
        """主入口：執行完整 plan→execute→review 流程，回傳最終 state"""

使用範例：

from src.graph.harness_graph import HarnessGraph

graph = HarnessGraph(llm_provider="custom")
result = graph.run(
    user_request="幫我寫一個 Python Hello World",
    session_context={"history": []},
)
print(result["final_output"])

節點流程：

START → planner → executor (loop) → reviewer
                                       │
              ┌──────────────────────┤
              │  pass / max_retries  → finalize → END
              │  retry_actor         → count_retry → executor
              │  replan              → count_retry → planner
              │  ask_user            → finalize (顯示 question)

---

Service 層

src/service/config_service.py

讀寫 ~/.harness_config.json，持久化使用者設定。

from src.service.config_service import load_config, save_config

# 讀取（不存在回傳預設值）
config = load_config()
# → {"lang": "zh", "provider": "custom"}

# 寫入
save_config({"lang": "en", "provider": "openai"})

---

src/service/i18n_service.py

中英文 UI 字串查詢。

from src.service.i18n_service import t, t_cmd
from src.ui.constants import SLASH_COMMANDS

# 取得 UI 字串
label = t("you", "zh")       # → "你"
label = t("you", "en")       # → "You"

# 取得 slash command 說明
desc = t_cmd("help", "zh", SLASH_COMMANDS)  # → "列出所有可用指令"

---

UI 層

src/ui/app.py

Textual App 主體，負責初始化主題與路由至正確的 Screen。

from src import HarnessChatApp, main

# 直接啟動
main()

# 或程式化使用
app = HarnessChatApp()
app.run()

---

src/ui/screens/chat_screen.py

主對話畫面。透過 HarnessGraph 執行請求，支援 slash 指令與 token 追蹤。

class ChatScreen(Screen):
    def __init__(self, provider: str, lang: str = "zh") -> None

重要屬性：

屬性	說明
provider	目前使用的 LLM provider
lang	介面語言（zh / en）
history	本 session 對話歷史 [{"user": ..., "assistant": ...}]
harness	HarnessGraph 實例
_last_retry_count	上一輪執行的重試次數（從 HarnessState["retry_count"] 讀取）

每輪對話結束後，#token-counter 若 retry_count > 0 會額外顯示 retries: N：

Session token：1,234  retries: 2

---

src/ui/screens/provider_screen.py

首次啟動設定畫面，讓使用者選擇語言與 provider，寫入 config 後推入 ChatScreen。

---

src/ui/constants.py

所有 UI 常數集中定義，避免 magic string 散落各處。

from src.ui.constants import (
    SLASH_COMMANDS,  # slash 指令定義
    MODELS,          # provider → model 清單
    THEMES,          # Textual Theme 物件
    AGENT_COLORS,    # agent 名稱 → Rich 顏色
    PROVIDER_OPTIONS, LANG_OPTIONS,  # Select 元件選項
    BASE_CSS,        # 共用 CSS
)

新增 slash 指令：在 SLASH_COMMANDS 新增 key，在 ChatScreen._handle_slash() 的 handlers dict 加對應 handler。

---

MCP 層

mcp.json（根目錄）

Claude Desktop 標準格式，定義專案使用的 MCP server 清單。${VAR} 占位符在執行時從環境變數解析。

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_PERSONAL_ACCESS_TOKEN}"
      }
    }
  }
}

目前定義的 server：serena、github、context7、firecrawl。

---

src/mcp/config.py

讀取 mcp.json 並解析環境變數。

from src.mcp.config import MCPConfig

config = MCPConfig.load("mcp.json")

config.names()           # → ["serena", "github", "context7", "firecrawl"]
config.get("github")     # → MCPServerConfig(name="github", command="npx", ...)
config.servers()         # → dict[str, MCPServerConfig]

MCPServerConfig 屬性：name、command、args、env（已解析）、full_command（完整指令字串）。

---

src/mcp/loader.py

用 MCP Python SDK 連線 stdio server，查詢可用工具。每次連線為短暫生命週期（spawn → initialize → list_tools → cleanup）。

from src.mcp.loader import MCPServerManager, test_all
from src.mcp.config import MCPConfig

config = MCPConfig.load()
srv = config.get("github")

# 測試單一 server
manager = MCPServerManager(srv)
result = asyncio.run(manager.test())
print(result.success, result.tools, result.latency_ms)

# 並行測試全部
results = asyncio.run(test_all(list(config.servers().values())))

TestResult 屬性：name、success、tools（list[ToolInfo]）、error、latency_ms。

---

mcp_server.py（根目錄，CLI）

互動式管理 CLI，透過 uv run mcp_server.py 執行。

uv run mcp_server.py list          # 列出 mcp.json 中定義的所有 server
uv run mcp_server.py status        # 檢查各 server 的指令是否在 PATH 中
uv run mcp_server.py test          # 連線並列出所有 server 的可用 tools
uv run mcp_server.py test github   # 測試單一 server

---

Tool 層

src/tool/base_tool.py

所有工具的抽象基底，繼承 LangChain BaseTool。繼承後自動向 ToolRegistry 登錄，無需手動呼叫。

HarnessBaseTool

from src.tool.base_tool import HarnessBaseTool

class SearchWebTool(HarnessBaseTool):
    name: str = "search_web"
    description: str = "Search the web for information"

    def _run(self, query: str) -> str:
        # 實作工具邏輯
        return results

規則： - name 必須為非空字串，且不以 _ 開頭，才會自動登錄 - description 必填，LLM 依此判斷何時呼叫此工具 - 實作 _run()；如需非同步支援，另實作 _arun()

定義完成後，在 src/tool/__init__.py 加上 import 以觸發登錄：

# src/tool/__init__.py
from src.tool.search_tool import SearchWebTool  # 加這行

---

src/tool/registry.py

單例 registry，儲存所有已登錄工具類別（{name: cls}）。

from src.tool.registry import ToolRegistry

# 取得所有已登錄工具名稱
ToolRegistry.names()             # → ["search_web", "read_file"]

# 取得所有工具類別（供 agent 使用）
ToolRegistry.get_all()           # → {"search_web": SearchWebTool, ...}

# 取得單一工具類別
ToolRegistry.get("search_web")   # → SearchWebTool

# 建構工具實例並傳給 ExecutorAgent
tools = [cls() for cls in ToolRegistry.get_all().values()]
agent = ExecutorAgent(tools=tools)

---

Enum 層

src/enum/tool_registry.py

import 時從 ToolRegistry 動態建立 ToolName enum。每個已登錄工具對應一個成員，key 為工具名稱大寫，value 為原始名稱字串。

使用前提：src/tool/__init__.py 中已 import 所有具體工具（觸發登錄）。

from src.enum import ToolName

ToolName.SEARCH_WEB        # → <ToolName.SEARCH_WEB: 'search_web'>
ToolName.SEARCH_WEB.value  # → "search_web"

# 用於 plan step 的 required_tool 比對
step["required_tool"] == ToolName.SEARCH_WEB.value  # → True / False

新增工具後 enum 自動更新，無需手動維護。

---

Utils 層

src/utils/llm_factory.py

依 provider 建構 LangChain LLM 實例，讀取 .env 設定。

from src.utils.llm_factory import LLMFactory, LLMConfig

# 建構 LLM（供 agent 使用）
llm = LLMFactory("custom").build()
llm = LLMFactory("gemini", model_name="gemini-1.5-pro").build()

# 測試連線（僅 custom provider）
ok, detail = LLMFactory("custom").test_connection()
# → (True, "OK") 或 (False, "HTTP 403: ...")

# 讀取設定
config: LLMConfig = LLMFactory("openai").load_config()
# → LLMConfig(provider="openai", api_key="sk-...", model_name="gpt-4o")

支援的 Provider：

Provider	LangChain 類別	.env 前綴
openai	ChatOpenAI	OPENAI_
anthropic	ChatAnthropic	ANTHROPIC_
gemini	ChatGoogleGenerativeAI	GEMINI_
custom	ChatOpenAI + custom base_url	CUSTOM_

Custom Provider 注意事項

custom provider 使用 httpx event hook 替換 user-agent header， 避免部分 WAF 封鎖 OpenAI/Python x.x.x 的問題。

---

src/utils/token_tracker.py

LangChain callback，累計 session 內所有 LLM 呼叫的 token 用量。

from src.utils.token_tracker import TokenTracker
from langchain_core.language_models import BaseChatModel

tracker = TokenTracker()
llm.invoke(messages, config={"callbacks": [tracker]})

print(tracker.prompt_tokens)      # → 1234
print(tracker.completion_tokens)  # → 567
print(tracker.total)              # → 1801

HarnessGraph 在 run() 時自動注入 TokenTracker，ChatScreen 透過 harness.token_tracker.total 讀取並更新 UI。