Architecture — Harness-Sample

AI Agent Harness：Terminal UI + Planner–Executor–Reviewer 三角架構，由 LangGraph 驅動。

Tech Stack

層	技術
UI	Textual — TUI framework
Agent Orchestration	LangGraph `StateGraph`
LLM Abstraction	LangChain (`langchain-openai`, `langchain-anthropic`, `langchain-google-genai`)
LLM Providers	OpenAI / Anthropic / Gemini / Custom (任何 OpenAI-compatible endpoint)
Config 持久化	JSON file at `~/.harness_config.json`
Token 追蹤	LangChain `BaseCallbackHandler`
ASCII Banner	pyfiglet
Env 管理	python-dotenv
Python	3.12+
套件管理	uv (`pyproject.toml`)

目錄結構

Harness-Sample/
├── main.py                   # 入口：呼叫 textual_app.main()
├── textual_app.py            # 薄層 re-export，讓 main.py 保持乾淨
├── pyproject.toml            # 依賴宣告
├── .env                      # API keys（不進 git）
├── prompts/
│   ├── planner.md            # PlannerAgent system prompt
│   ├── executor.md           # ExecutorAgent system prompt
│   └── reviewer.md           # ReviewerAgent system prompt
├── src/
│   ├── agent/
│   │   ├── base_langraph_agent.py   # 抽象基底 + HarnessState TypedDict
│   │   ├── planner_agent.py         # 拆解任務 → plan JSON
│   │   ├── executor_agent.py        # 執行單一步驟 → result JSON
│   │   └── reviewer_agent.py        # 評估輸出 → verdict JSON
│   ├── graph/
│   │   └── harness_graph.py         # LangGraph 主圖：串接三個 agent
│   ├── service/
│   │   ├── config_service.py        # 讀寫 ~/.harness_config.json
│   │   └── i18n_service.py          # zh / en 字串對照表
│   ├── ui/
│   │   ├── app.py                   # HarnessChatApp（Textual App）
│   │   ├── constants.py             # 主題、slash commands、model 清單
│   │   └── screens/
│   │       ├── chat_screen.py       # 主聊天畫面
│   │       └── provider_screen.py   # 首次啟動設定畫面
│   └── utils/
│       ├── llm_factory.py           # 依 provider 建構 LangChain LLM
│       └── token_tracker.py         # Callback：累計 prompt/completion tokens
├── tests/
│   └── test_harness_graph.py        # HarnessGraph routing 單元測試
└── dev/
    └── architecture.md              # 本文件（原始版）

資料流

啟動流程

main.py
  └─ textual_app.py
       └─ src/ui/app.py :: HarnessChatApp.on_mount()
            ├─ ~/.harness_config.json 存在 → ChatScreen
            └─ 不存在 → ProviderScreen（選 lang / provider）→ ChatScreen

一次對話的完整流程

用戶在 TextArea 輸入 → Enter
  └─ ChatScreen.action_submit_input()
       └─ @work(thread=True)  ← 背景執行，不阻塞 UI
            └─ HarnessGraph.run(user_request, session_context, thread_id)
                 └─ LangGraph.invoke()
                      │
                      ▼
                 [planner node]
                   PlannerAgent.run(state)
                   └─ LLM.invoke(system_prompt + user_request JSON)
                   └─ parse JSON → plan{task_goal, steps[], success_criteria[]}
                      │
                      ▼ (loop per step)
                 [executor node]
                   ExecutorAgent.run(state)
                   └─ LLM.invoke(system_prompt + current_step JSON)
                   └─ parse JSON → {status, result_summary, tool_outputs[]}
                   └─ current_step_index += 1
                      │
                      ▼
                 [reviewer node]
                   ReviewerAgent.run(state)
                   └─ LLM.invoke(system_prompt + all actor_outputs JSON)
                   └─ parse JSON → {verdict, score, issues[], next_action}
                      │
                      ├─ verdict == "pass" → [finalize]
                      ├─ retry_count >= max_retries → [finalize]
                      ├─ next_action == "replan" → [count_retry] → [planner]
                      └─ next_action == "retry_actor" → [count_retry] → [executor]
                      │
                      ▼
                 [finalize node]
                   組合 final_output
                      │
                      ▼
            call_from_thread(_show_result)

LangGraph 節點圖

START
  │
  ▼
planner ──────────────────────────────────────┐
  │                                           │ (replan)
  ▼                                           │
executor ◄──── count_retry ◄── reviewer ──────┘
  │ (all steps done)      │
  ▼                       ▼
reviewer               finalize
                          │
                         END

重要設計決策

1. `HarnessState` 作為單一狀態來源

HarnessState（TypedDict）定義在 base_langraph_agent.py，所有 graph node 共用同一份 state schema。

2. 每個 Agent 本身也是 LangGraph

BaseLangGraphAgent 在 __init__ 時就 compile 一個小 StateGraph。設計上允許未來每個 agent 內部再擴充多節點流程。

3. JSON-only LLM 輸出 + 硬編 fallback

所有 agent 強制要求 LLM 輸出合法 JSON。parse_json_response() 支援 markdown code block 剝除與 brace extraction。

4. `@work(thread=True)` 非同步執行

HarnessGraph.run() 是同步阻塞呼叫。Textual 的 @work(thread=True) 把它推到背景執行緒，UI 保持響應。

5. `status_callback` 解耦 agent 狀態顯示

HarnessGraph 接受可選的 status_callback(agent, msg)。UI 傳入 _on_agent_status()，CLI 模式印到 stdout。

6. Config 存在 home 目錄

~/.harness_config.json 不在 project 目錄，避免進 git。

7. System prompt 從 `prompts/` 讀取

修改 prompt 不需動 Python 程式碼。

8. Token 追蹤用 LangChain Callback

TokenTracker 注入到 graph.invoke(config={"callbacks": [...]})，累計 prompt / completion tokens。

環境變數（`.env`）

變數	Provider
`OPENAI_API_KEY`, `OPENAI_MODEL_NAME`	openai
`ANTHROPIC_API_KEY`, `ANTHROPIC_MODEL_NAME`	anthropic
`GEMINI_API_KEY`, `GEMINI_MODEL_NAME`	gemini
`CUSTOM_API_KEY`, `CUSTOM_MODEL_NAME`, `CUSTOM_BASE_URL`	custom

custom provider 使用 ChatOpenAI + 自訂 base_url，相容任何 OpenAI-compatible API。

執行方式

# 啟動 TUI
uv run python main.py

# 執行測試
uv run pytest tests/ -v