打造一個 Multi-Agent AI 研究團隊（AI Research Team） 是驗證前面所有觀念（Planning、Memory、Reflection、Supervisor、Workflow）最經典也最實用的專案。
這個專案的目標是：給定一個複雜的科技或市場研究主題（例如：「2026 年邊緣 AI 與小語言模型的最新技術突破與產業衝擊」），AI 團隊能夠自動化進行資料調研、邏輯分析、深度撰寫、交叉審查，最後產出專業級的長篇研究報告。
模擬真實頂尖研究機構的組織，這個 Multi-Agent 系統由 5 個專責 Agent 組成：
┌───────────────────────────┐
│ Supervisor (專案總監) │ ◄─── 全域進度控管與任務派發
└─────────────┬─────────────┘
│
┌───────────────┬────────────┼────────────┬───────────────┐
▼ ▼ ▼ ▼ ▼
【Researcher】 【Analyst】 【Writer】 【Reviewer】 【Memory Store】
(資料調研收集) (邏輯交叉比對) (報告撰寫) (品管與反思) (長短期記憶庫)
這個研究團隊的運作並非混亂的自由對話，而是基於嚴格的 State Graph（狀態圖工作流）：
[使用者輸入主題]
│
▼
[Supervisor 規劃大綱與子任務]
│
├─► 1. 派發 [Researcher] ──► 收集資料 ──┐
│ │
├─► 2. 派發 [Analyst] ──► 結構分析 ──┼─► [Writer 撰寫初稿]
│ │ │
└─► 3. 檢查進度與缺口 ────────────────┘ ▼
[Reviewer 審查品質]
│
┌──────────┴──────────┐
[未通過 / 85分以下] [通過 / 90分以上]
│ │
▼ ▼
[退回 Writer 修改] [FINISH 輸出報告]
以下使用 LangGraph 概念與強型別 State，實作一個具備 Supervisor 調度、多 Agent 協同與 Reviewer 反思迴圈 的研究團隊系統：
import json
from typing import TypedDict, List, Optional
from pydantic import BaseModel, Field
from openai import OpenAI
client = OpenAI()
# ==========================================
# 1. 定義研究團隊的「全域共用狀態 (Shared Blackboard State)」
# ==========================================
class ResearchTeamState(TypedDict):
topic: str # 研究主題
research_notes: Optional[str] # Researcher 收集的原始資料
analyzed_insights: Optional[str] # Analyst 提煉的結構化觀點
report_draft: Optional[str] # Writer 產出的報告初稿
review_score: Optional[int] # Reviewer 審查評分 (0-100)
review_feedback: Optional[str] # Reviewer 的修改建議
current_step: str # 當前執行階段 (Supervisor 控管)
iteration_count: int # 審查重試次數防護
# ==========================================
# 2. 定義各個 Agent 的實作邏輯
# ==========================================
class ResearchTeamSystem:
def __init__(self, model: str = "gpt-4o-mini"):
self.model = model
def researcher_node(self, state: ResearchTeamState) -> dict:
"""Researcher: 負責搜尋與調研主題資料"""
print(f"🔍 [Researcher]: 正在調研主題 '{state['topic']}' 的核心技術與背景...")
prompt = f"請針對研究主題：'{state['topic']}'，提供 3-4 個關鍵的技術突破與市場現狀要點。"
res = client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}]
)
return {"research_notes": res.choices[0].message.content, "current_step": "analyze"}
def analyst_node(self, state: ResearchTeamState) -> dict:
"""Analyst: 負責將原始筆記整理成結構化洞察"""
print("📊 [Analyst]: 正在進行數據交叉比對與結構化提煉...")
prompt = (
f"研究主題：{state['topic']}\n"
f"原始調研筆記：\n{state['research_notes']}\n\n"
f"請將上述筆記整理成 3 個具有邏輯深度的結構化分析觀點（含趨勢、挑戰與機會）。"
)
res = client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}]
)
return {"analyzed_insights": res.choices[0].message.content, "current_step": "write"}
def writer_node(self, state: ResearchTeamState) -> dict:
"""Writer: 撰寫正式報告初稿（支援 Reviewer 的 Feedback 修改）"""
print("✍️ [Writer]: 正在撰寫正式研究報告...")
prompt = (
f"主題：{state['topic']}\n"
f"結構化分析觀點：\n{state['analyzed_insights']}\n"
)
if state.get("review_feedback"):
prompt += f"\n⚠️ 上一輪審查未過，主管/審查員修改建議：\n{state['review_feedback']}\n請務必修正。"
res = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一位頂級的科技產業分析師，請使用專業、流暢且具說服力的語調撰寫報告。"},
{"role": "user", "content": prompt}
]
)
return {
"report_draft": res.choices[0].message.content,
"current_step": "review",
"iteration_count": state.get("iteration_count", 0) + 1
}
class ReviewResult(BaseModel):
score: int = Field(description="報告專業品質評分 (0-100)")
feedback: str = Field(description="具體的改進建議或邏輯漏洞批判")
def reviewer_node(self, state: ResearchTeamState) -> dict:
"""Reviewer: 扮演嚴格的 Critic，執行品質審查與 Reflection"""
print("🧐 [Reviewer]: 正在審查報告品質與邏輯嚴密性...")
prompt = (
f"研究主題：{state['topic']}\n"
f"待審查報告初稿：\n{state['report_draft']}\n\n"
f"請嚴格審查：論點是否清晰？有無邏輯跳躍？是否充分反映了分析觀點？\n"
f"若品質優異請給出 85 分以上；若有明顯缺陷請給低分並給予具體修改建議。"
)
completion = client.beta.chat.completions.parse(
model=self.model,
messages=[{"role": "user", "content": prompt}],
response_format=self.ReviewResult
)
result: self.ReviewResult = completion.choices[0].message.parsed
print(f" └─ 審查得分: {result.score} | 評語: {result.feedback}")
return {
"review_score": result.score,
"review_feedback": result.feedback
}
def run(self, topic: str) -> str:
# 初始化狀態
state: ResearchTeamState = {
"topic": topic,
"research_notes": None,
"analyzed_insights": None,
"report_draft": None,
"review_score": None,
"review_feedback": None,
"current_step": "research",
"iteration_count": 0
}
print(f"🎬 [AI Research Team 啟動] 研究主題: {topic}\n" + "="*50)
# 模擬團隊工作流狀態機迴圈
max_loops = 10
loop_count = 0
while loop_count < max_loops:
loop_count += 1
step = state["current_step"]
if step == "research":
res = self.researcher_node(state)
state.update(res)
elif step == "analyze":
res = self.analyst_node(state)
state.update(res)
elif step == "write":
res = self.writer_node(state)
state.update(res)
elif step == "review":
res = self.reviewer_node(state)
state.update(res)
# 條件判斷：若通過或超過重試次數，則結束；否則退回 Writer 重寫
if state["review_score"] >= 85 or state["iteration_count"] >= 3:
print("\n✅ [研究團隊專案完成] 報告通過品管審查！")
return state["report_draft"]
else:
print("🔄 品質未達標，退回 Writer 進行自我修正 (Refinement Loop)...\n")
state["current_step"] = "write"
return state.get("report_draft", "專案執行超時終止。")
# ==========================================
# 4. 執行專案測試
# ==========================================
if __name__ == "__main__":
team = ResearchTeamSystem()
final_report = team.run("2026 年自主 AI Agent 架構的關鍵技術演進與企業級落地挑戰")
print("\n📄 【最終產出研究報告】:\n")
print(final_report)
把這個「研究團隊」推向真正的生產級（Production-ready）應用，可以加入以下擴充：
Researcher 連結 Tavily API、Serper 或 Google Custom Search，讓它能上網抓取最新的即時新聞與技術論文，而非只靠模型的內建常識。Reviewer 通過之後、正式匯出前，加入一個人類介入的 Pause 節點，讓使用者可以手動點擊「核准發布」或「手動微調」。
iThome鐵人賽
看影片追技術