기존 에이전트 단계 추가

기존 AI 에이전트를 워크플로의 단계로 추가하여 작업을 전문 에이전트로 이전할 수 있습니다. 이를 통해 각각 더 큰 프로세스의 특정 부분을 처리하는 여러 에이전트 간에 조정하는 멀티 에이전트 워크플로를 작성할 수 있습니다.

시작하기 전에

이 기능을 사용하려면 다음 단계를 완료하세요.

  • Gemini Enterprise 관리자가 앱의 기능 관리 설정에서 워크플로 빌더 기능 전환을 사용 설정했는지 확인합니다. 자세한 내용은 웹 앱 기능 관리를 참조하세요.

  • 추가하려는 에이전트가 조직에서 사용 가능한지 확인합니다. 이 기능은 다음 에이전트의 추가를 지원합니다.

기존 에이전트 단계 추가

앱

  1. 브라우저에서 Gemini Enterprise 웹 앱을 엽니다.
  2. 기존 에이전트로 이동하거나 새 에이전트를 만듭니다.
  3. 흐름 빌더에서 단계 추가 를 클릭합니다.
  4. 단계 추가 패널에서 기존 에이전트 를 클릭하여 섹션을 펼칩니다.

  5. 사용 가능한 에이전트 목록을 둘러봅니다.

  6. 추가하려는 에이전트를 클릭합니다.

    시스템에서 기존 에이전트 단계를 워크플로에 추가합니다. 선택한 에이전트는 워크플로의 일부로 실행되며 이전 단계에서 입력을 받고 출력을 이후 단계로 전달합니다.

  7. 캔버스에서 기존 에이전트 단계를 클릭하여 구성 패널을 열고 다음을 구성합니다.

    • 프롬프트: 에이전트를 트리거하는 데 사용되는 프롬프트 또는 안내를 입력합니다. 정적 텍스트를 포함하거나 이전 단계의 출력 (예: ${step_name.output})을 참조할 수 있습니다.

에이전트 단계는 선택한 에이전트에 대한 동적 참조입니다. 소스 에이전트가 업데이트되거나 새 버전이 게시되면 워크플로에서 자동으로 최신 버전을 사용합니다.

이 기능의 제한사항에 관한 자세한 내용은 알려진 문제 및 제한사항을 참고하세요.

확인 질문의 작동 방식

기존 에이전트가 워크플로에서 실행될 때 작업을 완료하는 데 누락된 정보가 필요한 경우 실행을 일시중지하고 사용자에게 확인 질문을 할 수 있습니다. 확인 질문은 참조된 에이전트가 input-required 신호 (A2A) 또는 RequestInput 이벤트 (ADK)를 내보낼 때 기본적으로 지원됩니다.

기존 에이전트 단계가 실행되고 추가 정보가 필요한 경우 다음을 수행합니다.

  1. 실행 일시중지: 워크플로가 기존 에이전트 단계에서 실행을 일시중지하고 인간 참여형 (HITL) 프롬프트를 시작합니다.
  2. 사용자가 프롬프트에 답변: Gemini Enterprise 웹 앱은 에이전트의 확인 질문을 사용자의 응답을 입력할 수 있는 텍스트 필드와 함께 사용자에게 표시합니다.
  3. 실행 재개: 사용자가 답변을 제출하면 워크플로가 기존 에이전트를 재개하고 사용자의 답변을 에이전트에 다시 전달합니다.
  4. 멀티턴 지원: 에이전트는 필요한 모든 정보를 수집하고 최종 출력을 생성할 때까지 여러 턴에 걸쳐 연속으로 여러 확인 질문을 할 수 있습니다.

컨텍스트 격리

확인 질문 교환 중에 참조된 에이전트는 사용자 및 노드에 구성된 초기 프롬프트와의 자체 상호작용 기록만 수신합니다. 워크플로의 다른 단계에 있는 출력 및 변수는 단계의 프롬프트에서 명시적으로 참조되지 않는 한 참조된 에이전트에 자동으로 전달되지 않습니다.

에이전트에 확인 질문 구현

확인 질문을 지원하려면 추가 정보가 필요할 때 프로토콜 수준 요청 입력 신호를 내보내도록 커스텀 에이전트를 구성합니다.

A2A 에이전트 (기본)

Agent-to-Agent (A2A) 프로토콜을 직접 구현하는 에이전트의 경우 메시지 페이로드에 확인 질문을 제공하여 TaskUpdater를 사용하여 input-required 상태로 작업을 반환합니다.

from a2a.server.tasks import TaskUpdater
from a2a.types import Message, Part, Role, TextPart

# Inside your custom agent's task executor class:
async def execute(self, context, event_queue):
    updater = TaskUpdater(event_queue, context.task_id, context.context_id)
    if context.current_task is None:
        await updater.submit()
    await updater.start_work()

    user_input = context.get_user_input()
    if "weather" in user_input and "tokyo" not in user_input.lower():
        await updater.requires_input(
            Message(
                message_id=f"q-{context.task_id}",
                role=Role.agent,
                parts=[Part(root=TextPart(
                    text="Which city would you like the weather for?"))],
                task_id=context.task_id,
                context_id=context.context_id,
            )
        )
        return

    await updater.complete(
        Message(
            message_id=f"a-{context.task_id}",
            role=Role.agent,
            parts=[Part(root=TextPart(
                text="It is sunny and 72°F in Tokyo."))],
            task_id=context.task_id,
            context_id=context.context_id,
        )
    )

사용자의 응답은 동일한 taskId로 message/send 요청으로 에이전트에 다시 전송됩니다.

A2A 에이전트 (ADK로 빌드됨)

에이전트가 에이전트 개발 키트 (ADK)로 빌드되고 A2A를 통해 (to_a2a() 사용) 노출되는 경우 질문 함수를 LongRunningFunctionTool로 래핑합니다.

from google.adk.agents import Agent
from google.adk.tools.long_running_tool import LongRunningFunctionTool
from google.adk.a2a.utils.agent_to_a2a import to_a2a

def get_weather(city: str) -> str:
    """Returns weather information for a city."""
    return f"It is sunny and 72°F in {city}."

def ask_clarifying_question(question: str) -> None:
    """Asks the user a question and waits for their reply."""
    return None

root_agent = Agent(
    model="gemini-2.5-flash",
    name="weather_agent",
    instruction=(
        "If the user asks for weather without specifying a city, call "
        "`ask_clarifying_question` and wait. Do not guess."
    ),
    tools=[get_weather, LongRunningFunctionTool(ask_clarifying_question)],
)

a2a_app = to_a2a(root_agent, port=8000)

Agent Engine의 ADK 에이전트

Agent Engine의 추론 엔진 호스팅 ADK 에이전트의 경우 생성기 도구 또는 커스텀 에이전트 단계 내에서 RequestInput 이벤트를 내보냅니다.

from google.adk.events import RequestInput

# Inside a generator tool or custom agent step
def ask_for_details():
    """Requests user input during tool or step execution."""
    yield RequestInput(
        interrupt_id="weather_city",
        message="Which city would you like the weather for?",
    )

다음 단계

  • 커넥터 작업 사용 방법 알아보기
  • MCP 서버 연결 방법 알아보기 connect MCP servers.
  • 흐름 제어 단계 사용 방법 알아보기