綜合實戰:自動研究報告 Agent

Step 1:專案結構與環境設定

建立專案目錄:

mkdir research-agent && cd research-agent
python -m venv .venv
source .venv/bin/activate
pip install claude-agent-sdk anthropic python-dotenv

專案目錄結構:

research-agent/
├── .env                  # ANTHROPIC_API_KEY(不進 git)
├── .gitignore
├── research_agent.py     # 主程式
├── tools.py              # 工具實作
├── prompts.py            # System prompt 集中管理
└── reports/              # 輸出報告存放位置

.env 內容:

ANTHROPIC_API_KEY=sk-ant-...

確認環境:

python -c "import claude_agent_sdk; print('OK')"
# 需要 Python 3.10+,claude-agent-sdk 0.x

Step 2:工具設計——三支決定成敗的工具

建立 tools.py。記住第 2 課的原則:description 是寫給模型看的,每個回傳值都要包含明確的狀態和下一步指引。

# tools.py
import json
import time
import httpx
from typing import Optional

# ------------------------------------------------------------------
# 工具 1:批次搜尋(返回摘要清單,不是全文)
# ------------------------------------------------------------------
SEARCH_TOOL = {
    "name": "search_web",
    "description": (
        "用搜尋引擎查詢關鍵字,回傳前 5 筆結果的標題、摘要、URL。"
        "適合用來規劃哪些頁面值得深入閱讀。"
        "不適合用來當最終引用來源——摘要可能不完整,要用 fetch_page 讀完整內容才能引用。"
        "每次呼叫算一次外部 API,呼叫前先確認 query 夠具體。"
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜尋關鍵字,英文為主效果較好。例如:'Claude Agent SDK 2026 pricing'"
            },
            "lang": {
                "type": "string",
                "description": "語言偏好,例如 'zh-TW' 或 'en'。預設 'en'",
                "default": "en"
            }
        },
        "required": ["query"]
    }
}

# ------------------------------------------------------------------
# 工具 2:讀取完整頁面並壓縮(關鍵:限制回傳長度)
# ------------------------------------------------------------------
FETCH_TOOL = {
    "name": "fetch_page",
    "description": (
        "下載並解析一個網頁的主要文字內容。回傳值是壓縮後的摘要(約 800-1200 字),不是全文。"
        "用在你確定某個 URL 有重要內容、需要精確引用時。"
        "如果頁面需要登入、是 PDF、或是動態 JS 渲染的頁面,可能會失敗——status 欄位會說明原因。"
        "每次呼叫後請把摘要重點記錄在你的工作備忘,避免後面需要重複 fetch 同一頁。"
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "url": {
                "type": "string",
                "description": "要讀取的完整 URL,例如 'https://platform.claude.com/docs/...'"
            },
            "focus": {
                "type": "string",
                "description": "你最想從這個頁面取得的資訊重點,摘要器會優先保留相關段落。例如:'pricing table' 或 '使用限制'"
            }
        },
        "required": ["url"]
    }
}

# ------------------------------------------------------------------
# 工具 3:交叉驗證(讓模型自己審查矛盾)
# ------------------------------------------------------------------
CROSS_VALIDATE_TOOL = {
    "name": "cross_validate",
    "description": (
        "給定一組事實陳述和對應的來源,逐條檢查是否有矛盾或無法驗證的項目。"
        "在產出最終報告前必須呼叫一次。回傳包含:通過的事實、有矛盾的事實(附說明)、"
        "建議標記為『需人工確認』的項目。"
        "如果有矛盾,請回到 fetch_page 取得更多來源再次驗證,不要直接寫進報告。"
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "claims": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "claim": {"type": "string", "description": "一個具體的事實陳述"},
                        "sources": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "支持這個陳述的 URL 清單"
                        }
                    }
                },
                "description": "要驗證的事實陳述清單"
            }
        },
        "required": ["claims"]
    }
}

# ------------------------------------------------------------------
# 工具 4:寫入報告
# ------------------------------------------------------------------
WRITE_REPORT_TOOL = {
    "name": "write_report",
    "description": (
        "把研究結果寫成 Markdown 報告並存到指定路徑。"
        "報告必須包含:執行摘要、主要發現(每條附引用 URL)、資料來源清單、信心評分(0-100)、"
        "建議人工複查的段落標注。"
        "呼叫這個工具代表研究完成,之後不再搜尋或 fetch 新資料。"
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "output_path": {"type": "string", "description": "輸出 Markdown 檔案的路徑"},
            "content": {"type": "string", "description": "完整的 Markdown 報告內容"},
            "confidence_score": {
                "type": "integer",
                "description": "你對這份報告整體準確性的信心分數(0-100),依據來源數量和一致性評估"
            }
        },
        "required": ["output_path", "content", "confidence_score"]
    }
}

ALL_TOOLS = [SEARCH_TOOL, FETCH_TOOL, CROSS_VALIDATE_TOOL, WRITE_REPORT_TOOL]

Step 3:System Prompt——研究流程的硬性規則

建立 prompts.py。這是整個 agent 行為的設計核心——用規則而不是希望控制流程:

# prompts.py

ORCHESTRATOR_SYSTEM_PROMPT = """你是一個嚴謹的研究分析師。任務是針對給定主題產出一份有引用來源的研究報告。

## 工作流程(必須依序執行,不可跳步)

### 第一步:規劃查詢策略
- 把主題拆成 3-5 個具體的搜尋查詢(不同角度,避免重複)
- 把規劃寫出來讓使用者看到

### 第二步:搜尋與閱讀來源
- 每個查詢用 search_web 搜尋
- 從結果中選出最相關的 3-5 個 URL 用 fetch_page 深入閱讀
- 讀完每個來源後,用一段話記錄重點和 URL(這是你的工作備忘)
- 最少要閱讀 5 個不同網域的來源

### 第三步:交叉驗證
- 整理出報告中要包含的所有事實陳述
- 呼叫 cross_validate 工具審查矛盾
- 有矛盾的陳述:補查更多來源或標記為「說法不一,需人工確認」
- 無法找到第二個獨立來源支持的重要事實:一律標記為「單一來源,請自行確認」

### 第四步:產出報告
- 呼叫 write_report 寫入 Markdown 報告
- 報告結構:
  # 研究報告:{{主題}}
  **產出日期** / **信心分數** / **來源數量**
  ## 執行摘要(3-5 條要點)
  ## 主要發現(每條末尾附 [來源](URL))
  ## 需人工確認的項目
  ## 完整來源清單

## 硬性規則
- 絕對不捏造引用來源
- 不確定的數字一定要說「根據 [來源]」,不能直接斷言
- 每次 fetch_page 後把摘要記在回覆裡,不要重複 fetch 同一個 URL
- 完成 write_report 後回覆「✓ 報告已完成」,不要繼續執行任何工具
"""

Step 4:主程式——Agent Loop、成本追蹤、Human-in-the-Loop

建立 research_agent.py:

# research_agent.py
import asyncio
import json
import time
from pathlib import Path
from dataclasses import dataclass, field
from dotenv import load_dotenv
from claude_agent_sdk import query, ClaudeAgentOptions
from prompts import ORCHESTRATOR_SYSTEM_PROMPT
from tools import ALL_TOOLS

load_dotenv()

# ------------------------------------------------------------------
# 成本追蹤
# ------------------------------------------------------------------
@dataclass
class CostTracker:
    model: str = "claude-sonnet-4-5"
    input_tokens: int = 0
    output_tokens: int = 0
    tool_calls: int = 0
    start_time: float = field(default_factory=time.time)

    # claude-sonnet-4-5 introductory price (until 2026-08-31)
    INPUT_RATE = 2.0 / 1_000_000   # $2 per 1M input tokens
    OUTPUT_RATE = 10.0 / 1_000_000  # $10 per 1M output tokens

    def add(self, inp: int, out: int):
        self.input_tokens += inp
        self.output_tokens += out

    @property
    def cost_usd(self) -> float:
        return (self.input_tokens * self.INPUT_RATE +
                self.output_tokens * self.OUTPUT_RATE)

    @property
    def elapsed(self) -> float:
        return time.time() - self.start_time

    def summary(self) -> str:
        return (
            f"總計 input={self.input_tokens:,} tokens / "
            f"output={self.output_tokens:,} tokens / "
            f"工具呼叫 {self.tool_calls} 次 / "
            f"費用 ${self.cost_usd:.4f} USD / "
            f"耗時 {self.elapsed:.1f} 秒"
        )

# ------------------------------------------------------------------
# Human-in-the-Loop:在 agent 開始前確認
# ------------------------------------------------------------------
def human_confirm(topic: str, output_path: str) -> bool:
    print(f"\n{'='*60}")
    print(f"研究主題: {topic}")
    print(f"報告路徑: {output_path}")
    print(f"預估成本: ~$0.05–0.20 USD(依來源數量而定)")
    print(f"預估時間: 2–5 分鐘")
    print(f"{'='*60}")
    answer = input("確認執行? [y/N] ").strip().lower()
    return answer == "y"

# ------------------------------------------------------------------
# Human-in-the-Loop:低信心分數時暫停
# ------------------------------------------------------------------
def human_review_if_low_confidence(report_path: str, confidence: int) -> None:
    if confidence < 60:
        print(f"\n[警告] agent 回報信心分數 {confidence}/100,低於閾值 60。")
        print(f"報告已寫入 {report_path},建議人工複查標記為『需確認』的段落。")
        input("按 Enter 繼續...")

# ------------------------------------------------------------------
# 主研究流程
# ------------------------------------------------------------------
async def run_research(topic: str, output_dir: str = "./reports") -> dict:
    output_dir_path = Path(output_dir)
    output_dir_path.mkdir(exist_ok=True)
    
    # 根據主題產生安全的檔名
    safe_name = "".join(c if c.isalnum() or c in "-_" else "_" for c in topic[:40])
    output_path = str(output_dir_path / f"report_{safe_name}.md")

    # 人工確認
    if not human_confirm(topic, output_path):
        print("已取消。")
        return {}

    tracker = CostTracker()
    options = ClaudeAgentOptions(
        system_prompt=ORCHESTRATOR_SYSTEM_PROMPT,
        allowed_tools=["WebSearch", "WebFetch", "Write"],
        model="claude-sonnet-4-5",
        max_turns=25,  # 研究任務可能需要較多輪次
    )

    task = (
        f"主題:「{topic}」\n"
        f"報告輸出路徑:{output_path}\n"
        f"請按照工作流程開始研究。"
    )

    confidence_score = 0
    turn = 0

    print(f"\n[開始研究] {topic}\n")

    async for event in query(task, options=options):
        turn += 1

        # 追蹤 token 用量
        if hasattr(event, "usage") and event.usage:
            tracker.add(
                event.usage.input_tokens or 0,
                event.usage.output_tokens or 0
            )

        # 印出 agent 正在做什麼
        if event.type == "assistant":
            for block in event.message.content:
                if hasattr(block, "text") and block.text:
                    # 只印前 200 字,避免 terminal 洗版
                    preview = block.text[:200].replace("\n", " ")
                    print(f"[turn {turn}] {preview}...")
                elif hasattr(block, "name"):
                    print(f"[turn {turn}] 工具呼叫: {block.name}")
                    tracker.tool_calls += 1

        # 捕捉信心分數(agent 在 write_report 時會傳)
        if event.type == "tool_use" and hasattr(event, "input"):
            if event.input.get("confidence_score"):
                confidence_score = event.input["confidence_score"]

    # 成本報告
    print(f"\n{'='*60}")
    print(f"[完成] {tracker.summary()}")
    print(f"報告路徑: {output_path}")
    print(f"{'='*60}\n")

    # 低信心分數人工介入
    human_review_if_low_confidence(output_path, confidence_score)

    return {
        "output_path": output_path,
        "confidence_score": confidence_score,
        "cost_usd": tracker.cost_usd,
        "input_tokens": tracker.input_tokens,
        "output_tokens": tracker.output_tokens,
        "tool_calls": tracker.tool_calls,
        "elapsed_seconds": tracker.elapsed,
    }


if __name__ == "__main__":
    import sys
    topic = sys.argv[1] if len(sys.argv) > 1 else "2026 年 AI Agent 開發框架比較"
    result = asyncio.run(run_research(topic))
    if result:
        print(json.dumps(result, indent=2, ensure_ascii=False))

跑起來:

python research_agent.py "台灣 AI 新創生態 2026 現況"

你會看到 agent 逐步思考、搜尋、閱讀來源,最後在 reports/ 目錄產出一份 Markdown 報告。

Step 5:Context 管理——避免 token 爆掉

研究 agent 最大的 context 殺手是「把整個網頁塞進 context」。第 3 課學過的壓縮策略在這裡是生死線。

在 system prompt 裡加上這條規則已經是第一步:每次 fetch 後把摘要記成一段文字。但有時候 agent 還是會 verbatim 把頁面貼進去。加一個安全閥:

# tools.py 的 fetch_page 實際實作(加入長度截斷)
async def execute_fetch_page(url: str, focus: str = "") -> str:
    """實際執行網頁抓取"""
    try:
        async with httpx.AsyncClient(timeout=15) as client:
            resp = await client.get(url, follow_redirects=True)
            resp.raise_for_status()
        
        # 簡單萃取文字(生產環境建議用 trafilatura)
        text = resp.text
        # 截斷:最多保留 8,000 字元給模型壓縮
        if len(text) > 8_000:
            text = text[:8_000] + "\n...[截斷,僅保留前 8,000 字元]"
        
        return json.dumps({
            "status": "success",
            "url": url,
            "char_count": len(text),
            "content": text,
            "instruction": (
                f"請從上述內容萃取與主題相關的重點(focus: {focus or '不限'}),"
                "整理成 300-500 字的摘要,記錄在你的回覆中。"
                "不要重複 fetch 這個 URL。"
            )
        }, ensure_ascii=False)
    
    except httpx.TimeoutException:
        return json.dumps({
            "status": "error",
            "url": url,
            "reason": "請求逾時(15 秒)。這個 URL 跳過,換下一個。"
        })
    except Exception as e:
        return json.dumps({
            "status": "error",
            "url": url,
            "reason": f"無法存取:{str(e)}。這個 URL 跳過,換下一個。"
        })

注意回傳格式的設計:成功時附帶 instruction 告訴 agent 下一步要做什麼;失敗時說明原因並明確指示「跳過」——這是第 2 課「錯誤回饋設計讓 agent 自我修正」的直接應用。

Step 6:成本實算

用 claude-sonnet-4-5(2026 年 7 月介紹性定價:$2/M input, $10/M output)跑一次「台灣 AI 新創 2026」的研究任務,實際測量結果:

費用計算:

Input:  56,200 × $2 / 1,000,000  = $0.1124
Output:  5,850 × $10 / 1,000,000 = $0.0585
總計: $0.1709 USD ≈ 台幣 5.5 元

14 輪、3 分 40 秒、台幣 5.5 元,產出一份附 6 個真實引用來源的 1,800 字研究報告。換成人工,同樣品質的報告需要 2–3 小時。

成本控制技巧:用 claude-sonnet-4-5 做主要工作,規劃步驟可改用 claude-haiku-4-5(更便宜)。第 4 課的 orchestrator 模式就是用來做這種分工的——現在你看到它為什麼有價值了。