LangGraph로 블로그 글 자동화 파이프라인 구축하기: 리서치부터 발행까지

from typing import TypedDict, Annotated, List
import operator
from langchain_core.messages import BaseMessage

# 1. 상태 정의
class BlogState(TypedDict):
    topic: str
    research_data: str
    draft_content: str
    feedback: str
    needs_revision: bool
    final_content: str
    messages: Annotated[List[BaseMessage], operator.add]

# 2. 노드 정의
def research_node(state: BlogState):
    """주어진 주제에 대한 정보를 리서치합니다."""
    topic = state.get("topic", "")
    print(f"리서치 중: {topic}")
    # 실제 구현에서는 웹 검색 도구 등을 사용하여 데이터를 수집합니다.
    researched_info = (
        f"LangGraph는 LLM 기반 다중 에이전트 구축에 사용됩니다. "
        f"상태, 노드, 엣지가 핵심입니다."
    )
    return {
        "research_data": researched_info,
        "messages": [("user", f"주제 '{topic}'에 대한 리서치 완료.")]
    }

def write_node(state: BlogState):
    """리서치 데이터를 바탕으로 블로그 글 초안을 작성합니다."""
    research_data = state.get("research_data", "")
    topic = state.get("topic", "")
    print(f"초안 작성 중: {topic}")
    # 실제 구현에서는 LLM을 사용하여 초안을 생성합니다.
    draft = f"## {topic} 서론\n{research_data}\nLangGraph는..."
    return {
        "draft_content": draft,
        "messages": [("user", "블로그 초안 작성 완료.")]
    }

def critique_node(state: BlogState):
    """작성된 초안을 비평하고 수정 필요 여부를 결정합니다."""
    draft = state.get("draft_content", "")
    print("초안 비평 중...")
    # 실제 구현에서는 LLM을 사용하여 비평을 생성합니다.
    if len(draft) < 500:
        feedback = "글의 내용이 너무 짧습니다. 더 상세한 내용을 추가해야 합니다."
        needs_revision = True
    else:
        feedback = "내용이 적절합니다."
        needs_revision = False
    return {
        "feedback": feedback,
        "needs_revision": needs_revision,
        "messages": [("user", f"비평 완료: {feedback}")]
    }

def revise_node(state: BlogState):
    """비평을 바탕으로 글을 수정합니다."""
    draft = state.get("draft_content", "")
    feedback = state.get("feedback", "")
    print(f"글 수정 중. 피드백: {feedback}")
    # 실제 구현에서는 LLM을 사용하여 피드백을 반영합니다.
    revised_draft = draft + f"\n\n--- 수정 사항 반영: {feedback} ---"
    return {
        "draft_content": revised_draft,
        "needs_revision": False,
        "messages": [("user", "글 수정 완료.")]
    }

def human_review_node(state: BlogState):
    """인간이 최종적으로 글을 검토하고 승인하거나 수정을 요청합니다."""
    print("\n--- 인간 검토 대기 중 ---")
    print("현재 초안:\n", state["draft_content"])
    review_status = input("글을 승인하시겠습니까? (yes/no): ").lower()
    if review_status == "yes":
        return {
            "final_content": state["draft_content"],
            "needs_revision": False,
            "messages": [("user", "인간 검토: 승인")]
        }
    else:
        additional_feedback = input("어떤 수정이 필요합니까?: ")
        return {
            "feedback": additional_feedback,
            "needs_revision": True,
            "messages": [("user", f"인간 검토: 수정 요청 ({additional_feedback})")]
        }

# 3. 조건부 엣지 결정 함수
def should_continue_writing(state: BlogState):
    if state["needs_revision"]:
        return "revise"
    else:
        return "human_review"

# 4. 그래프 구축
from langgraph.graph import StateGraph, END

workflow = StateGraph(BlogState)

workflow.add_node("research", research_node)
workflow.add_node("write", write_node)
workflow.add_node("critique", critique_node)
workflow.add_node("revise", revise_node)
workflow.add_node("human_review", human_review_node)

workflow.set_entry_point("research")

workflow.add_edge("research", "write")
workflow.add_edge("write", "critique")

workflow.add_conditional_edges(
    "critique",
    should_continue_writing,
    {"revise": "revise", "human_review": "human_review"}
)

# 수정 후 다시 비평 노드로 돌아가는 루프
workflow.add_edge("revise", "critique")

workflow.add_conditional_edges(
    "human_review",
    should_continue_writing,
    {"revise": "revise", "human_review": END}
)

app = workflow.compile()

# 파이프라인 실행
initial_state = {
    "topic": "LangGraph 블로그 글 자동화 파이프라인",
    "messages": []
}
for s in app.stream(initial_state):
    print(s)
    print("---")