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` 實例

`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。

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。

Module 參考

Agent 層

src/agent/base_langraph_agent.py

HarnessState（TypedDict）

BaseLangGraphAgent（ABC）

src/agent/planner_agent.py

src/agent/executor_agent.py

src/agent/reviewer_agent.py

Graph 層

src/graph/harness_graph.py