핸드오프로 고객 지원 만들기

핸드오프로 고객 지원 만들기 (Build customer support with handoffs)

핸드오프(상태 머신) 패턴은 에이전트가 작업의 여러 상태를 거치며 동작이 변하는 워크플로를 설명해요. 이 튜토리얼에서는 도구 호출로 단일 에이전트의 설정을 동적으로 바꾸는 방식으로 상태 머신을 구현하는 법을 배워요. 에이전트가 사용할 수 있는 도구와 지시를 현재 상태에 따라 업데이트하는 방식이죠. 상태는 에이전트의 과거 행동(도구 호출), 외부 상태(API 호출 결과), 심지어 초기 사용자 입력(예: 분류기로 사용자 의도를 판단) 등 여러 소스에서 결정될 수 있습니다.

출처: LangChain 공식 문서 — multi-agent/handoffs-customer-support

이 튜토리얼에서 만들 고객 지원 에이전트는 이렇게 동작해요:

• 진행 전에 워런티 정보를 수집합니다.
• 이슈를 하드웨어 또는 소프트웨어로 분류합니다.
• 해결책을 제공하거나 사람 지원으로 에스컬레이션합니다.
• 여러 턴에 걸쳐 대화 상태를 유지합니다.

서브에이전트 패턴처럼 서브에이전트를 도구로 호출하는 것과 달리, 상태 머신 패턴은 워크플로 진행에 따라 설정이 변하는 단일 에이전트를 사용해요. 각 "단계(step)"는 같은 기반 에이전트의 서로 다른 설정(시스템 프롬프트 + 도구)일 뿐이며, 상태에 따라 동적으로 선택됩니다.

설정 (Setup)

이 튜토리얼은 langchain 패키지가 필요해요. 자세한 설치는 설치 가이드를 참고하세요. LangSmith를 설정하면 에이전트 내부에서 일어나는 일을 살펴볼 수 있고, 적절한 환경 변수(LANGSMITH_TRACING, LANGSMITH_API_KEY 등)를 설정하세요.

LLM은 LangChain의 통합 목록(OpenAI, Anthropic, Azure, Google Gemini, AWS Bedrock, HuggingFace, OpenRouter)에서 채팅 모델을 고르면 돼요.

1. 커스텀 상태 정의 (Define custom state)

먼저 현재 활성화된 단계를 추적하는 커스텀 상태 스키마를 정의합니다.

from langchain.agents import AgentState
from typing_extensions import NotRequired
from typing import Literal

# Define the possible workflow steps
SupportStep = Literal["warranty_collector", "issue_classifier", "resolution_specialist"]

class SupportState(AgentState):
    """State for customer support workflow."""
    current_step: NotRequired[SupportStep]
    warranty_status: NotRequired[Literal["in_warranty", "out_of_warranty"]]
    issue_type: NotRequired[Literal["hardware", "software"]]

current_step 필드가 상태 머신 패턴의 핵심이에요 — 매 턴 어떤 설정(프롬프트 + 도구)을 로드할지 결정합니다.

2. 워크플로 상태를 관리하는 도구 만들기 (Create tools that manage workflow state)

워크플로 상태를 업데이트하는 도구를 만들어요. 이 도구들은 에이전트가 정보를 기록하고 다음 단계로 전환하게 해줍니다. 핵심은 Command로 상태를 업데이트한다는 것, 특히 current_step 필드를 포함한다는 점이에요.

from langchain.tools import tool, ToolRuntime
from langchain.messages import ToolMessage
from langgraph.types import Command

@tool
def record_warranty_status(
    status: Literal["in_warranty", "out_of_warranty"],
    runtime: ToolRuntime[None, SupportState],
) -> Command:
    """Record the customer's warranty status and transition to issue classification."""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"Warranty status recorded as: {status}",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
            "warranty_status": status,
            "current_step": "issue_classifier",
        }
    )

@tool
def record_issue_type(
    issue_type: Literal["hardware", "software"],
    runtime: ToolRuntime[None, SupportState],
) -> Command:
    """Record the type of issue and transition to resolution specialist."""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"Issue type recorded as: {issue_type}",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
            "issue_type": issue_type,
            "current_step": "resolution_specialist",
        }
    )

@tool
def escalate_to_human(reason: str) -> str:
    """Escalate the case to a human support specialist."""
    # In a real system, this would create a ticket, notify staff, etc.
    return f"Escalating to human support. Reason: {reason}"

@tool
def provide_solution(solution: str) -> str:
    """Provide a solution to the customer's issue."""
    return f"Solution provided: {solution}"

record_warranty_status와 record_issue_type이 데이터(warranty_status, issue_type) 와 current_step을 모두 업데이트하는 Command 객체를 반환한다는 점에 주목하세요. 이것이 상태 머신이 동작하는 방식입니다 — 도구가 워크플로 진행을 제어하죠.

3. 단계 설정 정의 (Define step configurations)

각 단계의 프롬프트와 도구를 정의해요. 먼저 각 단계의 프롬프트를 상수로 정의합니다.

# Define prompts as constants for easy reference
WARRANTY_COLLECTOR_PROMPT = """You are a customer support agent helping with device issues.
CURRENT STAGE: Warranty verification
At this step, you need to:
1. Greet the customer warmly
2. Ask if their device is under warranty
3. Use record_warranty_status to record their response and move to the next step
Be conversational and friendly. Don't ask multiple questions at once."""

ISSUE_CLASSIFIER_PROMPT = """You are a customer support agent helping with device issues.
CURRENT STAGE: Issue classification
CUSTOMER INFO: Warranty status is {warranty_status}
At this step, you need to:
1. Ask the customer to describe their issue
2. Determine if it's a hardware issue (physical damage, broken parts) or software issue (app crashes, performance)
3. Use record_issue_type to record the classification and move to the next step
If unclear, ask clarifying questions before classifying."""

RESOLUTION_SPECIALIST_PROMPT = """You are a customer support agent helping with device issues.
CURRENT STAGE: Resolution
CUSTOMER INFO: Warranty status is {warranty_status}, issue type is {issue_type}
At this step, you need to:
1. For SOFTWARE issues: provide troubleshooting steps using provide_solution
2. For HARDWARE issues:
   - If IN WARRANTY: explain warranty repair process using provide_solution
   - If OUT OF WARRANTY: escalate_to_human for paid repair options
Be specific and helpful in your solutions."""

그 다음 단계 이름을 설정으로 매핑하는 딕셔너리를 만듭니다.

# Step configuration: maps step name to (prompt, tools, required_state)
STEP_CONFIG = {
    "warranty_collector": {
        "prompt": WARRANTY_COLLECTOR_PROMPT,
        "tools": [record_warranty_status],
        "requires": [],
    },
    "issue_classifier": {
        "prompt": ISSUE_CLASSIFIER_PROMPT,
        "tools": [record_issue_type],
        "requires": ["warranty_status"],
    },
    "resolution_specialist": {
        "prompt": RESOLUTION_SPECIALIST_PROMPT,
        "tools": [provide_solution, escalate_to_human],
        "requires": ["warranty_status", "issue_type"],
    },
}

이 딕셔너리 기반 설정 덕분에:

• 모든 단계를 한눈에 볼 수 있고
• 새 단계를 그냥 항목 하나만 추가하면 되며
• 워크플로 의존성(requires 필드)을 이해할 수 있고
• 상태 변수(예: {warranty_status})가 있는 프롬프트 템플릿을 쓸 수 있어요.

4. 단계 기반 미들웨어 만들기 (Create step-based middleware)

상태에서 current_step을 읽고 적절한 설정을 적용하는 미들웨어를 만들어요. 깔끔한 구현을 위해 @wrap_model_call 데코레이터를 씁니다.

from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable

@wrap_model_call
def apply_step_config(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    """Configure agent behavior based on the current step."""
    # Get current step (defaults to warranty_collector for first interaction)
    current_step = request.state.get("current_step", "warranty_collector")

    # Look up step configuration
    stage_config = STEP_CONFIG[current_step]

    # Validate required state exists
    for key in stage_config["requires"]:
        if request.state.get(key) is None:
            raise ValueError(f"{key} must be set before reaching {current_step}")

    # Format prompt with state values (supports {warranty_status}, {issue_type}, etc.)
    system_prompt = stage_config["prompt"].format(**request.state)

    # Inject system prompt and step-specific tools
    request = request.override(
        system_prompt=system_prompt,
        tools=stage_config["tools"],
    )
    return handler(request)

이 미들웨어는:

1. 현재 단계를 읽고: 상태에서 current_step을 얻습니다(기본 warranty_collector).
2. 설정을 찾고: STEP_CONFIG에서 일치하는 항목을 찾습니다.
3. 의존성을 검증: 필수 상태 필드가 존재하는지 확인합니다.
4. 프롬프트를 포맷: 상태 값을 프롬프트 템플릿에 주입합니다.
5. 설정을 적용: 시스템 프롬프트와 사용 가능한 도구를 재정의합니다.

request.override() 메서드가 핵심이에요 — 별도의 에이전트 인스턴스를 만들지 않고도 상태에 따라 에이전트 동작을 동적으로 바꿀 수 있게 해줍니다.

5. 에이전트 만들기 (Create the agent)

이제 단계 기반 미들웨어와 상태 지속을 위한 체크포인터가 있는 에이전트를 만듭니다.

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

# Collect all tools from all step configurations
all_tools = [
    record_warranty_status,
    record_issue_type,
    provide_solution,
    escalate_to_human,
]

# Create the agent with step-based configuration
agent = create_agent(
    model,
    tools=all_tools,
    state_schema=SupportState,
    middleware=[apply_step_config],
    checkpointer=InMemorySaver(),
)

왜 체크포인터가 필요한가요? 체크포인터는 대화 턴 사이의 상태를 유지합니다. 없다면 current_step 상태가 사용자 메시지 사이에 사라져서 워크플로가 깨져요.

6. 워크플로 테스트 (Test the workflow)

전체 워크플로를 테스트해봐요. 여기서는 요약 미들웨어(SummarizationMiddleware)도 함께 적용한 버전을 보여드립니다.

from langchain.agents.middleware import SummarizationMiddleware

# Create the agent with step-based configuration and summarization
agent = create_agent(
    model,
    tools=all_tools,
    state_schema=SupportState,
    middleware=[
        apply_step_config,
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=("tokens", 4000),
            keep=("messages", 10)
        )
    ],
    checkpointer=InMemorySaver(),
)

# Test the workflow
if __name__ == "__main__":
    thread_id = str(uuid7())
    config = {"configurable": {"thread_id": thread_id}}

    result = agent.invoke(
        {"messages": [HumanMessage("Hi, my phone screen is cracked")]}, config
    )
    result = agent.invoke(
        {"messages": [HumanMessage("Yes, it's still under warranty")]}, config
    )
    result = agent.invoke(
        {"messages": [HumanMessage("The screen is physically cracked from dropping it")]}, config
    )
    result = agent.invoke(
        {"messages": [HumanMessage("What should I do?")]}, config
    )

    for msg in result['messages']:
        msg.pretty_print()

다음 단계 (Next steps)

• 집중 오케스트레이션을 위한 subagents 패턴 배우기
• 더 많은 동적 동작을 위한 미들웨어 탐색하기
• 패턴을 비교하려면 멀티 에이전트 개요 읽기
• LangSmith로 멀티 에이전트 시스템 디버깅하고 모니터링하기

더 알아보기 (Learn more)

• Handoffs 패턴
• 미들웨어 개요
• 커스텀 미들웨어