進階應用｜AI Agent 開發教學

Multi-agent 架構設計原則

當單一 Agent 面對複雜任務時，往往難以在一個 context window 內完成所有推理與行動。Multi-agent 架構透過職責分離與並行執行，解決這個限制。設計時應遵循以下原則：

單一職責（Single Responsibility）：每個 Agent 只負責一個明確的子任務，例如「研究員 Agent」只做資料收集，「撰稿 Agent」只做文字生成，避免一個 Agent 承擔過多角色。
最小溝通原則：Agent 間訊息傳遞應攜帶完成任務所需的最少資訊，避免將整個對話歷史傳遞給每個子 Agent，減少 Token 消耗。
明確的輸入輸出契約：定義每個 Agent 接收的輸入格式與輸出格式（建議用 Pydantic 模型），讓上下游 Agent 能可靠地解析彼此的結果。
Orchestrator / Worker 模式：由一個主 Agent（Orchestrator）負責任務分解與結果整合，多個子 Agent（Worker）並行執行具體工作，最後由 Orchestrator 彙整輸出。
避免循環依賴：Agent 之間的依賴關係應是有向無環圖（DAG），若出現 A → B → A 的循環，需引入終止條件或最大迭代次數。

設計 Multi-agent 系統時，先畫出資料流圖：哪些 Agent 產生資料、哪些 Agent 消費資料、哪些可以並行。可觀察性從設計階段就要考慮進去，而非事後補加。

Agent 評估與測試方法

Agent 的非確定性讓傳統單元測試難以直接套用，需要針對性的評估策略。

from langchain.evaluation import load_evaluator
from langchain_openai import ChatOpenAI

# --- 方法一：LLM-as-Judge（用另一個 LLM 評分）---
evaluator = load_evaluator(
    "criteria",
    llm=ChatOpenAI(model="gpt-4o"),
    criteria={
        "correctness": "答案是否事實正確？",
        "completeness": "答案是否完整回應問題？",
        "conciseness": "答案是否簡潔不冗餘？"
    }
)

result = evaluator.evaluate_strings(
    prediction="LangChain 是一個 LLM 應用框架，提供 Chain 和 Agent 等核心元件。",
    input="什麼是 LangChain？",
    reference="LangChain 是開源的 LLM 應用開發框架。"
)
print(result)  # {"score": 1, "reasoning": "..."}

# --- 方法二：RAG 評估（RAGAS 框架）---
# pip install ragas
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

dataset = {
    "question": ["LangChain 支援哪些向量資料庫？"],
    "answer": ["LangChain 支援 FAISS、Chroma、Pinecone 等。"],
    "contexts": [["LangChain 整合了超過 50 種向量資料庫..."]],
    "ground_truth": ["LangChain 支援 FAISS、Chroma、Pinecone、Weaviate 等。"]
}
score = evaluate(dataset, metrics=[faithfulness, answer_relevancy])
print(score)

# --- 方法三：確定性測試（針對工具呼叫路徑）---
def test_agent_calls_search_tool():
    """驗證 Agent 在需要查詢資料時確實呼叫了搜尋工具"""
    call_log = []
    original_search = search_tool.func

    def mock_search(query):
        call_log.append(query)
        return "模擬搜尋結果"

    search_tool.func = mock_search
    agent.invoke({"input": "搜尋最新的 AI 新聞"})
    assert len(call_log) > 0, "Agent 應該呼叫搜尋工具"
    search_tool.func = original_search

評估應覆蓋三個層次：工具呼叫正確性（是否選對工具）、答案品質（是否準確完整）、端對端任務完成率（是否達成使用者目標）。建議維護一個「黃金測試集」，每次版本更新都自動跑評估。

Memory 管理策略

記憶是 Agent 能夠跨對話保持上下文的關鍵機制，不同場景需要不同的記憶策略。

短期記憶（Short-term / Buffer Memory）：保留當前對話的完整歷史，直接注入 Prompt。優點是保真度高，缺點是長對話會超出 context window。適合輪次較少的對話場景。
摘要記憶（Summary Memory）：當對話超過一定長度時，自動用 LLM 將舊對話壓縮為摘要。適合長期對話，但摘要過程本身消耗 Token，且可能遺失細節。
情節記憶（Episodic Memory）：記錄過去完整的「任務事件」，例如「上次幫用戶分析了哪份報告」，以時間序列形式儲存，查詢時依時間或相關性檢索。
語意記憶（Semantic Memory）：儲存通用知識與使用者偏好（「用戶偏好 Python 風格的程式碼」），以向量形式存入資料庫，查詢時用語意相似度檢索。是跨會話持久化記憶的首選。

from langchain.memory import (
    ConversationBufferMemory,        # 短期：保留全部
    ConversationSummaryBufferMemory, # 摘要 + Buffer 混合策略
    ConversationTokenBufferMemory,   # 依 Token 數截斷
)
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

# 混合策略：近期保留原文，超過 token 限制的部分自動摘要
memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,   # 超過 2000 token 就開始摘要
    return_messages=True
)

# 語意記憶：將使用者偏好持久化到向量資料庫
from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

vectorstore = FAISS.from_texts(["初始化"], OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
semantic_memory = VectorStoreRetrieverMemory(retriever=retriever)

# 儲存使用者偏好
semantic_memory.save_context(
    {"input": "我偏好簡潔的程式碼，不喜歡過多註解"},
    {"output": "已記住您的偏好"}
)

# 下次對話時，相關偏好會自動注入 Prompt
relevant = semantic_memory.load_memory_variables({"prompt": "幫我寫一段排序程式碼"})

Tool Design 原則

工具設計的品質直接影響 Agent 的可靠性。設計不良的工具是 Agent 失敗的最常見原因。

冪等性（Idempotency）：相同輸入多次呼叫應得到相同結果，且不產生副作用。讀取操作天然冪等，寫入操作需額外設計（例如用唯一 ID 防重複）。Agent 在不確定是否成功時可能重試，冪等性是防止重複寫入的保障。
清晰的 docstring：工具的 description 是 LLM 決定「何時呼叫此工具」的唯一依據，應說明用途、適用場景、輸入格式與預期輸出。模糊的 description 是工具被誤用的根源。
輸入驗證：用 Pydantic BaseModel 定義輸入 schema，在工具內部做型別檢查，讓錯誤訊息對 LLM 有意義（LLM 可根據錯誤訊息自我修正）。
結構化輸出：回傳 JSON 而非自由格式文字，讓上游 Agent 或 Orchestrator 能夠可靠地解析結果。

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type
import json

# 定義輸入 Schema（Pydantic 自動驗證型別）
class SearchInput(BaseModel):
    query: str = Field(description="搜尋關鍵字，應為具體的問題或主題")
    max_results: int = Field(default=5, ge=1, le=20,
                             description="回傳結果數量，預設 5，最多 20")

class SearchTool(BaseTool):
    name: str = "web_search"
    description: str = (
        "搜尋網路上的最新資訊。"
        "適合用於查詢近期新聞、技術文件、特定主題的背景資料。"
        "不適合用於數學計算或固定事實查詢。"
    )
    args_schema: Type[BaseModel] = SearchInput

    def _run(self, query: str, max_results: int = 5) -> str:
        try:
            # 實際呼叫搜尋 API
            results = self._call_search_api(query, max_results)
            # 回傳結構化 JSON，方便 Agent 解析
            return json.dumps({
                "status": "success",
                "query": query,
                "results": results
            }, ensure_ascii=False)
        except Exception as e:
            # 錯誤訊息要對 LLM 有意義，讓它知道如何重試
            return json.dumps({
                "status": "error",
                "error_type": type(e).__name__,
                "message": str(e),
                "suggestion": "請嘗試使用更簡短的關鍵字重試"
            }, ensure_ascii=False)

    def _call_search_api(self, query, max_results):
        # 實際 API 呼叫實作
        return [{"title": "範例結果", "url": "https://example.com"}]

錯誤處理與重試機制

生產環境中，API 超時、速率限制、網路抖動是常態。健壯的 Agent 必須有完善的錯誤處理與重試策略。

from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableConfig
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import openai

# --- 方法一：LangChain 內建重試 ---
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_retry = llm.with_retry(
    retry_if_exception_type=(
        openai.RateLimitError,
        openai.APITimeoutError,
        openai.APIConnectionError
    ),
    stop_after_attempt=3,
    wait_exponential_jitter=True  # 指數退避 + 隨機抖動，避免雷群效應
)

# --- 方法二：tenacity 自訂重試（更細粒度控制）---
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type((openai.RateLimitError, TimeoutError)),
    reraise=True
)
def call_llm_with_retry(prompt: str) -> str:
    return llm.invoke(prompt).content

# --- 方法三：Fallback（降級到備用模型）---
from langchain_anthropic import ChatAnthropic

primary_llm = ChatOpenAI(model="gpt-4o")
fallback_llm = ChatAnthropic(model="claude-3-haiku-20240307")

# 主要模型失敗時自動切換到備用模型
llm_with_fallback = primary_llm.with_fallbacks([fallback_llm])

# --- 超時設定 ---
config = RunnableConfig(
    timeout=30,           # 整體 30 秒超時
    max_concurrency=5     # 最多 5 個並行請求
)
result = llm_with_fallback.invoke("你好", config=config)

成本控制策略

Token 費用在 Agent 應用中容易快速累積，尤其是 Multi-agent 架構或長對話場景。以下策略可有效控制成本：

模型分級使用：簡單任務（分類、摘要、格式轉換）用小模型（GPT-4o-mini、Claude Haiku），複雜推理才用大模型。費用差距可達 10–20 倍。
Prompt 快取（Prompt Caching）：Anthropic Claude 和 OpenAI 均支援 Prompt Cache，對於重複的 System Prompt 或長文件，快取後重複讀取費用降低 90%。
LLM 快取（Exact Match Cache）：完全相同的輸入直接回傳快取結果，適用於常見問題的 FAQ Bot。
Context 壓縮：傳入 LLM 前先對文件做摘要或過濾，只保留與當前問題最相關的段落，減少無效 Token。
設定費用告警：在 OpenAI / Anthropic 後台設定月費用上限，並透過 LangSmith 監控每個 Agent 的 Token 消耗。

成本監控的最佳實踐：在開發階段用 get_openai_callback() 追蹤每次呼叫的 Token 用量，找出最耗費 Token 的步驟，優先優化那些步驟。

from langchain_community.callbacks import get_openai_callback
from langchain_openai import ChatOpenAI
from langchain.globals import set_llm_cache
from langchain_community.cache import SQLiteCache

# 啟用 SQLite 快取（精確匹配快取，開發環境推薦）
set_llm_cache(SQLiteCache(database_path=".langchain_cache.db"))

llm = ChatOpenAI(model="gpt-4o-mini")

# 追蹤單次呼叫的 Token 消耗
with get_openai_callback() as cb:
    result = llm.invoke("解釋一下 RAG 的原理")
    print(f"輸入 Token：{cb.prompt_tokens}")
    print(f"輸出 Token：{cb.completion_tokens}")
    print(f"總費用（USD）：${cb.total_cost:.6f}")

# 第二次相同呼叫直接從快取回傳，費用為 $0
result2 = llm.invoke("解釋一下 RAG 的原理")

生產環境部署注意事項

將 Agent 應用從開發環境遷移到生產環境時，有許多開發階段容易忽略的面向需要處理。

將所有 API Key 移入環境變數或 Secret Manager（AWS Secrets Manager、HashiCorp Vault），絕不硬編碼在程式碼中
設定每個使用者或每個請求的 Rate Limit，防止單一使用者消耗過多資源
為 Agent 設定最大迴圈次數（max_iterations）和最大執行時間，防止無限迴圈
使用 Docker 容器隔離程式碼執行環境，特別是允許 Agent 執行任意程式碼的場景
實作 Input Validation：對使用者輸入做長度限制與內容過濾，防止 Prompt Injection 攻擊
設定 Circuit Breaker：當外部 API 連續失敗達閾值時，暫停呼叫並回傳降級結果
所有 Agent 行為應有完整的 Audit Log，包含輸入、工具呼叫記錄、輸出，方便合規審計

可觀測性與日誌設計

Agent 的非確定性讓傳統應用的監控方式不足以應對，需要專為 LLM 應用設計的可觀測性體系。

import logging
import json
import uuid
from datetime import datetime
from langchain.callbacks.base import BaseCallbackHandler

# 自訂 Callback Handler：記錄完整的 Agent 執行 trace
class StructuredLogHandler(BaseCallbackHandler):
    def __init__(self):
        self.trace_id = str(uuid.uuid4())
        self.logger = logging.getLogger("agent.trace")

    def on_llm_start(self, serialized, prompts, **kwargs):
        self.logger.info(json.dumps({
            "trace_id": self.trace_id,
            "event": "llm_start",
            "timestamp": datetime.utcnow().isoformat(),
            "model": serialized.get("name"),
            "prompt_length": sum(len(p) for p in prompts)
        }))

    def on_tool_start(self, serialized, input_str, **kwargs):
        self.logger.info(json.dumps({
            "trace_id": self.trace_id,
            "event": "tool_start",
            "timestamp": datetime.utcnow().isoformat(),
            "tool": serialized.get("name"),
            "input": input_str[:500]  # 截斷避免日誌過大
        }))

    def on_tool_end(self, output, **kwargs):
        self.logger.info(json.dumps({
            "trace_id": self.trace_id,
            "event": "tool_end",
            "timestamp": datetime.utcnow().isoformat(),
            "output_length": len(str(output))
        }))

    def on_agent_finish(self, finish, **kwargs):
        self.logger.info(json.dumps({
            "trace_id": self.trace_id,
            "event": "agent_finish",
            "timestamp": datetime.utcnow().isoformat(),
            "output": finish.return_values.get("output", "")[:500]
        }))

# 使用方式
from langchain.agents import AgentExecutor
handler = StructuredLogHandler()
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[handler],  # 注入自訂 handler
    verbose=False          # 生產環境關閉 verbose
)

可觀測性工具推薦：LangSmith（LangChain 官方，最易整合）、Langfuse（開源，可自架）、Helicone（代理層方案，零程式碼整合）、OpenTelemetry（標準化分散式追蹤，適合已有 APM 基礎設施的團隊）。建議至少整合其中一個，否則生產環境除錯會非常困難。