[AI] 멀티 에이전트 파이프라인 with LangChain

왜 멀티 에이전트를 파이프라인으로 만들까?

이전 글에서는 SupervisorAgent가 여러 하위 에이전트를 도구처럼 호출하는 구조를 만들어봤습니다.

SupervisorAgent
  ├─ ExperienceAnalyzerAgent
  ├─ JobMatcherAgent
  ├─ ResumeWriterAgent
  └─ ResumeReviewAgent

 

이 구조는 중앙 에이전트가 전체 작업을 조율한다는 점에서 좋습니다.

하지만 이력서 작성처럼 작업 순서가 비교적 명확한 경우에는 이런 생각이 들 수 있습니다.

업무 경험을 먼저 분석하고, 그 다음 목표 직무와 매칭하고, 그 다음 이력서와 포트폴리오를 작성하고, 마지막으로 리뷰하면 되지 않을까?

 

즉, 모든 흐름을 에이전트에게 맡기기보다 개발자가 실행 순서를 명확히 정해두는 편이 더 이해하기 쉬울 때가 있습니다.

이럴 때 사용할 수 있는 구조가 멀티 에이전트 파이프라인입니다.

파이프라인 방식에서는 여러 에이전트를 만들고, 각 에이전트를 정해진 순서대로 실행합니다.

업무 경험 분석
→ 직무 매칭
→ 이력서 작성
→ 포트폴리오 작성
→ 리뷰
→ 최종 저장

파이프라인으로 만들면 좋은 점

1. 실행 순서를 예측하기 쉽다.
2. 중간 결과를 확인하기 쉽다.
3. 역할별로 개선하기 쉽다.
4. 같은 작업을 반복 실행하기 좋다.
5. LangGraph로 확장하기 좋다.

마지막 5번은 파이프라인은 나중에 LangGraph로 넘어갈 때 자연스럽게 바꿀 수 있습니다.

Agent = Node
실행 순서 = Edge
중간 결과 = State

파이프라인이 항상 정답은 아니다

그렇다고 모든 멀티 에이전트를 파이프라인으로 만들어야 하는 것은 아닙니다.

파이프라인은 실행 순서가 정해져 있을 때 강합니다.

반대로 사용자의 요청마다 흐름이 자주 바뀌거나, 에이전트가 상황에 따라 도구를 자유롭게 선택해야 한다면 Supervisor 방식이 더 잘 맞을 수 있습니다.

 

상황	어울리는 방식
실행 순서가 명확하다	Pipeline
매번 같은 업무 절차를 반복한다	Pipeline
중간 결과를 저장하고 검토해야 한다	Pipeline
사용자의 요청마다 처리 순서가 달라진다	Supervisor
에이전트가 도구를 자유롭게 선택해야 한다	Supervisor
분기, 반복, 되돌아가기 흐름이 많다	LangGraph / Agent Loop

 

하지만 저희가 만들고 있던 이력서 작성은 보통 아래 순서가 자연스럽습니다.

경험 분석 → 직무 매칭 → 작성 → 리뷰

그래서 파이프라인으로 만들기 좋습니다.

하지만 사용자가 이렇게 요청한다고 생각해보겠습니다.

내 이력서도 봐주고,
채용공고도 찾아보고,
부족한 기술은 학습 계획으로 만들고,
필요하면 포트폴리오 프로젝트 아이디어도 추천해줘.

이런 요청은 순서가 고정되어 있다기보다 상황에 따라 필요한 도구와 에이전트가 달라집니다.

이 경우에는 Supervisor 방식이나 LangGraph 기반 흐름이 더 적합할 수 있습니다.

파이프라인 만들기

이번 글에서는 아래처럼 순서가 명확한 이력서 작성 파이프라인을 만들어보겠습니다.

load input files
→ ExperienceAnalyzerAgent
→ JobMatcherAgent
→ ResumeWriterAgent
→ PortfolioWriterAgent
→ ReviewAgent
→ save final report

 

각 에이전트는 하나의 역할만 담당하고, 앞 단계의 결과가 다음 단계의 입력으로 들어갑니다.

ExperienceAnalyzerAgent
  입력: 업무 경험
  출력: 업무 경험 분석 결과

JobMatcherAgent
  입력: 업무 경험 분석 결과 + 목표 직무
  출력: 직무 매칭 결과

ResumeWriterAgent
  입력: 프로필 + 업무 분석 + 직무 매칭
  출력: 이력서 초안

PortfolioWriterAgent
  입력: 업무 분석 + 직무 매칭
  출력: 포트폴리오 초안

ReviewAgent
  입력: 이력서 초안 + 포트폴리오 초안
  출력: 리뷰 결과와 보완 질문

프로젝트 구조

multi-agent-pipeline/
  .env
  data/
    profile.md
    work_history.md
    target_role.md
  outputs/
  main.py

필요한 패키지

pip install langchain langchain-openai python-dotenv

 

OpenAI 연동은 되어 있다고 생각하고 진행합니다.

예시 데이터

data/profile.md

# 기본 프로필

- 이름: 홍길동
- 목표: AI Agent Engineer로 성장하기
- 관심 분야: LLM 애플리케이션, LangChain, LangGraph, RAG, 업무 자동화
- 현재 강점: Python 백엔드 개발, API 설계, 데이터 기반 기능 구현

 

data/work_history.md

# 업무 경험

## AI 교육 플랫폼 프로젝트

- 역할: 백엔드 개발자
- 기간: 2024.03 ~ 2024.12
- 사용 기술: Python, FastAPI, PostgreSQL, LangChain
- 주요 업무:
  - 수강생 질문 데이터를 기반으로 AI 튜터 응답 기능 개발
  - 강의별 Q&A 검색 API 구현
  - 관리자용 학습 리포트 API 개발
- 성과:
  - 반복 질문 대응 시간을 줄이기 위한 자동 응답 흐름 설계
  - 프롬프트 템플릿 관리 구조 개선

## 사내 운영 자동화 프로젝트

- 역할: 백엔드 개발자
- 기간: 2025.01 ~ 2025.04
- 사용 기술: Python, FastAPI, PostgreSQL, Docker
- 주요 업무:
  - 반복 운영 업무를 처리하는 내부 API 개발
  - 관리자 입력 데이터를 검증하는 자동화 로직 구현
  - 작업 실패 로그를 조회하는 관리 기능 개발
- 성과:
  - 수동으로 확인하던 운영 데이터를 API 기반으로 조회할 수 있게 개선

 

data/target_role.md

# 목표 직무

AI Agent Engineer

## 요구사항

- Python 기반 LLM 애플리케이션 개발 경험
- LangChain 또는 LangGraph 사용 경험
- Tool Calling, Agent Workflow 이해
- RAG 또는 검색 기반 응답 시스템 경험
- 백엔드 API 개발 경험
- 문제 해결 과정을 문서화할 수 있는 능력

전체 코드

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI


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


EXPERIENCE_ANALYZER_PROMPT = """
너는 ExperienceAnalyzerAgent다.

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

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

출력 형식:

# 업무 경험 분석

## 프로젝트별 요약

## 사용 기술

## 문제 해결 경험

## 성과 후보

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


JOB_MATCHER_PROMPT = """
너는 JobMatcherAgent다.

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

출력 형식:

# 직무 매칭 분석

## 강하게 연결되는 경험

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

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

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


RESUME_WRITER_PROMPT = """
너는 ResumeWriterAgent다.

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

출력 형식:

# 이력서 초안

## 한 줄 소개

## 핵심 역량

## 경력 요약

## 프로젝트 경험

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


PORTFOLIO_WRITER_PROMPT = """
너는 PortfolioWriterAgent다.

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

출력 형식:

# 포트폴리오 초안

## 프로젝트 1

### 문제 상황

### 맡은 역할

### 해결 방법

### 사용 기술

### 결과

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


REVIEW_AGENT_PROMPT = """
너는 ReviewAgent다.

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

출력 형식:

# 리뷰 결과

## 좋은 점

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

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

## 추가 질문

## 수정 제안
"""


@dataclass
class PipelineResult:
    """
    파이프라인 실행 결과를 한 곳에 모아두기 위한 데이터 클래스입니다.
    """

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


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):
    """
    파이프라인의 한 단계를 담당할 LangChain agent를 생성합니다.
    """
    return create_agent(
        model=model,
        tools=[],
        system_prompt=system_prompt,
        name=name,
    )


class ResumePipeline:
    """
    여러 에이전트를 정해진 순서대로 실행하는 파이프라인입니다.
    """

    def __init__(self, model: ChatOpenAI):
        self.experience_agent = create_pipeline_agent(
            model=model,
            name="experience_analyzer_agent",
            system_prompt=EXPERIENCE_ANALYZER_PROMPT,
        )

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

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

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

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

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

        return last_message_text(result)

    def run(self) -> PipelineResult:
        """
        파이프라인을 순서대로 실행합니다.
        """
        profile = read_markdown("profile.md")
        work_history = read_markdown("work_history.md")
        target_role = read_markdown("target_role.md")

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

        job_match_report = self.invoke_agent(
            self.job_matcher_agent,
            f"""
[업무 경험 분석]
{experience_analysis}

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

        resume_draft = self.invoke_agent(
            self.resume_writer_agent,
            f"""
[기본 프로필]
{profile}

[업무 경험 분석]
{experience_analysis}

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

        portfolio_draft = self.invoke_agent(
            self.portfolio_writer_agent,
            f"""
[업무 경험 분석]
{experience_analysis}

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

        review_report = self.invoke_agent(
            self.review_agent,
            f"""
[이력서 초안]
{resume_draft}

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

[업무 경험 분석]
{experience_analysis}

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

        return PipelineResult(
            profile=profile,
            work_history=work_history,
            target_role=target_role,
            experience_analysis=experience_analysis,
            job_match_report=job_match_report,
            resume_draft=resume_draft,
            portfolio_draft=portfolio_draft,
            review_report=review_report,
        )

    def render_markdown(self, result: PipelineResult) -> str:
        """
        파이프라인 결과를 하나의 markdown 문서로 합칩니다.
        """
        return f"""
# 멀티 에이전트 파이프라인 결과

## 1. 업무 경험 분석

{result.experience_analysis}

## 2. 직무 매칭 분석

{result.job_match_report}

## 3. 이력서 초안

{result.resume_draft}

## 4. 포트폴리오 초안

{result.portfolio_draft}

## 5. 리뷰 결과

{result.review_report}
""".strip()


def main() -> None:
    load_dotenv()

    model = create_model()
    pipeline = ResumePipeline(model)

    result = pipeline.run()
    final_markdown = pipeline.render_markdown(result)

    output_path = save_markdown(
        filename="resume_pipeline_result.md",
        content=final_markdown,
    )

    print(f"파이프라인 결과 저장 완료: {output_path}")


if __name__ == "__main__":
    main()

코드 설명

1. PipelineResult

@dataclass
class PipelineResult:

PipelineResult는 파이프라인에서 생성된 중간 결과를 모아두기 위한 클래스입니다.

이번 파이프라인에서는 아래 결과가 순서대로 생성됩니다.

• experience_analysis
• job_match_report
• resume_draft
• portfolio_draft
• review_report

중간 결과를 하나의 객체로 모아두면 나중에 저장하거나 디버깅하기 쉽습니다.

2. create_pipeline_agent

def create_pipeline_agent(model: ChatOpenAI, name: str, system_prompt: str):

이 함수는 파이프라인의 각 단계를 담당할 에이전트를 생성합니다.

모든 에이전트는 같은 모델을 사용하지만, 서로 다른 시스템 프롬프트를 가집니다.

3. ResumePipeline 클래스

class ResumePipeline:

ResumePipeline은 여러 에이전트를 순서대로 실행하는 클래스입니다.

이전 글의 SupervisorAgent는 어떤 도구를 호출할지 LLM이 판단했습니다.

이번 글의 ResumePipeline은 실행 순서를 Python 코드가 직접 정합니다.

4. invoke_agent

def invoke_agent(self, agent, content: str) -> str:

invoke_agent는 에이전트를 실행하는 공통 함수입니다.

LangChain 에이전트는 messages 형식으로 입력을 받습니다.

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": content,
            }
        ]
    }
)

5. run

def run(self) -> PipelineResult:

run 메서드가 실제 파이프라인의 핵심입니다.

이 메서드 안에서 에이전트들이 정해진 순서대로 실행됩니다.

experience_analysis = self.invoke_agent(...)
job_match_report = self.invoke_agent(...)
resume_draft = self.invoke_agent(...)
portfolio_draft = self.invoke_agent(...)
review_report = self.invoke_agent(...)

여기서 중요한 점은 앞 단계의 결과가 다음 단계의 입력이 된다는 것입니다.

예를 들어 JobMatcherAgent는 원본 업무 경험만 보는 것이 아니라, ExperienceAnalyzerAgent가 정리한 분석 결과를 입력으로 받습니다.

job_match_report = self.invoke_agent(
    self.job_matcher_agent,
    f"""
[업무 경험 분석]
{experience_analysis}

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

 

그리고 이 결과들을 모아서 결국 markdown 파일로 만드는 것이죠.

정리

• 멀티 에이전트는 역할을 나누는 구조다.
• 파이프라인은 실행 순서를 코드로 고정하는 구조다.
• 앞 단계의 결과가 다음 단계의 입력이 된다.
• 정해진 업무 흐름이 있을 때는 파이프라인 방식이 이해하기 쉽다.
• 분기, 반복, 상태 관리가 필요해지면 LangGraph 같은 것을 활용해서 상태 기반 시스템으로 확장할 수 있다.