自主 AI Agent 開發實戰指南：2026 年從概念到生產部署的完整攻略

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Optional

# 定義 Agent 的狀態
class MeetingState(TypedDict):
    meetings_today: List[dict]
    current_meeting: Optional[dict]
    transcript: Optional[str]
    summary: Optional[str]
    action_items: List[str]
    errors: List[str]

# 定義節點
def fetch_calendar(state: MeetingState):
    """Step 1: 讀取行事曆"""
    meetings = calendar_api.get_today_events()
    return {"meetings_today": meetings}

def join_and_transcribe(state: MeetingState):
    """Step 2: 加入會議並取得文字稿"""
    meeting = state["meetings_today"][0]
    transcript = meeting_service.join_and_get_transcript(meeting["id"])
    return {"current_meeting": meeting, "transcript": transcript}

def summarize_meeting(state: MeetingState):
    """Step 3: LLM 總結會議"""
    prompt = f"""以下是會議文字稿，請用繁體中文整理：
    1. 會議重點（3-5 點）
    2. 決策事項
    3. 爭議點（若有）
    
    文字稿：{state['transcript'][:8000]}"""
    
    summary = llm.invoke(prompt)
    return {"summary": summary}

def extract_actions(state: MeetingState):
    """Step 4: 提取待辦事項"""
    prompt = f"""從以下會議摘要中提取所有待辦事項，
    格式：誰負責 | 要做什麼 | 截止時間
    
    摘要：{state['summary']}"""
    
    actions = llm.invoke(prompt)
    return {"action_items": parse_actions(actions)}

def write_to_trello(state: MeetingState):
    """Step 5: 寫入 Trello"""
    for item in state["action_items"]:
        trello_api.create_card(
            list_id="會議待辦",
            name=item["task"],
            description=f"負責人：{item['owner']}\n截止：{item['deadline']}"
        )
    return {}

def send_summary_email(state: MeetingState):
    """Step 6: 寄送摘要"""
    email_body = f"""
    會議摘要：{state['summary']}
    
    待辦事項：
    {chr(10).join(f'- {item}' for item in state['action_items'])}
    """
    
    for attendee in state["current_meeting"]["attendees"]:
        email_api.send(attendee, "會議摘要", email_body)
    return {}

# 建構 Graph
builder = StateGraph(MeetingState)

builder.add_node("fetch_calendar", fetch_calendar)
builder.add_node("join_and_transcribe", join_and_transcribe)
builder.add_node("summarize_meeting", summarize_meeting)
builder.add_node("extract_actions", extract_actions)
builder.add_node("write_to_trello", write_to_trello)
builder.add_node("send_summary_email", send_summary_email)

builder.set_entry_point("fetch_calendar")
builder.add_edge("fetch_calendar", "join_and_transcribe")
builder.add_edge("join_and_transcribe", "summarize_meeting")
builder.add_edge("summarize_meeting", "extract_actions")
builder.add_edge("extract_actions", "write_to_trello")
builder.add_edge("write_to_trello", "send_summary_email")
builder.add_edge("send_summary_email", END)

graph = builder.compile()

def robust_call(tool_func, retries=3, fallback=None):
    """通用的重試裝飾器"""
    for attempt in range(retries):
        try:
            return tool_func()
        except APIError as e:
            if e.status_code == 429:  # Rate limit
                time.sleep(2 ** attempt)  # 指數退避
            elif e.status_code == 503:  # 服務暫時不可用
                time.sleep(5)
            else:
                if attempt == retries - 1:
                    return fallback if fallback else f" 失敗：{str(e)}"
    return fallback

# 在 Graph 中加入條件邊（Conditional Edge）
def should_retry(state: MeetingState):
    """檢查是否需要重試"""
    if state.get("errors") and len(state["errors"]) < 3:
        return "retry"
    return "continue"

from crewai import Agent, Task, Crew

# 定義三個專業 Agent
researcher = Agent(
    role="市場研究員",
    goal="收集並分析最新市場資料",
    backstory="擅長從多個來源交叉驗證資訊",
    tools=[web_search, data_analyzer],
    verbose=True
)

writer = Agent(
    role="報告撰寫員",
    goal="將研究結果轉化為易讀的報告",
    backstory="曾擔任科技媒體主編，擅長化繁為簡",
    allow_delegation=True
)

reviewer = Agent(
    role="品質審查員",
    goal="確保報告的準確性與完整性",
    backstory="十年經驗的資深分析師",
    tools=[fact_checker]
)

# 定義任務
research_task = Task(
    description="分析 2026 年 AI Agent 市場趨勢",
    agent=researcher,
    expected_output="結構化的市場分析報告"
)

write_task = Task(
    description="根據研究報告撰寫部落格文章",
    agent=writer,
    expected_output="一篇約 2000 字的繁體中文部落格文章"
)

review_task = Task(
    description="審查部落格文章的準確性",
    agent=reviewer,
    expected_output="審查意見與修改建議"
)

# 組裝
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    verbose=True,
    process="sequential"  # 依序執行
)

result = crew.kickoff()

from langgraph.checkpoint import PostgresSaver

# 使用 PostgreSQL 儲存 Agent 狀態
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/agent_db"
)

graph = builder.compile(checkpointer=checkpointer)

# 每個 session 都有自己的記憶空間
config = {"configurable": {"thread_id": "user_123_session_456"}}
result = graph.invoke(initial_state, config)

from langsmith import traceable
import logging

@traceable(run_type="chain")
def agent_pipeline(user_input: str):
    """完整 Agent 流程，自動追蹤到 LangSmith"""
    logging.info(f"收到用戶輸入: {user_input[:100]}...")
    
    result = graph.invoke({"user_input": user_input})
    
    logging.info(f"Agent 輸出: {result[:200]}...")
    return result