[AI] LangGraph 써보기

LangGraph란?

LangGraph는 LangChain 생태계에서 에이전트와 워크플로우를 그래프 형태로 만들 수 있게 해주는 프레임워크입니다.

LangChain만 사용해도 에이전트를 만들 수 있습니다.

하지만 여러 단계가 연결되고, 중간 상태를 저장하고, 조건에 따라 다음 실행 흐름을 바꾸고 싶어지면 단순한 함수 호출만으로는 코드가 금방 복잡해집니다.

LangGraph는 이런 흐름을 State, Node, Edge로 나누어 표현합니다.

State = 여러 단계가 공유하는 데이터
Node = 하나의 작업 단계
Edge = 다음에 실행할 단계
Graph = State, Node, Edge를 연결한 실행 흐름

 

공식 문서에서는 LangGraph의 그래프 API를 사용할 때 StateGraph로 상태 구조를 정의하고, add_node, add_edge로 노드와 실행 흐름을 연결합니다.

예를 들어 이력서 작성 시스템을 LangGraph로 표현하면 아래처럼 볼 수 있습니다.

ResumeGraphState
  ├─ profile
  ├─ work_history
  ├─ target_role
  ├─ experience_analysis
  ├─ job_match_report
  ├─ resume_draft
  ├─ portfolio_draft
  └─ review_report

 

그리고 각각의 작업은 노드가 됩니다.(에이전트가 노드는 아닙니다.)

load_inputs_node
→ analyze_experience_node
→ match_job_node
→ write_resume_node
→ write_portfolio_node
→ review_outputs_node
→ save_result_node

 

제가 이해한 LangGraph는 아래처럼 정리할 수 있습니다.

LangChain Agent = 각 작업을 수행하는 실행자
LangGraph = 여러 실행자의 순서, 상태, 분기, 반복을 관리하는 흐름 제어 도구

즉, LangGraph가 LangChain Agent를 대체하는 것은 아닙니다.

오히려 LangChain Agent를 노드 안에서 실행하면서, 전체 작업 흐름은 LangGraph가 관리하도록 만들 수 있습니다.

LangGraph를 왜 쓰지?

에이전트 파이프라인을 계속 확장하다 보면 아래와 같은 질문이 생깁니다.

• 각 단계의 중간 결과를 어디에 저장할까?
• 특정 단계가 실패하면 어떻게 다시 실행할까?
• 리뷰 결과에 따라 다시 작성 단계로 돌아가려면 어떻게 할까?
• 전체 흐름을 시각적으로 확인할 수 있을까?
• 나중에 사람 검토 단계를 넣으려면 어떻게 할까?

단순한 순차 실행은 Python 함수만으로도 충분합니다.

하지만 상태 관리, 분기, 반복, 재실행 같은 요구사항이 생기면 파이프라인 코드가 점점 복잡해집니다.

이때 사용할 수 있는 도구가 LangGraph입니다.

3편에서 만든 파이프라인을 LangGraph 관점으로 바꾸면 이렇게 됩니다.

PipelineResult = State
ExperienceAnalyzerAgent 실행 함수 = Node
다음 단계로 넘어가는 순서 = Edge
전체 ResumePipeline = Graph

LangGraph를 쓰면 좋은 점

1. 상태가 한 곳에 모인다.
2. 실행 흐름이 노드와 엣지로 표현.
3. 나중에 분기와 반복으로 확장하기 쉽다.
4. 그래프를 시각화하기 좋다.
5. 나중에 체크포인트와 Human-in-the-loop로 확장

LangGraph 써보기

만들 구조는 아래와 같습니다.

START
→ load_inputs
→ analyze_experience
→ match_job
→ write_resume
→ write_portfolio
→ review_outputs
→ render_final_markdown
→ save_result
→ END

 

각 노드는 하나의 Python 함수입니다.

각 함수는 현재 상태를 읽고, 자신이 만든 결과만 반환합니다.

노드 입력 = 현재 State
노드 출력 = State에 업데이트할 값

프로젝트 구조

langgraph-resume-workflow/
  .env
  data/
    profile.md
    work_history.md
    target_role.md
  outputs/
  main.py

설치 및 환경 설정

pip install langchain langchain-openai langgraph python-dotenv

예시 데이터

이전 파이프라인 작업시 사용했던 예시 데이터 그대로 사용합니다.

전체 코드

from __future__ import annotations

import os
from pathlib import Path
from typing import TypedDict

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph


ROOT_DIR = Path(__file__).parent
DATA_DIR = ROOT_DIR / "data"
OUTPUT_DIR = ROOT_DIR / "outputs"


class ResumeGraphState(TypedDict, total=False):
    """
    LangGraph 전체 흐름에서 공유할 상태입니다.

    total=False를 사용하면 처음부터 모든 키가 없어도 됩니다.
    각 노드가 실행되면서 필요한 값을 하나씩 채워갑니다.
    """

    request: str
    profile: str
    work_history: str
    target_role: str
    experience_analysis: str
    job_match_report: str
    resume_draft: str
    portfolio_draft: str
    review_report: str
    final_markdown: str
    output_path: str


EXPERIENCE_ANALYZER_PROMPT = """
너는 ExperienceAnalyzerAgent다.

역할:
- 사용자의 업무 경험을 분석한다.
- 프로젝트별 역할, 사용 기술, 해결한 문제, 결과를 구조화한다.
- 이력서 문장으로 바로 바꾸기보다 먼저 재료를 정리한다.

주의:
- 주어진 정보에 없는 수치나 회사명은 만들지 않는다.
- 성과가 모호하면 '근거 보완 필요'라고 표시한다.

출력 형식:

# 업무 경험 분석

## 프로젝트별 요약

## 사용 기술

## 문제 해결 경험

## 성과 후보

## 근거 보완 필요 항목
"""


JOB_MATCHER_PROMPT = """
너는 JobMatcherAgent다.

역할:
- 목표 직무 요구사항과 업무 경험 분석 결과를 비교한다.
- 어떤 경험이 어떤 요구사항과 연결되는지 정리한다.
- 부족한 요구사항도 함께 표시한다.

출력 형식:

# 직무 매칭 분석

## 강하게 연결되는 경험

## 부분적으로 연결되는 경험

## 부족한 경험 또는 추가 확인 필요

## 이력서에서 강조할 키워드
"""


RESUME_WRITER_PROMPT = """
너는 ResumeWriterAgent다.

역할:
- 업무 경험 분석과 직무 매칭 결과를 바탕으로 이력서 초안을 작성한다.
- 문장은 간결하고 성과 중심으로 작성한다.
- 과장하지 않고 주어진 근거 안에서만 작성한다.

출력 형식:

# 이력서 초안

## 한 줄 소개

## 핵심 역량

## 경력 요약

## 프로젝트 경험

## 추가로 확인할 내용
"""


PORTFOLIO_WRITER_PROMPT = """
너는 PortfolioWriterAgent다.

역할:
- 업무 경험 분석과 직무 매칭 결과를 바탕으로 포트폴리오 초안을 작성한다.
- 프로젝트별로 문제, 역할, 해결 방법, 사용 기술, 결과를 정리한다.
- 실제 포트폴리오 페이지에 옮길 수 있는 구조로 작성한다.

출력 형식:

# 포트폴리오 초안

## 프로젝트 1

### 문제 상황

### 맡은 역할

### 해결 방법

### 사용 기술

### 결과

### 더 보완할 자료
"""


REVIEW_AGENT_PROMPT = """
너는 ReviewAgent다.

역할:
- 이력서 초안과 포트폴리오 초안을 검토한다.
- 과장된 표현, 근거 부족, 빠진 질문을 찾아낸다.
- 최종 제출 전 보완 방향을 정리한다.

출력 형식:

# 리뷰 결과

## 좋은 점

## 과장 가능성이 있는 표현

## 근거 보완이 필요한 부분

## 추가 질문

## 수정 제안
"""


def read_markdown(filename: str) -> str:
    """
    data 폴더 안의 markdown 파일을 읽습니다.
    """
    path = DATA_DIR / filename

    if not path.exists():
        raise FileNotFoundError(f"파일을 찾을 수 없습니다: {path}")

    return path.read_text(encoding="utf-8")


def save_markdown(filename: str, content: str) -> Path:
    """
    outputs 폴더에 markdown 파일을 저장합니다.
    """
    OUTPUT_DIR.mkdir(exist_ok=True)

    safe_filename = filename.replace("/", "_").replace("\\", "_")
    if not safe_filename.endswith(".md"):
        safe_filename = f"{safe_filename}.md"

    path = OUTPUT_DIR / safe_filename
    path.write_text(content, encoding="utf-8")

    return path


def last_message_text(result: dict) -> str:
    """
    LangChain agent 실행 결과에서 마지막 메시지 내용을 문자열로 꺼냅니다.
    """
    message = result["messages"][-1]
    content = getattr(message, "content", "")

    if isinstance(content, str):
        return content

    return str(content)


def create_model() -> ChatOpenAI:
    """
    모든 노드에서 사용할 언어 모델을 생성합니다.
    """
    model_name = os.getenv("OPENAI_MODEL", "gpt-5.5")

    return ChatOpenAI(
        model=model_name,
        temperature=0.2,
    )


def create_pipeline_agent(model: ChatOpenAI, name: str, system_prompt: str):
    """
    LangGraph 노드 내부에서 호출할 LangChain agent를 생성합니다.
    """
    return create_agent(
        model=model,
        tools=[],
        system_prompt=system_prompt,
        name=name,
    )


def invoke_agent(agent, content: str) -> str:
    """
    LangChain agent를 실행하고 마지막 응답 텍스트만 반환합니다.
    """
    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": content,
                }
            ]
        }
    )

    return last_message_text(result)


def render_final_markdown(state: ResumeGraphState) -> str:
    """
    그래프의 최종 상태를 하나의 markdown 문서로 합칩니다.
    """
    return f"""
# LangGraph 이력서 작성 결과

## 1. 업무 경험 분석

{state["experience_analysis"]}

## 2. 직무 매칭 분석

{state["job_match_report"]}

## 3. 이력서 초안

{state["resume_draft"]}

## 4. 포트폴리오 초안

{state["portfolio_draft"]}

## 5. 리뷰 결과

{state["review_report"]}
""".strip()


def create_resume_graph():
    """
    이력서 작성 LangGraph workflow를 생성합니다.
    """
    model = create_model()

    experience_agent = create_pipeline_agent(
        model=model,
        name="experience_analyzer_agent",
        system_prompt=EXPERIENCE_ANALYZER_PROMPT,
    )

    job_matcher_agent = create_pipeline_agent(
        model=model,
        name="job_matcher_agent",
        system_prompt=JOB_MATCHER_PROMPT,
    )

    resume_writer_agent = create_pipeline_agent(
        model=model,
        name="resume_writer_agent",
        system_prompt=RESUME_WRITER_PROMPT,
    )

    portfolio_writer_agent = create_pipeline_agent(
        model=model,
        name="portfolio_writer_agent",
        system_prompt=PORTFOLIO_WRITER_PROMPT,
    )

    review_agent = create_pipeline_agent(
        model=model,
        name="review_agent",
        system_prompt=REVIEW_AGENT_PROMPT,
    )

    def load_inputs_node(state: ResumeGraphState) -> ResumeGraphState:
        """
        입력 markdown 파일을 읽어서 state에 추가합니다.
        """
        return {
            "profile": read_markdown("profile.md"),
            "work_history": read_markdown("work_history.md"),
            "target_role": read_markdown("target_role.md"),
        }

    def analyze_experience_node(state: ResumeGraphState) -> ResumeGraphState:
        """
        업무 경험을 분석합니다.
        """
        experience_analysis = invoke_agent(
            experience_agent,
            f"""
[업무 경험]
{state["work_history"]}
""".strip(),
        )

        return {"experience_analysis": experience_analysis}

    def match_job_node(state: ResumeGraphState) -> ResumeGraphState:
        """
        업무 경험 분석 결과와 목표 직무를 매칭합니다.
        """
        job_match_report = invoke_agent(
            job_matcher_agent,
            f"""
[업무 경험 분석]
{state["experience_analysis"]}

[목표 직무]
{state["target_role"]}
""".strip(),
        )

        return {"job_match_report": job_match_report}

    def write_resume_node(state: ResumeGraphState) -> ResumeGraphState:
        """
        이력서 초안을 작성합니다.
        """
        resume_draft = invoke_agent(
            resume_writer_agent,
            f"""
[기본 프로필]
{state["profile"]}

[업무 경험 분석]
{state["experience_analysis"]}

[직무 매칭 분석]
{state["job_match_report"]}
""".strip(),
        )

        return {"resume_draft": resume_draft}

    def write_portfolio_node(state: ResumeGraphState) -> ResumeGraphState:
        """
        포트폴리오 초안을 작성합니다.
        """
        portfolio_draft = invoke_agent(
            portfolio_writer_agent,
            f"""
[업무 경험 분석]
{state["experience_analysis"]}

[직무 매칭 분석]
{state["job_match_report"]}
""".strip(),
        )

        return {"portfolio_draft": portfolio_draft}

    def review_outputs_node(state: ResumeGraphState) -> ResumeGraphState:
        """
        이력서와 포트폴리오 초안을 검토합니다.
        """
        review_report = invoke_agent(
            review_agent,
            f"""
[이력서 초안]
{state["resume_draft"]}

[포트폴리오 초안]
{state["portfolio_draft"]}

[업무 경험 분석]
{state["experience_analysis"]}

[직무 매칭 분석]
{state["job_match_report"]}
""".strip(),
        )

        return {"review_report": review_report}

    def render_final_markdown_node(state: ResumeGraphState) -> ResumeGraphState:
        """
        모든 결과를 하나의 markdown 문서로 합칩니다.
        """
        return {"final_markdown": render_final_markdown(state)}

    def save_result_node(state: ResumeGraphState) -> ResumeGraphState:
        """
        최종 markdown 결과를 파일로 저장합니다.
        """
        output_path = save_markdown(
            filename="langgraph_resume_result.md",
            content=state["final_markdown"],
        )

        return {"output_path": str(output_path)}

    workflow = StateGraph(ResumeGraphState)

    workflow.add_node("load_inputs", load_inputs_node)
    workflow.add_node("analyze_experience", analyze_experience_node)
    workflow.add_node("match_job", match_job_node)
    workflow.add_node("write_resume", write_resume_node)
    workflow.add_node("write_portfolio", write_portfolio_node)
    workflow.add_node("review_outputs", review_outputs_node)
    workflow.add_node("render_final_markdown", render_final_markdown_node)
    workflow.add_node("save_result", save_result_node)

    workflow.add_edge(START, "load_inputs")
    workflow.add_edge("load_inputs", "analyze_experience")
    workflow.add_edge("analyze_experience", "match_job")
    workflow.add_edge("match_job", "write_resume")
    workflow.add_edge("write_resume", "write_portfolio")
    workflow.add_edge("write_portfolio", "review_outputs")
    workflow.add_edge("review_outputs", "render_final_markdown")
    workflow.add_edge("render_final_markdown", "save_result")
    workflow.add_edge("save_result", END)

    return workflow.compile()


def main() -> None:
    load_dotenv()

    app = create_resume_graph()

    result = app.invoke(
        {
            "request": "목표 직무에 맞는 이력서와 포트폴리오 초안을 작성해줘.",
        }
    )

    print(f"LangGraph 결과 저장 완료: {result['output_path']}")

    print("\n그래프 구조:")
    print(app.get_graph().draw_mermaid())


if __name__ == "__main__":
    main()

코드 설명

1. ResumeGraphState

class ResumeGraphState(TypedDict, total=False):

ResumeGraphState는 그래프 전체가 공유하는 상태입니다.

3편에서는 PipelineResult라는 dataclass에 중간 결과를 모았습니다.

이번에는 LangGraph의 State 역할을 하는 TypedDict를 사용합니다.

상태에는 각 단계의 입력과 출력이 모두 들어갑니다.

profile: str
work_history: str
target_role: str
experience_analysis: str
job_match_report: str
resume_draft: str
portfolio_draft: str
review_report: str
final_markdown: str
output_path: str

total=False를 사용한 이유는 처음부터 모든 값이 존재하지 않기 때문입니다.

처음 실행할 때는 request만 있고, 노드가 실행되면서 값이 하나씩 추가됩니다.

2. Node

LangGraph에서 Node는 하나의 작업 단계입니다.

이번 코드에서는 아래 함수들이 노드입니다.

def load_inputs_node(state: ResumeGraphState) -> ResumeGraphState:
def analyze_experience_node(state: ResumeGraphState) -> ResumeGraphState:
def match_job_node(state: ResumeGraphState) -> ResumeGraphState:
def write_resume_node(state: ResumeGraphState) -> ResumeGraphState:
def write_portfolio_node(state: ResumeGraphState) -> ResumeGraphState:
def review_outputs_node(state: ResumeGraphState) -> ResumeGraphState:
def render_final_markdown_node(state: ResumeGraphState) -> ResumeGraphState:
def save_result_node(state: ResumeGraphState) -> ResumeGraphState:

각 노드는 현재 상태를 입력으로 받습니다.

그리고 전체 상태를 직접 수정하지 않고, 업데이트할 값만 반환합니다.

예를 들어 업무 경험 분석 노드는 아래처럼 동작합니다.

return {"experience_analysis": experience_analysis}

이 반환값이 기존 state에 합쳐지면서 다음 노드로 전달됩니다.

3. Edge

Edge는 노드와 노드를 연결하는 실행 흐름입니다.

workflow.add_edge(START, "load_inputs")
workflow.add_edge("load_inputs", "analyze_experience")
workflow.add_edge("analyze_experience", "match_job")
workflow.add_edge("match_job", "write_resume")

4. compile

START, END, 노드와 엣지들로 구성한 그래프를 컴파일을 통해서 실행할 수 있게 만듭니다.

return workflow.compile()

컴파일된 그래프는 아래처럼 실행할 수 있습니다.

result = app.invoke(
    {
        "request": "목표 직무에 맞는 이력서와 포트폴리오 초안을 작성해줘.",
    }
)

LangGraph는 실행 결과로 최종 state를 반환합니다.

그래서 저장 경로도 아래처럼 꺼낼 수 있습니다.

print(result["output_path"])

5. LangChain Agent와 LangGraph Node의 관계

이번 예제에서는 LangGraph 노드 안에서 LangChain agent를 호출했습니다.

experience_analysis = invoke_agent(
    experience_agent,
    f"""
        [업무 경험]
        {state["work_history"]}
    """.strip(),
)

구조를 그림처럼 보면 아래와 같습니다.

LangGraph Node
  └─ LangChain Agent 실행
       └─ LLM 호출

즉, LangGraph가 LLM 자체를 대체하는 것은 아닙니다.

LangGraph는 여러 작업 단계와 상태 흐름을 관리하고, 각 노드 안에서는 LangChain agent나 일반 Python 함수를 실행할 수 있습니다.

정리

이번 편에서는 단순한 일자형 그래프를 만들었습니다.

일자형 그래프는 파이프라인과 다를바가 거의 없죠.

다음 글에서는 이 일자형 그래프에 분기와 반복을 추가해서 리뷰 결과에 따라 다시 작성 단계로 돌아가는 Agentic loop를 만들어볼 예정입니다.

참고

1. 노드는 state를 직접 수정하지 않는 것이 좋습니다.
   • 노드 함수는 현재 state를 읽고, 업데이트할 값만 반환하는 방식으로 작성하는 것이 좋습니다.
2. state 이름을 명확하게 짓는 것이 좋습니다.
   
   • 변수명을 명확히 짓듯이 지어야 합니다.
3. LangGraph는 파이프라인을 대체한다기보다 확장합니다.