워크플로우와 에이전트 패턴

개요

이 가이드에서는 에이전트 시스템의 일반적인 패턴을 검토합니다. 이러한 시스템을 설명할 때 “워크플로우”와 “에이전트”를 구분하는 것이 유용합니다. Anthropic의 Building Effective Agents 블로그 포스트에서 이 차이를 잘 설명하고 있습니다:

워크플로우(Workflows)는 LLM과 도구가 미리 정의된 코드 경로를 통해 조율되는 시스템입니다. 반면 에이전트(Agents)는 LLM이 자체 프로세스와 도구 사용을 동적으로 지시하고, 작업 수행 방법에 대한 제어권을 유지하는 시스템입니다.

LangGraph는 에이전트와 워크플로우를 구축할 때 지속성(persistence), 스트리밍, 디버깅 지원, 배포 기능 등 다양한 이점을 제공합니다.

구성 요소: 증강된 LLM

LLM은 워크플로우와 에이전트 구축을 지원하는 증강 기능을 가지고 있습니다. 여기에는 구조화된 출력(structured outputs)과 도구 호출(tool calling)이 포함됩니다.

구조화된 출력

from pydantic import BaseModel, Field
 
class SearchQuery(BaseModel):
    search_query: str = Field(None, description="웹 검색에 최적화된 쿼리")
    justification: str = Field(
        None, description="이 쿼리가 사용자 요청과 관련된 이유"
    )
 
# 구조화된 출력을 위한 스키마로 LLM 증강
structured_llm = llm.with_structured_output(SearchQuery)
 
# 증강된 LLM 호출
output = structured_llm.invoke("칼슘 CT 점수와 고콜레스테롤의 관계는?")

도구 호출

def multiply(a: int, b: int) -> int:
    return a * b
 
# 도구로 LLM 증강
llm_with_tools = llm.bind_tools([multiply])
 
# 도구 호출을 트리거하는 입력으로 LLM 호출
msg = llm_with_tools.invoke("2 곱하기 3은?")
msg.tool_calls

워크플로우 패턴

1. 프롬프트 체이닝 (Prompt Chaining)

프롬프트 체이닝에서는 각 LLM 호출이 이전 호출의 출력을 처리합니다.

Anthropic 블로그에서 설명하는 바와 같이:

워크플로우: 프롬프트 체이닝

프롬프트 체이닝은 작업을 일련의 단계로 분해하며, 각 LLM 호출은 이전 호출의 출력을 처리합니다. 중간 단계에서 프로그래밍 방식의 검사(“게이트”)를 추가하여 프로세스가 제대로 진행되고 있는지 확인할 수 있습니다.

적용 시기: 작업을 고정된 하위 작업으로 쉽고 명확하게 분해할 수 있는 상황에 이상적입니다. 주요 목표는 각 LLM 호출을 더 쉬운 작업으로 만들어 지연 시간을 희생하고 정확도를 높이는 것입니다.

프롬프트 체이닝이 유용한 예:

• 마케팅 카피를 생성한 다음 다른 언어로 번역하기
• 문서 개요를 작성하고, 개요가 특정 기준을 충족하는지 확인한 다음, 개요를 기반으로 문서 작성하기

LangGraph에서의 구현

LangGraph에서 프롬프트 체이닝 패턴을 구현하는 방법은 워크플로우와 에이전트 패턴 - 프롬프트 체이닝을 참조하세요.

원본 링크

LangGraph 구현 예제

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
 
# 그래프 상태
class State(TypedDict):
    topic: str
    joke: str
    improved_joke: str
    final_joke: str
 
# 노드
def generate_joke(state: State):
    """첫 번째 LLM 호출로 초기 농담 생성"""
    msg = llm.invoke(f"{state['topic']}에 대한 짧은 농담을 작성하세요")
    return {"joke": msg.content}
 
def check_punchline(state: State):
    """농담에 펀치라인이 있는지 확인하는 게이트 함수"""
    if "?" in state["joke"] or "!" in state["joke"]:
        return "Pass"
    return "Fail"
 
def improve_joke(state: State):
    """두 번째 LLM 호출로 농담 개선"""
    msg = llm.invoke(f"이 농담에 말장난을 추가하여 더 재미있게 만드세요: {state['joke']}")
    return {"improved_joke": msg.content}
 
def polish_joke(state: State):
    """세 번째 LLM 호출로 최종 다듬기"""
    msg = llm.invoke(f"이 농담에 놀라운 반전을 추가하세요: {state['improved_joke']}")
    return {"final_joke": msg.content}
 
# 워크플로우 구축
workflow = StateGraph(State)
workflow.add_node("generate_joke", generate_joke)
workflow.add_node("improve_joke", improve_joke)
workflow.add_node("polish_joke", polish_joke)
 
workflow.add_edge(START, "generate_joke")
workflow.add_conditional_edges(
    "generate_joke", check_punchline, {"Fail": "improve_joke", "Pass": END}
)
workflow.add_edge("improve_joke", "polish_joke")
workflow.add_edge("polish_joke", END)
 
chain = workflow.compile()

리소스:

• LangSmith Trace
• LangChain Academy: Prompt Chaining 레슨

2. 병렬화 (Parallelization)

병렬화를 통해 LLM이 작업에 대해 동시에 작업하고 출력을 프로그래밍 방식으로 집계할 수 있습니다.

병렬화는 두 가지 주요 변형으로 나타납니다:

• 섹션화(Sectioning): 작업을 병렬로 실행되는 독립적인 하위 작업으로 분할
• 투표(Voting): 다양한 출력을 얻기 위해 동일한 작업을 여러 번 실행

적용 시기: 분할된 하위 작업을 속도를 위해 병렬화할 수 있거나, 더 높은 신뢰도를 위해 여러 관점이나 시도가 필요한 경우에 효과적입니다.

LangGraph 구현 예제

class State(TypedDict):
    topic: str
    joke: str
    story: str
    poem: str
    combined_output: str
 
def call_llm_1(state: State):
    """농담 생성"""
    msg = llm.invoke(f"{state['topic']}에 대한 농담을 작성하세요")
    return {"joke": msg.content}
 
def call_llm_2(state: State):
    """이야기 생성"""
    msg = llm.invoke(f"{state['topic']}에 대한 이야기를 작성하세요")
    return {"story": msg.content}
 
def call_llm_3(state: State):
    """시 생성"""
    msg = llm.invoke(f"{state['topic']}에 대한 시를 작성하세요")
    return {"poem": msg.content}
 
def aggregator(state: State):
    """농담, 이야기, 시를 하나의 출력으로 결합"""
    combined = f"{state['topic']}에 대한 이야기, 농담, 시입니다!\n\n"
    combined += f"이야기:\n{state['story']}\n\n"
    combined += f"농담:\n{state['joke']}\n\n"
    combined += f"시:\n{state['poem']}"
    return {"combined_output": combined}
 
# 워크플로우 구축
parallel_builder = StateGraph(State)
parallel_builder.add_node("call_llm_1", call_llm_1)
parallel_builder.add_node("call_llm_2", call_llm_2)
parallel_builder.add_node("call_llm_3", call_llm_3)
parallel_builder.add_node("aggregator", aggregator)
 
# 병렬 실행을 위한 엣지 추가
parallel_builder.add_edge(START, "call_llm_1")
parallel_builder.add_edge(START, "call_llm_2")
parallel_builder.add_edge(START, "call_llm_3")
parallel_builder.add_edge("call_llm_1", "aggregator")
parallel_builder.add_edge("call_llm_2", "aggregator")
parallel_builder.add_edge("call_llm_3", "aggregator")
parallel_builder.add_edge("aggregator", END)
 
parallel_workflow = parallel_builder.compile()

리소스:

• LangSmith Trace
• 병렬화 문서
• LangChain Academy: 병렬화 레슨

3. 라우팅 (Routing)

라우팅은 입력을 분류하고 특화된 후속 작업으로 안내합니다.

라우팅은 입력을 분류하고 특화된 후속 작업으로 안내합니다. 이 워크플로우는 관심사의 분리와 더 특화된 프롬프트 구축을 가능하게 합니다. 이 워크플로우가 없다면 한 종류의 입력에 대한 최적화가 다른 입력의 성능을 저하시킬 수 있습니다.

적용 시기: 별도로 처리하는 것이 더 나은 명확한 카테고리가 있고, LLM이나 전통적인 분류 모델/알고리즘으로 분류를 정확하게 처리할 수 있는 복잡한 작업에 적합합니다.

LangGraph 구현 예제

from typing_extensions import Literal
from langchain_core.messages import HumanMessage, SystemMessage
 
# 라우팅 로직으로 사용할 구조화된 출력 스키마
class Route(BaseModel):
    step: Literal["poem", "story", "joke"] = Field(
        None, description="라우팅 프로세스의 다음 단계"
    )
 
# 구조화된 출력으로 LLM 증강
router = llm.with_structured_output(Route)
 
def llm_call_router(state: State):
    """입력을 적절한 노드로 라우팅"""
    decision = router.invoke([
        SystemMessage(content="사용자 요청에 따라 입력을 story, joke, poem으로 라우팅하세요."),
        HumanMessage(content=state["input"]),
    ])
    return {"decision": decision.step}
 
def route_decision(state: State):
    """방문할 다음 노드 이름 반환"""
    if state["decision"] == "story":
        return "llm_call_1"
    elif state["decision"] == "joke":
        return "llm_call_2"
    elif state["decision"] == "poem":
        return "llm_call_3"
 
# 워크플로우 구축
router_builder = StateGraph(State)
router_builder.add_node("llm_call_router", llm_call_router)
router_builder.add_edge(START, "llm_call_router")
router_builder.add_conditional_edges(
    "llm_call_router",
    route_decision,
    {
        "llm_call_1": "llm_call_1",
        "llm_call_2": "llm_call_2",
        "llm_call_3": "llm_call_3",
    },
)

리소스:

• LangSmith Trace
• LangChain Academy: 라우팅 레슨
• 예제: RAG 라우팅 워크플로우

4. 조율자-작업자 (Orchestrator-Worker)

조율자-작업자 패턴에서는 중앙 LLM이 작업을 분해하고 작업자 LLM에 위임하며 결과를 종합합니다.

조율자-작업자 워크플로우에서는 중앙 LLM이 작업을 동적으로 분해하고, 작업자 LLM에 위임하며, 결과를 종합합니다.

적용 시기: 필요한 하위 작업을 예측할 수 없는 복잡한 작업에 적합합니다. 병렬화와 위상적으로는 유사하지만, 주요 차이점은 유연성입니다. 하위 작업이 미리 정의되지 않고 특정 입력에 따라 조율자가 결정합니다.

LangGraph에서 작업자 생성

LangGraph는 Send API를 통해 조율자-작업자 워크플로우를 지원합니다. 이를 통해 작업자 노드를 동적으로 생성하고 각각에 특정 입력을 전송할 수 있습니다.

from typing import Annotated, List
import operator
from langgraph.types import Send
 
class Section(BaseModel):
    name: str = Field(description="보고서 섹션 이름")
    description: str = Field(description="이 섹션에서 다룰 주요 주제 및 개념의 간략한 개요")
 
class Sections(BaseModel):
    sections: List[Section] = Field(description="보고서 섹션")
 
planner = llm.with_structured_output(Sections)
 
# 그래프 상태
class State(TypedDict):
    topic: str
    sections: list[Section]
    completed_sections: Annotated[list, operator.add]
    final_report: str
 
# 작업자 상태
class WorkerState(TypedDict):
    section: Section
    completed_sections: Annotated[list, operator.add]
 
def orchestrator(state: State):
    """보고서 계획을 생성하는 조율자"""
    report_sections = planner.invoke([
        SystemMessage(content="보고서 계획을 생성하세요."),
        HumanMessage(content=f"보고서 주제: {state['topic']}"),
    ])
    return {"sections": report_sections.sections}
 
def llm_call(state: WorkerState):
    """보고서 섹션을 작성하는 작업자"""
    section = llm.invoke([
        SystemMessage(content="제공된 이름과 설명에 따라 보고서 섹션을 작성하세요."),
        HumanMessage(content=f"섹션 이름: {state['section'].name}, 설명: {state['section'].description}"),
    ])
    return {"completed_sections": [section.content]}
 
def assign_workers(state: State):
    """계획의 각 섹션에 작업자 할당"""
    return [Send("llm_call", {"section": s}) for s in state["sections"]]
 
# 워크플로우 구축
orchestrator_worker_builder = StateGraph(State)
orchestrator_worker_builder.add_node("orchestrator", orchestrator)
orchestrator_worker_builder.add_node("llm_call", llm_call)
orchestrator_worker_builder.add_edge(START, "orchestrator")
orchestrator_worker_builder.add_conditional_edges("orchestrator", assign_workers, ["llm_call"])

리소스:

• LangSmith Trace
• LangChain Academy: 조율자-작업자 레슨
• 예제: 보고서 계획 및 작성 프로젝트

5. 평가자-최적화자 (Evaluator-Optimizer)

평가자-최적화자 워크플로우에서는 한 LLM 호출이 응답을 생성하고 다른 LLM이 루프에서 평가와 피드백을 제공합니다.

적용 시기: 명확한 평가 기준이 있고 반복적인 개선이 측정 가능한 가치를 제공할 때 특히 효과적입니다. 두 가지 적합성 신호는: 첫째, 사람이 피드백을 표현할 때 LLM 응답이 명백히 개선될 수 있으며, 둘째, LLM이 그러한 피드백을 제공할 수 있다는 것입니다.

LangGraph 구현 예제

class Feedback(BaseModel):
    grade: Literal["funny", "not funny"] = Field(description="농담이 재미있는지 판단")
    feedback: str = Field(description="농담이 재미없다면 개선 방법 피드백 제공")
 
evaluator = llm.with_structured_output(Feedback)
 
def llm_call_generator(state: State):
    """LLM이 농담 생성"""
    if state.get("feedback"):
        msg = llm.invoke(f"{state['topic']}에 대한 농담을 작성하되 피드백을 고려하세요: {state['feedback']}")
    else:
        msg = llm.invoke(f"{state['topic']}에 대한 농담을 작성하세요")
    return {"joke": msg.content}
 
def llm_call_evaluator(state: State):
    """LLM이 농담 평가"""
    grade = evaluator.invoke(f"농담을 평가하세요: {state['joke']}")
    return {"funny_or_not": grade.grade, "feedback": grade.feedback}
 
def route_joke(state: State):
    """평가자의 피드백에 따라 농담 생성기로 돌아가거나 종료"""
    if state["funny_or_not"] == "funny":
        return "Accepted"
    elif state["funny_or_not"] == "not funny":
        return "Rejected + Feedback"
 
# 워크플로우 구축
optimizer_builder = StateGraph(State)
optimizer_builder.add_node("llm_call_generator", llm_call_generator)
optimizer_builder.add_node("llm_call_evaluator", llm_call_evaluator)
optimizer_builder.add_edge(START, "llm_call_generator")
optimizer_builder.add_edge("llm_call_generator", "llm_call_evaluator")
optimizer_builder.add_conditional_edges(
    "llm_call_evaluator",
    route_joke,
    {"Accepted": END, "Rejected + Feedback": "llm_call_generator"},
)

리소스:

• LangSmith Trace
• 예제: 보고서 개선 어시스턴트
• 예제: 환각 탐지 RAG 워크플로우

에이전트 패턴

도구 호출 에이전트 (Tool-Calling Agent)

에이전트는 일반적으로 환경 피드백을 기반으로 (도구 호출을 통해) 작업을 수행하는 LLM이 루프에서 구현됩니다.

Anthropic 블로그에서 설명하는 바와 같이:

에이전트는 정교한 작업을 처리할 수 있지만 구현은 종종 간단합니다. 일반적으로 루프에서 도구를 사용하여 환경 피드백을 기반으로 하는 LLM일 뿐입니다. 따라서 도구 세트와 문서를 명확하고 신중하게 설계하는 것이 중요합니다.

적용 시기: 필요한 단계 수를 예측하기 어렵거나 불가능한 개방형 문제에 사용할 수 있으며, 고정된 경로를 하드코딩할 수 없습니다. LLM은 잠재적으로 많은 턴 동안 작동하며 의사 결정에 어느 정도 신뢰가 있어야 합니다.

LangGraph 구현 예제

from langchain_core.tools import tool
 
@tool
def multiply(a: int, b: int) -> int:
    """a와 b를 곱합니다."""
    return a * b
 
@tool
def add(a: int, b: int) -> int:
    """a와 b를 더합니다."""
    return a + b
 
# 도구로 LLM 증강
tools = [add, multiply]
llm_with_tools = llm.bind_tools(tools)
 
from langgraph.graph import MessagesState
 
def llm_call(state: MessagesState):
    """LLM이 도구 호출 여부 결정"""
    return {
        "messages": [
            llm_with_tools.invoke([
                SystemMessage(content="입력 세트에 대한 산술 연산을 수행하는 도우미입니다.")
            ] + state["messages"])
        ]
    }
 
def should_continue(state: MessagesState) -> Literal["Action", END]:
    """LLM이 도구를 호출했는지에 따라 계속할지 결정"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "Action"
    return END
 
# 워크플로우 구축
agent_builder = StateGraph(MessagesState)
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges("llm_call", should_continue, {"Action": "environment", END: END})
agent_builder.add_edge("environment", "llm_call")
agent = agent_builder.compile()

사전 구축된 에이전트

LangGraph는 create_react_agent 함수를 통해 에이전트를 생성하는 사전 구축된 방법도 제공합니다:

from langgraph.prebuilt import create_react_agent
 
pre_built_agent = create_react_agent(llm, tools=tools)

리소스:

• LangSmith Trace
• LangChain Academy: 에이전트 레슨
• 사전 구축된 에이전트 문서

LangGraph가 제공하는 것

위의 각 패턴을 LangGraph로 구축하면 다음과 같은 기능을 얻을 수 있습니다:

지속성: 휴먼-인-더-루프

LangGraph 지속성 계층은 작업의 중단 및 승인을 지원합니다. 휴먼-인-더-루프를 참조하세요.

지속성: 메모리

LangGraph 지속성 계층은 대화형(단기) 메모리와 장기 메모리를 지원합니다. LangChain Academy의 모듈 2와 5를 참조하세요.

스트리밍

LangGraph는 워크플로우/에이전트 출력 또는 중간 상태를 스트리밍하는 여러 방법을 제공합니다. LangChain Academy의 모듈 3을 참조하세요.

배포

LangGraph는 배포, 관찰 가능성, 평가를 위한 쉬운 진입로를 제공합니다. LangGraph 배포 가이드를 참조하세요.

관련 문서

• 에이전트란 무엇인가요?
• 에이전트 아키텍처 개념
• LLM 에이전트 아키텍처 패턴
• 사전 구축된 에이전트로 빠르게 시작하기