핸드오프
핸드오프 (Handoffs)
핸드오프(handoffs) 아키텍처에서는 상태에 따라 동작이 동적으로 변해요. 핵심 메커니즘은 이렇습니다: 도구가 턴 사이에 지속되는 상태 변수(예: current_step 또는 active_agent)를 업데이트하고, 시스템이 이 변수를 읽어 동작을 조정하죠 — 다른 설정(시스템 프롬프트, 도구)을 적용하거나 다른 에이전트로 라우팅합니다. 이 패턴은 서로 다른 에이전트 사이의 핸드오프와 단일 에이전트 내부의 동적 설정 변경을 모두 지원해요.
핸드오프라는 용어는 OpenAI가 도구 호출(예: transfer_to_sales_agent)로 에이전트나 상태 사이의 제어권을 이전하는 것에서 유래했어요.
핵심 특징 (Key characteristics)
- 상태 기반 동작: 상태 변수(예:
current_step또는active_agent)에 따라 동작이 바뀝니다. - 도구 기반 전환: 도구가 상태 변수를 업데이트해 상태 사이를 이동합니다.
- 직접 사용자 상호작용: 각 상태의 설정이 사용자 메시지를 직접 처리합니다.
- 지속 상태: 상태가 대화 턴 사이에서 생존합니다.
언제 쓰나요? (When to use)
순차 제약을 강제해야 하거나(선행 조건이 충족된 후에만 능력을 개방), 에이전트가 서로 다른 상태에서 사용자와 직접 대화해야 하거나, 다단계 대화 흐름을 만들 때 핸드오프 패턴을 쓰면 돼요. 특히 고객 지원 시나리오에서 특정 순서로 정보를 수집해야 할 때 — 예를 들어 환불을 처리하기 전에 워런티 ID를 수집하는 식이죠 — 이 패턴이 아주 유용합니다.
기본 구현 (Basic implementation)
핵심 메커니즘은 상태를 업데이트하는 Command를 반환해서 새 단계나 에이전트로의 전환을 트리거하는 도구예요.
from langchain.tools import tool
from langchain.messages import ToolMessage
from langgraph.types import Command
@tool
def transfer_to_specialist(runtime) -> Command:
"""Transfer to the specialist agent."""
return Command(
update={
"messages": [
ToolMessage(
content="Transferred to specialist",
tool_call_id=runtime.tool_call_id
)
],
"current_step": "specialist" # Triggers behavior change
}
)
왜 ToolMessage를 포함할까요? LLM이 도구를 호출하면 응답을 기대합니다. tool_call_id가 일치하는 ToolMessage가 이 요청-응답 주기를 완성합니다. 없으면 대화 기록이 손상되죠. 핸드오프 도구가 메시지를 업데이트할 때는 항상 필수예요.
구현 방식 (Implementation approaches)
핸드오프를 구현하는 방법은 두 가지입니다: 단일 에이전트 + 미들웨어(동적 설정을 가진 하나의 에이전트) 또는 다중 에이전트 서브그래프(그래프 노드로 분리된 서로 다른 에이전트).
단일 에이전트 + 미들웨어 (Single agent with middleware)
단일 에이전트가 상태에 따라 동작을 바꿔요. 미들웨어가 각 모델 호출을 가로채 시스템 프롬프트와 사용 가능한 도구를 동적으로 조정하고, 도구가 상태 변수를 업데이트해 전환을 트리거합니다.
from langchain.tools import ToolRuntime, tool
from langchain.messages import ToolMessage
from langgraph.types import Command
@tool
def record_warranty_status(
status: str,
runtime: ToolRuntime[None, SupportState]
) -> Command:
"""Record warranty status and transition to next step."""
return Command(
update={
"messages": [
ToolMessage(
content=f"Warranty status recorded: {status}",
tool_call_id=runtime.tool_call_id
)
],
"warranty_status": status,
"current_step": "specialist" # Update state to trigger transition
}
)
전체 구현:
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain.tools import tool, ToolRuntime
from langchain.messages import ToolMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from typing import Callable
# 1. Define state with current_step tracker
class SupportState(AgentState):
"""Track which step is currently active."""
current_step: str = "triage"
warranty_status: str | None = None
# 2. Tools update current_step via Command
@tool
def record_warranty_status(
status: str,
runtime: ToolRuntime[None, SupportState]
) -> Command:
"""Record warranty status and transition to next step."""
return Command(update={
"messages": [
ToolMessage(
content=f"Warranty status recorded: {status}",
tool_call_id=runtime.tool_call_id
)
],
"warranty_status": status,
# Transition to next step
"current_step": "specialist"
})
# 3. Middleware applies dynamic configuration based on current_step
@wrap_model_call
def apply_step_config(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Configure agent behavior based on current_step."""
step = request.state.get("current_step", "triage")
# Map steps to their configurations
configs = {
"triage": {
"prompt": "Collect warranty information...",
"tools": [record_warranty_status]
},
"specialist": {
"prompt": "Provide solutions based on warranty: {warranty_status}",
"tools": [provide_solution, escalate]
}
}
config = configs[step]
request = request.override(
system_prompt=config["prompt"].format(**request.state),
tools=config["tools"]
)
return handler(request)
# 4. Create agent with middleware
agent = create_agent(
model,
tools=[record_warranty_status, provide_solution, escalate],
state_schema=SupportState,
middleware=[apply_step_config],
checkpointer=InMemorySaver() # Persist state across turns
)
다중 에이전트 서브그래프 (Multiple agent subgraphs)
여러 개의 서로 다른 에이전트가 그래프의 개별 노드로 존재해요. 핸드오프 도구는 Command.PARENT를 사용해 다음에 실행할 노드를 지정하면서 에이전트 노드 사이를 이동합니다.
서브그래프 핸드오프는 신중한 **컨텍스트 엔지니어링**이 필요해요. 단일 에이전트 미들웨어(메시지 기록이 자연스럽게 흐르는)와 달리, 에이전트 사이를 오가는 메시지를 명시적으로 결정해야 합니다. 여기서 실수하면 에이전트가 손상된 대화 기록이나 부풀어 오른 컨텍스트를 받게 되죠. 아래 "컨텍스트 엔지니어링"을 참고하세요.
from langchain.messages import AIMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.types import Command
@tool
def transfer_to_sales(
runtime: ToolRuntime,
) -> Command:
"""Transfer to the sales agent."""
last_ai_message = next(
msg for msg in reversed(runtime.state["messages"]) if isinstance(msg, AIMessage)
)
transfer_message = ToolMessage(
content="Transferred to sales agent",
tool_call_id=runtime.tool_call_id,
)
return Command(
goto="sales_agent",
update={
"active_agent": "sales_agent",
"messages": [last_ai_message, transfer_message],
},
graph=Command.PARENT
)
판매(sales)와 지원(support) 에이전트를 분리한 멀티 에이전트 시스템의 예시를 보여드릴게요. 각 에이전트는 별개의 그래프 노드이고, 핸드오프 도구가 에이전트들이 서로 대화를 넘겨줄 수 있게 합니다.
from typing import Literal
from langchain.agents import AgentState, create_agent
from langchain.messages import AIMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing_extensions import NotRequired
# 1. Define state with active_agent tracker
class MultiAgentState(AgentState):
active_agent: NotRequired[str]
# 2. Create handoff tools
@tool
def transfer_to_sales(
runtime: ToolRuntime,
) -> Command:
"""Transfer to the sales agent."""
last_ai_message = next(
msg for msg in reversed(runtime.state["messages"]) if isinstance(msg, AIMessage)
)
transfer_message = ToolMessage(
content="Transferred to sales agent from support agent",
tool_call_id=runtime.tool_call_id,
)
return Command(
goto="sales_agent",
update={
"active_agent": "sales_agent",
"messages": [last_ai_message, transfer_message],
},
graph=Command.PARENT,
)
@tool
def transfer_to_support(
runtime: ToolRuntime,
) -> Command:
"""Transfer to the support agent."""
last_ai_message = next(
msg for msg in reversed(runtime.state["messages"]) if isinstance(msg, AIMessage)
)
transfer_message = ToolMessage(
content="Transferred to support agent from sales agent",
tool_call_id=runtime.tool_call_id,
)
return Command(
goto="support_agent",
update={
"active_agent": "support_agent",
"messages": [last_ai_message, transfer_message],
},
graph=Command.PARENT,
)
# 3. Create agents with handoff tools
sales_agent = create_agent(
model="google_genai:gemini-3.6-flash",
tools=[transfer_to_support],
system_prompt="You are a sales agent. Help with sales inquiries. If asked about technical issues or support, transfer to the support agent.",
)
support_agent = create_agent(
model="google_genai:gemini-3.6-flash",
tools=[transfer_to_sales],
system_prompt="You are a support agent. Help with technical issues. If asked about pricing or purchasing, transfer to the sales agent.",
)
# 4. Create agent nodes that invoke the agents
def call_sales_agent(state: MultiAgentState) -> Command:
"""Node that calls the sales agent."""
response = sales_agent.invoke(state)
return response
def call_support_agent(state: MultiAgentState) -> Command:
"""Node that calls the support agent."""
response = support_agent.invoke(state)
return response
# 5. Create router that checks if we should end or continue
def route_after_agent(
state: MultiAgentState,
) -> Literal["sales_agent", "support_agent", "__end__"]:
"""Route based on active_agent, or END if the agent finished without handoff."""
messages = state.get("messages", [])
# Check the last message - if it's an AIMessage without tool calls, we're done
if messages:
last_msg = messages[-1]
if isinstance(last_msg, AIMessage) and not last_msg.tool_calls:
return "__end__"
# Otherwise route to the active agent
active = state.get("active_agent", "sales_agent")
return active if active else "sales_agent"
def route_initial(
state: MultiAgentState,
) -> Literal["sales_agent", "support_agent"]:
"""Route to the active agent based on state, default to sales agent."""
return state.get("active_agent") or "sales_agent"
# 6. Build the graph
... # (그래프 구성: 노드 추가, START/END 엣지 연결)
for msg in result["messages"]:
msg.pretty_print()
대부분의 핸드오프 사용 사례에는 단일 에이전트 + 미들웨어를 쓰세요 — 훨씬 단순해요. 다중 에이전트 서브그래프는 리플렉션이나 검색 단계가 있는 복잡한 그래프인 노드처럼, 맞춤형 에이전트 구현이 필요할 때만 쓰는 게 좋습니다.
컨텍스트 엔지니어링 (Context engineering)
서브그래프 핸드오프에서는 에이전트 사이를 오가는 메시지를 정확히 제어합니다. 유효한 대화 기록을 유지하고, 하류 에이전트를 혼란시킬 수 있는 컨텍스트 비대화를 피하는 데 이 정밀함이 필수적이에요. 자세한 내용은 context engineering을 참고하세요.
핸드오프 동안 대화 기록이 유효하게 유지되도록 해야 합니다. LLM은 도구 호출이 그 응답과 짝지어지기를 기대하므로, Command.PARENT로 다른 에이전트에게 핸드오프할 때는 반드시 둘 다 포함해야 해요:
- 핸드오프를 트리거한 도구 호출을 담은
AIMessage - 핸드오프를 확인하는
ToolMessage(그 도구 호출에 대한 인공 응답)
이 짝이 없으면 받는 에이전트가 불완전한 대화를 보고 에러나 예기치 못한 동작을 일으킬 수 있어요. 아래 예시는 핸드오프 도구만 호출됐다고(병렬 도구 호출 없음) 가정합니다.
@tool
def transfer_to_sales(runtime: ToolRuntime) -> Command:
# Get the AI message that triggered this handoff
last_ai_message = runtime.state["messages"][-1]
# Create an artificial tool response to complete the pair
transfer_message = ToolMessage(
content="Transferred to sales agent",
tool_call_id=runtime.tool_call_id,
)
return Command(
goto="sales_agent",
update={
"active_agent": "sales_agent",
# Pass only these two messages, not the full subagent history
"messages": [last_ai_message, transfer_message],
},
graph=Command.PARENT,
)
왜 모든 서브에이전트 메시지를 넘기지 않을까요? 전체 서브에이전트 대화를 핸드오프에 포함할 수도 있지만, 종종 문제를 만듭니다. 받는 에이전트가 무관한 내부 추론에 혼란스러워질 수 있고, 토큰 비용도 불필요하게 늘어나요. 핸드오프 쌍만 넘기면 부모 그래프의 컨텍스트가 상위 수준 조율에 집중되게 유지할 수 있습니다. 받는 에이전트가 추가 컨텍스트가 필요하다면, 원시 메시지 기록을 넘기는 대신 ToolMessage 콘텐츠에 서브에이전트의 작업을 요약하는 것을 고려하세요.
사용자에게 제어권을 돌려줄 때는(에이전트의 턴을 종료) 마지막 메시지가 AIMessage인지 확인하세요. 이렇게 해야 대화 기록이 유효하게 유지되고 사용자 인터페이스에 에이전트가 작업을 끝냈음을 알릴 수 있어요.
구현 고려사항 (Implementation considerations)
멀티 에이전트 시스템을 설계할 때 고려할 것:
- 컨텍스트 필터링 전략: 각 에이전트가 전체 대화 기록을 받을까, 필터링된 일부를 받을까, 요약을 받을까? 역할에 따라 에이전트마다 다른 컨텍스트가 필요할 수 있어요.
- 도구 의미론: 핸드오프 도구가 라우팅 상태만 업데이트하는지, 부수 효과(side effects)도 수행하는지 명확히 하세요. 예를 들어
transfer_to_sales()가 지원 티켓도 생성해야 할까요, 아니면 그건 별도 액션이어야 할까요? - 토큰 효율: 컨텍스트 완전성과 토큰 비용의 균형을 맞추세요. 대화가 길어질수록 요약과 선택적 컨텍스트 전달이 더 중요해집니다.