# LangChain Sample 05: Multi-Agent with LangGraph
## 목표
LangGraph를 사용하여 여러 에이전트가 협업하는 멀티 에이전트 시스템 구축.
## 필수 설치
```bash
pip install langchain langchain-anthropic langgraph langchain-core
```
---
## 완전한 코드
```python
from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from typing import TypedDict, Annotated
import operator
import json
# ============================================================
# 1. State 정의 (에이전트 간 데이터 공유)
# ============================================================
class AgentState(TypedDict):
"""멀티 에이전트 시스템의 공유 상태"""
task: str # 원본 요청
research_findings: str # 연구원이 수집한 정보
analysis: str # 분석가의 분석 결과
action_plan: str # 실행 계획
messages: Annotated[list, operator.add] # 메시지 로그
# ============================================================
# 2. 도구 정의
# ============================================================
@tool
def search_internet(query: str) -> str:
"""인터넷 검색 시뮬레이션"""
# 실제 환경에서는 실제 검색 API 사용
search_results = {
"AI 시장": "2024년 AI 시장은 600억 달러 규모... AI 기업들이 빠르게 성장 중...",
"클라우드": "클라우드 시장은 연 20% 성장 중... AWS, Google Cloud, Azure가 주도...",
"데이터": "빅데이터 분석 시장이 확대... 데이터 보안이 중요해짐..."
}
for key, value in search_results.items():
if key.lower() in query.lower():
return value
return f"'{query}'에 대한 검색 결과: 정보를 찾을 수 없습니다."
@tool
def analyze_data(data: str) -> str:
"""데이터 분석 시뮬레이션"""
analysis = f"""
분석 결과:
- 핵심 트렌드: 기술 혁신과 시장 성장
- 위험 요소: 경쟁 심화, 규제 강화
- 기회 요소: 신규 시장, 기술 도입 확대
- 데이터 원본: "{data[:100]}..."
"""
return analysis
@tool
def create_action_plan(context: str) -> str:
"""행동 계획 수립"""
plan = f"""
제안된 행동 계획:
1. 시장 조사 진행 (2주)
2. 기술 검토 (1주)
3. 팀 구성 및 교육 (2주)
4. 파일럿 프로젝트 실행 (4주)
5. 평가 및 확대 (2주)
총 소요 기간: 11주
"""
return plan
# ============================================================
# 3. 에이전트 노드 정의
# ============================================================
def researcher_node(state: AgentState):
"""연구원: 정보 수집"""
print("\n🔬 연구원 활성화")
model = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)
model_with_tools = model.bind_tools([search_internet])
# 연구원의 프롬프트
messages = [{
"role": "system",
"content": "당신은 정보 수집 전문 연구원입니다. 주어진 주제에 대해 상세한 정보를 검색하세요."
}, {
"role": "user",
"content": f"다음 주제에 대해 조사해주세요: {state['task']}"
}]
# 도구 호출 루프
response = model_with_tools.invoke(messages)
# 간단히 처리 (실제는 도구 호출 루프 필요)
findings = search_internet.invoke({"query": state['task']})
updated_state = state.copy()
updated_state['research_findings'] = findings
updated_state['messages'].append(f"연구원: {findings}")
print(f"✓ 연구 완료")
return updated_state
def analyst_node(state: AgentState):
"""분석가: 데이터 분석"""
print("\n📊 분석가 활성화")
# 연구원의 결과를 받아 분석
analysis_result = analyze_data.invoke({"data": state['research_findings']})
updated_state = state.copy()
updated_state['analysis'] = analysis_result
updated_state['messages'].append(f"분석가: {analysis_result}")
print(f"✓ 분석 완료")
return updated_state
def planner_node(state: AgentState):
"""계획가: 행동 계획 수립"""
print("\n📋 계획가 활성화")
# 분석가의 결과를 받아 계획 수립
context = f"연구: {state['research_findings']}\n분석: {state['analysis']}"
plan = create_action_plan.invoke({"context": context})
updated_state = state.copy()
updated_state['action_plan'] = plan
updated_state['messages'].append(f"계획가: {plan}")
print(f"✓ 계획 수립 완료")
return updated_state
def summarizer_node(state: AgentState):
"""요약가: 최종 종합"""
print("\n📝 요약가 활성화")
model = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)
summary_prompt = f"""
다음 정보를 바탕으로 최종 보고서를 작성하세요:
원본 요청: {state['task']}
연구 결과:
{state['research_findings']}
분석:
{state['analysis']}
계획:
{state['action_plan']}
최종 보고서:"""
response = model.invoke([{"role": "user", "content": summary_prompt}])
updated_state = state.copy()
updated_state['messages'].append(f"요약가:\n{response.content}")
print(f"✓ 요약 완료\n")
return updated_state
# ============================================================
# 4. Graph 구성 (워크플로우)
# ============================================================
def create_multi_agent_graph():
"""멀티 에이전트 그래프 생성"""
# 그래프 생성
graph = StateGraph(AgentState)
# 노드 추가
graph.add_node("researcher", researcher_node)
graph.add_node("analyst", analyst_node)
graph.add_node("planner", planner_node)
graph.add_node("summarizer", summarizer_node)
# 엣지 (연결)
graph.add_edge(START, "researcher") # 시작 → 연구원
graph.add_edge("researcher", "analyst") # 연구원 → 분석가
graph.add_edge("analyst", "planner") # 분석가 → 계획가
graph.add_edge("planner", "summarizer") # 계획가 → 요약가
graph.add_edge("summarizer", END) # 요약가 → 종료
# 컴파일
return graph.compile()
# ============================================================
# 5. 실행 함수
# ============================================================
def run_multi_agent_system(task: str):
"""멀티 에이전트 시스템 실행"""
print("\n" + "="*70)
print(f"🚀 작업 시작: {task}")
print("="*70)
# 그래프 생성
graph = create_multi_agent_graph()
# 초기 상태
initial_state = AgentState(
task=task,
research_findings="",
analysis="",
action_plan="",
messages=[]
)
# 실행
print("\n⏳ 에이전트 협력 중...\n")
final_state = graph.invoke(initial_state)
# 최종 결과 출력
print("\n" + "="*70)
print("📌 최종 결과")
print("="*70)
for message in final_state['messages']:
print(f"\n{message}")
print("\n" + "="*70)
return final_state
# ============================================================
# 메인 실행
# ============================================================
if __name__ == "__main__":
# 예제 1
run_multi_agent_system("AI 기술을 우리 회사에 도입하려고 합니다. 어떻게 해야 할까요?")
# 예제 2
# run_multi_agent_system("클라우드 마이그레이션 전략을 수립해주세요.")
```
---
## 단계별 설명
### 1단계: State 정의
```python
class AgentState(TypedDict):
task: str
research_findings: str
analysis: str
action_plan: str
messages: Annotated[list, operator.add]
```
**Annotated[list, operator.add]** — 리스트 병합 자동화
- 각 노드가 반환하는 메시지가 자동으로 누적
### 2단계: 노드 정의
```python
def researcher_node(state: AgentState):
# 상태 읽기
task = state['task']
# 처리
findings = search_internet.invoke({"query": task})
# 상태 업데이트
updated_state = state.copy()
updated_state['research_findings'] = findings
updated_state['messages'].append(f"연구원: {findings}")
return updated_state
```
**패턴:**
1. 입력 상태 읽기
2. 처리 로직 실행
3. 상태 복사 및 업데이트
4. 업데이트된 상태 반환
### 3단계: Graph 구성
```python
graph = StateGraph(AgentState)
# 노드 추가
graph.add_node("researcher", researcher_node)
graph.add_node("analyst", analyst_node)
# 엣지 (흐름) 정의
graph.add_edge("researcher", "analyst")
```
**Graph 시각화:**
```
START → researcher → analyst → planner → summarizer → END
↓ ↓ ↓ ↓ ↓ ↓
[입력] [정보수집] [분석] [계획] [요약] [출력]
```
### 4단계: 실행
```python
graph = create_multi_agent_graph()
final_state = graph.invoke(initial_state)
```
---
## 핵심 개념
| 개념 | 설명 |
|------|------|
| **StateGraph** | 노드와 엣지로 워크플로우 구성 |
| **Node** | 에이전트 또는 처리 단위 |
| **Edge** | 노드 간 데이터 흐름 |
| **State** | 에이전트 간 공유 데이터 |
| **Annotated** | 리스트/딕셔너리 자동 병합 |
---
## 커스터마이징 팁
### 조건부 분기 (Conditional Edge)
```python
def route_decision(state: AgentState) -> str:
"""상태에 따라 다음 노드 결정"""
if "긴급" in state['task']:
return "fast_track"
else:
return "normal_track"
graph.add_conditional_edges(
"analyst",
route_decision,
{
"fast_track": "fast_planner",
"normal_track": "planner"
}
)
```
### 반복 처리 (Cycle)
```python
# 승인될 때까지 반복
def needs_approval(state: AgentState) -> str:
if state.get("approved"):
return "end"
else:
return "revise"
graph.add_edge("planner", "review")
graph.add_conditional_edges(
"review",
needs_approval,
{"revise": "planner", "end": END}
)
```
### Human-in-the-Loop
```python
def human_review_node(state: AgentState):
"""사용자 개입"""
print(f"현재 계획: {state['action_plan']}")
feedback = input("승인하시겠습니까? (yes/no): ")
state['approved'] = feedback.lower() == 'yes'
return state
graph.add_node("human_review", human_review_node)
```
---
## 다음 단계
👉 [[codes/agent-frameworks/langchain/samples/production-setup|Production Setup]] — LangSmith 관측성 및 메모리 저장소 설정