LangGraph로 콘텐츠 자동화 파이프라인 만들기 — 리서치부터 초안 발행까지 실전 코드 공개

매일 블로그 글 쓰는 게 힘드셨죠? 저도 그랬어요. 그래서 AI 에이전트한테 시켰습니다.

---

1일 1포스팅. 말은 쉬운데, 실제로 유지하다 보면 한계가 와요.

주제 잡고 → 자료 찾고 → 구조 잡고 → 글 쓰고 → 교정하고 → 발행하는 전 과정이 직장인 기준으로 하루에 2~3시간씩 잡아먹거든요. 개발 일도 하면서 블로그까지 유지하는 건 체력 싸움이에요.

그래서 LangGraph로 콘텐츠 자동화 파이프라인을 만들었어요. 결론부터 말하면, 지금은 제가 하는 일이 딱 두 가지예요. "주제 입력" 과 "최종 편집 후 발행 버튼 누르기". 그 사이의 리서치, 구조 잡기, 초안 작성, 사실 확인은 에이전트가 알아서 해요.

오늘은 그 파이프라인을 실제 코드와 함께 공개할게요.

---

LangGraph가 뭔데요? — 핵심만 30초 요약

이전 포스팅에서 AI 에이전트를 소개했는데, LangGraph는 그 에이전트들을 여러 개 연결해서 하나의 워크플로우로 만드는 프레임워크예요.

기존 LangChain이 "AI한테 시키는 것"이라면, LangGraph는 "AI 팀을 조직해서 프로젝트를 맡기는 것"에 더 가까워요.

[기존 방식]
나 → ChatGPT → 결과

[LangGraph 방식]
나 → [리서치 에이전트] → [작성 에이전트] → [검토 에이전트] → 결과

2026년 초 기준으로 LangGraph는 GitHub 스타 수에서 CrewAI를 추월했고, 기업 실무 도입이 가장 빠르게 늘고 있는 에이전트 프레임워크예요.

---

설치 및 환경 세팅

pip install langgraph langchain-anthropic langchain-community tavily-python

.env 파일에 API 키를 세팅해요:

ANTHROPIC_API_KEY=your_key_here
TAVILY_API_KEY=your_key_here   # 웹 검색 도구

⚠️ 반드시 챙겨야 할 설정 1가지: Tavily API는 무료 플랜이 월 1,000회 검색까지 가능해요. 자동화 파이프라인에서 호출 횟수가 생각보다 빨리 차니까, 처음엔 검색 쿼리 수를 최소화하는 로직을 넣어두세요.

---

실제로 어떻게 만들었냐면 — 전체 파이프라인 구조

제가 사용하는 콘텐츠 파이프라인은 4개 에이전트로 구성돼요.

주제 입력
    ↓
[1] 리서치 에이전트 — 웹 검색으로 최신 정보 수집
    ↓
[2] 구조 에이전트 — 블로그 포맷에 맞는 목차 생성
    ↓
[3] 작성 에이전트 — 섹션별 본문 작성
    ↓
[4] 검토 에이전트 — 사실 오류·어색한 문장 체크
    ↓
초안 완성 (파일 저장)

전체 코드

import os
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults

# ── 모델 및 도구 초기화 ─────────────────────────────────────
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0.7)
search_tool = TavilySearchResults(max_results=5)

# ── 상태 정의 (에이전트 간 공유되는 데이터) ─────────────────
class ContentState(TypedDict):
    topic: str            # 입력 주제
    research: str         # 리서치 결과
    outline: str          # 목차
    draft: str            # 초안
    reviewed_draft: str   # 검토 완료 초안
    messages: List[str]   # 로그


# ── 에이전트 1: 리서치 ────────────────────────────────────────
def research_agent(state: ContentState) -> ContentState:
    print("🔍 리서치 중...")
    topic = state["topic"]

    # 웹 검색 실행
    results = search_tool.invoke(topic)
    research_text = "\n\n".join([r["content"] for r in results[:3]])

    state["research"] = research_text
    state["messages"].append("✅ 리서치 완료")
    return state


# ── 에이전트 2: 구조 잡기 ─────────────────────────────────────
def outline_agent(state: ContentState) -> ContentState:
    print("📋 목차 구성 중...")

    prompt = f"""
다음 주제와 리서치 자료를 바탕으로 블로그 목차를 작성해줘.

주제: {state['topic']}
리서치 자료: {state['research'][:2000]}

요구사항:
- 개발자 독자 대상
- H2 헤더 4~6개
- 각 섹션 한 줄 설명 포함
- 코드 예시 섹션 반드시 포함
"""
    response = llm.invoke(prompt)
    state["outline"] = response.content
    state["messages"].append("✅ 목차 완성")
    return state


# ── 에이전트 3: 본문 작성 ─────────────────────────────────────
def write_agent(state: ContentState) -> ContentState:
    print("✍️ 본문 작성 중...")

    prompt = f"""
다음 목차와 리서치 자료를 바탕으로 블로그 본문을 작성해줘.

주제: {state['topic']}
목차: {state['outline']}
리서치: {state['research'][:3000]}

작성 규칙:
- 분량: 3,000자 이상
- 어미: ~요, ~해요 (친근한 경어체)
- 1인칭 개발자 시점
- 코드는 반드시 코드블록으로
- 솔직한 단점도 포함
"""
    response = llm.invoke(prompt)
    state["draft"] = response.content
    state["messages"].append("✅ 본문 작성 완료")
    return state


# ── 에이전트 4: 검토 ──────────────────────────────────────────
def review_agent(state: ContentState) -> ContentState:
    print("🔎 검토 중...")

    prompt = f"""
아래 블로그 초안을 검토하고 수정된 버전을 반환해줘.

체크 항목:
1. 사실 오류 또는 과장된 표현
2. 어색한 문장이나 문법 오류
3. 코드 블록 형식 오류
4. 3,000자 미만이면 내용 보강

초안:
{state['draft']}
"""
    response = llm.invoke(prompt)
    state["reviewed_draft"] = response.content
    state["messages"].append("✅ 검토 완료")
    return state


# ── 파일 저장 ─────────────────────────────────────────────────
def save_to_file(state: ContentState) -> ContentState:
    filename = f"tistory_{state['topic'][:10].replace(' ', '_')}.md"
    with open(filename, "w", encoding="utf-8") as f:
        f.write(state["reviewed_draft"])
    print(f"💾 저장 완료: {filename}")
    state["messages"].append(f"✅ 파일 저장: {filename}")
    return state


# ── 그래프 구성 ───────────────────────────────────────────────
def build_pipeline():
    graph = StateGraph(ContentState)

    graph.add_node("research",  research_agent)
    graph.add_node("outline",   outline_agent)
    graph.add_node("write",     write_agent)
    graph.add_node("review",    review_agent)
    graph.add_node("save",      save_to_file)

    graph.set_entry_point("research")
    graph.add_edge("research", "outline")
    graph.add_edge("outline",  "write")
    graph.add_edge("write",    "review")
    graph.add_edge("review",   "save")
    graph.add_edge("save",     END)

    return graph.compile()


# ── 실행 ──────────────────────────────────────────────────────
if __name__ == "__main__":
    pipeline = build_pipeline()

    result = pipeline.invoke({
        "topic": "2026년 AI 에이전트 프레임워크 비교",
        "research": "",
        "outline": "",
        "draft": "",
        "reviewed_draft": "",
        "messages": []
    })

    print("\n📝 파이프라인 로그:")
    for msg in result["messages"]:
        print(f"  {msg}")

---

조건부 분기 추가하기 — 검토 실패 시 재작성

기본 파이프라인에 "글이 짧으면 재작성 루프"를 추가하면 품질이 훨씬 안정돼요.

# 검토 후 품질 판단 함수
def should_rewrite(state: ContentState) -> str:
    draft = state.get("reviewed_draft", "")
    # 2,500자 미만이면 재작성 요청
    if len(draft) < 2500:
        print("⚠️ 분량 부족 — 재작성 요청")
        return "write"   # write 노드로 되돌아감
    return "save"        # 통과 → 저장

# 그래프에 조건부 엣지 추가
graph.add_conditional_edges(
    "review",
    should_rewrite,
    {
        "write": "write",
        "save":  "save"
    }
)

이 로직 하나로 "짧은 글이 그냥 발행되는" 문제가 거의 사라졌어요.

---

솔직한 한계 — 이것만 주의하세요

실행 비용이 쌓여요. 4개 에이전트가 각각 Claude API를 호출하니까, 포스팅 하나에 Claude Sonnet 기준 약 $0.05~0.15 정도 나와요. 하루 1개면 한 달에 $3~5 수준이라 감당할 만해요. 그런데 검토 루프가 여러 번 돌면 그 이상 나올 수 있어요.

100% 자동 발행은 비추예요. 에이전트가 가끔 사실을 잘못 인용하거나, 구어체가 갑자기 문어체로 바뀌는 경우가 있어요. 최종 발행 전 사람이 한 번 읽는 human-in-the-loop 단계는 꼭 유지하세요.

Tavily 무료 플랜 한도 주의. 월 1,000회 검색이 파이프라인 자동화하면 생각보다 빨리 차요. 검색 결과를 캐싱하거나, 같은 주제 재검색을 막는 로직을 추가하는 게 좋아요.

LangSmith 없으면 디버깅이 지옥이에요. 파이프라인이 어디서 막혔는지 추적하려면 LangSmith 연동을 강력 추천해요. 무료 플랜으로도 충분히 써요.

# LangSmith 연동 (환경변수 추가)
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your_langsmith_key"
os.environ["LANGCHAIN_PROJECT"] = "blog-pipeline"

---

마무리: 자동화의 목표는 "더 잘 쓰기 위한 시간 확보"예요

이 파이프라인 만들고 나서 달라진 게 있어요. 글 쓰는 시간이 줄어든 게 아니라, 더 좋은 글에 시간을 쓸 수 있게 됐어요.

에이전트가 초안을 잡아주면, 저는 그 위에서 개인 경험과 인사이트를 얹는 작업만 해요. 결과적으로 글 품질은 올라가고, 소요 시간은 줄었어요.

코드가 낯설게 느껴지면 일단 pip install langgraph부터 해보세요. 설치하고, 코드 붙여넣고, 실행해보는 것 — 그게 에이전트를 이해하는 가장 빠른 방법이에요.

다음 포스팅 예고: LangSmith로 에이전트 파이프라인 디버깅하기 — 어디서 왜 실패했는지 한눈에 보는 법

 

 

---

Sources:

• LangGraph Official Docs — LangChain
• LangGraph in 2026: Build Multi-Agent AI Systems — DEV Community
• Press: Multi-Agent LangGraph Content Pipeline — Vadim's Blog
• LangGraph Agents in Production 2026 — Apify
• LangGraph + MCP Multi-Agent Workflows 2026 Guide