DeerFlow 2.0 Multi-Agent 워크플로우 심화 — StateGraph, Plan-Execute, Human-in-the-Loop

from langgraph.graph import StateGraph, START, END

# 상태 정의
class AgentState(TypedDict):
    messages: list[BaseMessage]
    plan: list[dict]
    current_step: int
    results: dict

# 그래프 구성
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("coder", coder_node)
workflow.add_node("reporter", reporter_node)

# 엣지 정의
workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges("supervisor", route_to_agent)
workflow.add_edge("reporter", END)

def supervisor_node(state: AgentState) -> AgentState:
    # 1. 태스크 분석
    task_analysis = llm.invoke(
        f"Analyze this task and determine the best approach: {state['messages'][-1]}"
    )

    # 2. 라우팅 결정
    if task_analysis.requires_research:
        return {"next": "researcher"}
    elif task_analysis.requires_code:
        return {"next": "coder"}
    elif task_analysis.is_complex:
        return {"next": "planner"}  # 복잡한 태스크 → Planner로 위임
    else:
        return {"next": "reporter"}  # 단순 태스크 → 바로 보고서

# Planner가 생성하는 계획 구조
plan = {
    "goal": "AI 에이전트 프레임워크 비교 보고서 작성",
    "steps": [
        {
            "id": 1,
            "agent": "researcher",
            "task": "DeerFlow 2.0 기능 및 아키텍처 조사",
            "depends_on": [],
        },
        {
            "id": 2,
            "agent": "researcher",
            "task": "CrewAI, AutoGen, LangGraph 기능 조사",
            "depends_on": [],  # step 1과 병렬 실행 가능
        },
        {
            "id": 3,
            "agent": "analyst",
            "task": "수집된 데이터를 비교 표로 정리",
            "depends_on": [1, 2],  # step 1, 2 완료 후 실행
        },
        {
            "id": 4,
            "agent": "reporter",
            "task": "최종 비교 보고서 작성",
            "depends_on": [3],
        },
    ],
}

Step 1 (Researcher) ─┐
                     ├→ Step 3 (Analyst) → Step 4 (Reporter)
Step 2 (Researcher) ─┘

{
  "findings": [
    {
      "title": "DeerFlow 2.0 Architecture",
      "summary": "...",
      "source": "https://...",
      "relevance": 0.95
    }
  ],
  "citations": ["[1] https://..."]
}

# Coder 에이전트의 코드 실행 흐름
def coder_node(state: AgentState) -> AgentState:
    # 1. 코드 생성
    code = llm.invoke(f"Write code for: {state['current_task']}")

    # 2. 샌드박스에서 실행
    result = sandbox.execute(code)

    # 3. 에러 시 자동 수정
    if result.error:
        fixed_code = llm.invoke(f"Fix this error: {result.error}\nCode: {code}")
        result = sandbox.execute(fixed_code)

    return {"results": {state["current_step"]: result}}

from langgraph.types import interrupt

def human_feedback_node(state: AgentState) -> AgentState:
    # 워크플로우 일시 중지 — 사람의 응답을 기다림
    feedback = interrupt({
        "question": "이 계획대로 진행할까요?",
        "plan": state["plan"],
        "options": ["approve", "modify", "reject"],
    })

    if feedback["decision"] == "approve":
        return {"next": "execute"}
    elif feedback["decision"] == "modify":
        return {"next": "planner", "modifications": feedback["notes"]}
    else:
        return {"next": END}

# Task 도구 호출 예시
task_result = task_tool.invoke({
    "description": "Research the latest developments in DeerFlow",
    "agent_type": "researcher",
    "context": "Focus on v2.0 architecture changes",
})

# 에러 발생 시 Supervisor가 대안을 결정
def handle_agent_failure(state: AgentState) -> AgentState:
    error = state["last_error"]

    if error.type == "tool_error":
        # 대안 도구 사용
        return {"next": "retry_with_alternative_tool"}
    elif error.type == "timeout":
        # 태스크를 더 작은 단위로 분해
        return {"next": "planner", "instruction": "Break this into smaller steps"}
    else:
        # 사람에게 도움 요청
        return {"next": "human_feedback"}

from deerflow.harness import register_agent

@register_agent("translator")
def translator_node(state: AgentState) -> AgentState:
    """번역 에이전트 — 리서치 결과를 다국어로 번역"""
    content = state["results"].get("reporter", "")

    translated = llm.invoke(
        f"Translate the following report to Korean:\n\n{content}"
    )

    return {"results": {"translator": translated}}

# 기존 워크플로우에 번역 에이전트 추가
workflow.add_node("translator", translator_node)
workflow.add_edge("reporter", "translator")
workflow.add_edge("translator", END)