에이전트 실행기

워크플로에 AI 에이전트를 추가하는 경우 워크플로 엔진이 메시지를 라우팅하고, 세션 상태를 관리하고, 출력을 처리할 수 있도록 실행기로 래핑해야 합니다. 에이전트 실행기는 이 적응을 처리하는 기본 제공 실행기입니다.

개요

에이전트 실행기는 에이전트 추상화와 워크플로 실행 모델 간의 간격을 연결합니다. 그것은:

워크플로 그래프에서 형식화된 메시지를 수신하고 기본 에이전트에 전달합니다.
실행 간에 에이전트의 세션 및 대화 상태를 관리합니다.
워크플로 실행 모드(스트리밍 또는 비 스트리밍)에 따라 동작을 조정합니다.
관찰을 위해 워크플로 호출자에게 출력 이벤트(AgentResponse 또는 AgentResponseUpdate)를 생성합니다.
그래프 내에서 계속 처리하기 위해 연결된 다운스트림 실행기에 메시지를 보냅니다.
장기 실행 워크플로에 대한 체크포인트를 지원합니다.

작동 방식

C#에서 워크플로 엔진은 워크플로에 AIAgentHostExecutor이 추가될 때마다 내부적으로 AIAgent를 만듭니다. 이 특수 실행기는 ChatProtocolExecutor를 확장하고 턴 토큰 패턴을 사용합니다.

메시지 캐싱 - 메시지가 다른 실행기에서 도착하면 에이전트 실행기에서 메시지를 수집합니다. 사용하도록 설정된 경우 ForwardIncomingMessages (기본값) 들어오는 메시지도 다운스트림 실행기에 전달됩니다.
토큰 트리거 설정 - 에이전트는 캐시된 메시지를 받은 후에만 처리합니다 TurnToken.
에이전트 호출 - 실행기가 기본 에이전트에서 RunAsync (비 스트리밍) 호출이나 RunStreamingAsync (스트리밍) 호출을 합니다.
출력 제공 - 스트리밍 이벤트가 활성화되어 있는 경우, 각 단계적 AgentResponseUpdate이 워크플로 출력으로 산출됩니다. EmitAgentResponseEvents이 사용하도록 설정되면, 집계된 AgentResponse도 워크플로의 출력으로 생성됩니다.
다운스트림 메시징 - 에이전트의 응답 메시지가 연결된 다운스트림 실행기에 전송됩니다.
턴 토큰 통과 - 턴을 완료한 후 실행기는 체인의 다음 에이전트가 처리를 시작할 수 있도록 새 TurnToken 다운스트림을 보냅니다.

팁 (조언)

일부 시나리오에는 보다 특수화된 에이전트 실행기가 필요할 수 있습니다. 예를 들어 핸드오프 오케스트레이션 은 사용자 지정 라우팅 논리와 함께 전용 HandoffAgentExecutor 을 사용합니다.

암시적 및 명시적 생성

AIAgent을 WorkflowBuilder에 전달하면, 프레임워크는 이를 자동으로 AIAgentBinding로 래핑하여 기본 AIAgentHostExecutor를 생성합니다. 에이전트 실행기를 직접 인스턴스화할 필요가 없습니다.

AIAgent writerAgent = /* create your agent */;
AIAgent reviewerAgent = /* create your agent */;

// Agents are automatically wrapped — no manual executor creation required
var workflow = new WorkflowBuilder(writerAgent)
    .AddEdge(writerAgent, reviewerAgent)
    .Build();

AgentWorkflowBuilder의 도우미 메서드를 일반적인 패턴에 사용할 수도 있습니다.

// Build a sequential pipeline of agents
var workflow = AgentWorkflowBuilder.BuildSequential(writerAgent, reviewerAgent);

사용자 지정 구성

에이전트 실행기가 동작하는 방식을 사용자 지정하려면 다음을 BindAsExecutor사용합니다AIAgentHostOptions.

var options = new AIAgentHostOptions
{
    EmitAgentUpdateEvents = true,
    EmitAgentResponseEvents = true,
    ReassignOtherAgentsAsUsers = true,
    ForwardIncomingMessages = true,
};

ExecutorBinding writerBinding = writerAgent.BindAsExecutor(options);
var workflow = new WorkflowBuilder(writerBinding)
    .AddEdge(writerBinding, reviewerAgent)
    .Build();

입력 형식

C#의 에이전트 실행기는 다음과 같은 여러 입력 형식 string을 ChatMessage허용합니다 IEnumerable<ChatMessage>. 문자열 입력은 ChatMessage 인스턴스로 자동 변환되며, 이때 User 역할을 합니다. 들어오는 모든 메시지는 수신될 때까지 TurnToken 누적되며, 이때 실행기는 일괄 처리를 처리합니다. ReassignOtherAgentsAsUsers 사용하도록 설정되면(기본값) 다른 에이전트의 메시지가 역할에 다시 할당 User 되므로 기본 모델은 해당 메시지를 사용자 입력으로 처리하고 현재 에이전트의 메시지는 역할을 유지 Assistant 합니다.

출력 및 연결

에이전트가 턴을 완료하면 실행기는 다음을 수행합니다.

에이전트의 응답 메시지를 연결된 모든 다운스트림 실행기에 보냅니다.
체인의 다음 에이전트가 처리를 시작할 수 있도록 새 TurnToken 에이전트를 전달합니다.

이렇게 하면 체인 에이전트를 간단하게 만들 수 있습니다. 단순히 에지와 연결하기만 하면 됩니다.

var workflow = new WorkflowBuilder(frenchTranslator)
    .AddEdge(frenchTranslator, spanishTranslator)
    .AddEdge(spanishTranslator, englishTranslator)
    .Build();

스트리밍 동작

스트리밍 동작은 EmitAgentUpdateEvents 옵션 또는 동적 AIAgentHostOptions를 통해 TurnToken에서 제어됩니다.

사용하도록 설정하면 실행기가 에이전트를 호출 RunStreamingAsync 하고 각 AgentResponseUpdate를 워크플로 출력 이벤트로 산출합니다. 실시간 토큰별 업데이트를 제공합니다.
사용하지 않도록 설정하면 실행기가 단일 전체 응답을 호출 RunAsync 하고 생성합니다.

// Enable streaming events at the configuration level
var options = new AIAgentHostOptions
{
    EmitAgentUpdateEvents = true,
};

// Or enable streaming dynamically via TurnToken
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

공유 세션

각 에이전트 실행기는 기본적으로 자체 세션을 유지 관리합니다. 에이전트 간에 세션을 공유하려면 워크플로에 추가하기 전에 공통 세션 공급자와 에이전트를 구성합니다.

구성 옵션

AIAgentHostOptions 는 에이전트 실행자의 동작을 제어합니다.

Option	기본값	설명
`EmitAgentUpdateEvents`	`null`	실행 중에 스트리밍 업데이트 이벤트를 내보냅니다. `TurnToken` 설정하면 우선합니다. 둘 다인 `null`경우 스트리밍을 사용할 수 없습니다.
`EmitAgentResponseEvents`	`false`	집계된 에이전트 응답을 워크플로 출력 이벤트로 내보낸다.
`InterceptUserInputRequests`	`false`	`UserInputRequestContent`을 가로채어 처리할 워크플로 메시지로 라우팅합니다.
`InterceptUnterminatedFunctionCalls`	`false`	`FunctionCallContent`를 해당 결과 없이 가로채어 워크플로 메시지로 라우팅합니다.
`ReassignOtherAgentsAsUsers`	`true`	모델이 메시지를 사용자 입력으로 처리하도록 `User` 다른 에이전트의 메시지를 역할에 다시 할당합니다.
`ForwardIncomingMessages`	`true`	에이전트가 생성된 메시지 앞에 들어오는 메시지를 다운스트림 실행기에 전달합니다.

검사점 설정

에이전트 실행자는 장기 실행 워크플로에 대해 체크포인트 기능을 지원합니다. 검사점을 만들면 실행기는 다음을 직렬화합니다.

에이전트의 세션 상태(SerializeSessionAsync을 통해)입니다.
현재 턴의 이벤트 전송 설정(요청이 보류 중이고 실행기가 아직 들어오는 TurnToken를 넘기지 않은 경우에만 사용됨).
보류 중인 모든 사용자 입력 요청 및 함수 호출 요청입니다.

복원 시 실행기는 세션 및 보류 중인 요청 상태를 역직렬화하여 워크플로가 중단된 위치에서 다시 시작할 수 있도록 합니다.

작동 방식

AgentExecutor 클래스는 SupportsAgentRun 프로토콜을 구현하는 에이전트를 래핑합니다. 실행기가 메시지를 수신하는 경우:

메시지 정규화 - 입력이 개체 목록 Message 으로 정규화되고 실행기의 내부 캐시에 추가됩니다. 실행기는 캐싱하기 전에 입력을 정규화하는 전용 처리기로 라우팅되는 여러 입력 형식strMessage(, , list[str | Message], AgentExecutorRequest및 AgentExecutorResponse )을 허용합니다.
에이전트 호출 - 실행기가 캐시된 메시지를 사용하여 호출 agent.run() 하고 워크플로 실행 모드에 따라 스트리밍 또는 비 스트리밍 모드를 자동으로 선택합니다.
출력 방출 - 스트리밍 모드에서 각각 AgentResponseUpdate 워크플로 출력 이벤트로 생성됩니다. 비 스트리밍 모드에서는 단일 AgentResponse 이 생성됩니다.
다운스트림 디스패치 - 에이전트가 완료되면 실행기는 연결된 모든 다운스트림 실행기에 AgentExecutorResponse 메시지를 보냅니다. 이 응답에는 원활한 연결이 가능한 전체 대화 기록이 포함됩니다.
캐시 재설정 - 에이전트가 호출된 후 실행기 내부 메시지 캐시가 지워지므로 각 에이전트 호출에서 마지막 호출 이후 받은 새 메시지만 처리합니다.

팁 (조언)

일부 시나리오에는 보다 특수화된 에이전트 실행기가 필요할 수 있습니다. 예를 들어 핸드오프 오케스트레이션 은 사용자 지정 라우팅 논리가 있는 전용 실행기를 사용합니다.

암시적 및 명시적 생성

에이전트를 직접 전달할 때 WorkflowBuilder는 에이전트를 AgentExecutor 인스턴스로 자동으로 래핑합니다. 대부분의 워크플로에서는 암시적 만들기로 충분합니다.

from agent_framework import WorkflowBuilder

writer_agent = client.as_agent(name="Writer", instructions="...")
reviewer_agent = client.as_agent(name="Reviewer", instructions="...")

# Agents are automatically wrapped — no manual AgentExecutor creation required
workflow = (
    WorkflowBuilder(start_executor=writer_agent)
    .add_edge(writer_agent, reviewer_agent)
    .build()
)

명시적 생성

필요할 때 AgentExecutor을(를) 명시적으로 만드십시오.

여러 에이전트 간에 세션을 공유합니다.
라우팅 및 대상 런타임 kwargs에 대한 사용자 지정 실행기 ID를 제공합니다.
여러 에지에서 동일한 실행기 인스턴스를 참조합니다.

from agent_framework import AgentExecutor

writer_executor = AgentExecutor(writer_agent, id="my-writer")
reviewer_executor = AgentExecutor(reviewer_agent, id="my-reviewer")

workflow = (
    WorkflowBuilder(start_executor=writer_executor)
    .add_edge(writer_executor, reviewer_executor)
    .build()
)

생성자 매개 변수:

매개 변수	유형	설명
`agent`	`SupportsAgentRun`	래핑할 에이전트입니다.
`session`	`AgentSession \\| None`	에이전트 실행에 사용할 세션입니다. 이 경우 `None`에이전트에서 새 세션이 만들어집니다.
`id`	`str \\| None`	고유 실행기 ID입니다. 사용 가능한 경우 에이전트의 이름으로 기본값이 지정됩니다.

팁 (조언)

실행기 ID는 workflow.run(function_invocation_kwargs=...) 또는 client_kwargs=을 개별 에이전트를 대상으로 지정할 때 사용되는 키이기도 합니다. 생략 id하면 워크플로는 래핑된 에이전트의 이름을 사용합니다.

입력 형식

각 AgentExecutor 기 다른 입력 형식을 허용하는 여러 처리기 메서드를 정의합니다. 워크플로 엔진은 메시지 유형에 따라 올바른 처리기를 자동으로 디스패치합니다. 모든 입력 형식은 에이전트가 실행되는지 아니면 단순히 메시지를 캐시하는지를 플래그가 제어하는 위치를 AgentExecutorRequest 제외하고 should_respond 에이전트가 즉시 실행되도록 트리거합니다.

입력 형식	처리기	트리거 에이전트	설명
`AgentExecutorRequest`	`run`	Conditional	정식 입력 형식입니다. 메시지 목록과 `should_respond` 에이전트 실행 여부를 제어하는 플래그를 포함합니다.
`str`	`from_str`	늘	원시 문자열 프롬프트를 수락합니다.
`Message`	`from_message`	늘	단일 `Message` 개체를 허용합니다.
`list[str \\| Message]`	`from_messages`	늘	문자열 또는 `Message` 개체 목록을 대화 컨텍스트로 허용합니다.
`AgentExecutorResponse`	`from_response`	늘	이전 에이전트 실행자의 응답을 수락하여 직접 연결을 사용하도록 설정합니다.

AgentExecutorRequest 사용

AgentExecutorRequest 는 정식 입력 형식이며 가장 많은 컨트롤을 제공합니다.

from agent_framework import AgentExecutorRequest, Message

# Create a request with messages
request = AgentExecutorRequest(
    messages=[Message(role="user", contents=["Hello, world!"])],
    should_respond=True,
)

# Run the workflow
result = await workflow.run(request)

플래그는 should_respond 에이전트가 메시지를 즉시 처리할지 아니면 나중에 메시지를 캐시할지 여부를 제어합니다.

True (기본값) - 에이전트가 실행되고 응답을 생성합니다.
False - 메시지가 캐시에 추가되지만 에이전트가 실행되지 않습니다. 이는 응답을 트리거하기 전에 대화 컨텍스트를 미리 로드하는 데 유용합니다.

출력 및 연결

에이전트가 완료되면 실행자가 AgentExecutorResponse를 하류로 전송합니다. 이 데이터 클래스에는 다음이 포함됩니다.

분야	유형	설명
`executor_id`	`str`	응답을 생성한 실행기의 ID입니다.
`agent_response`	`AgentResponse`	기본 에이전트 응답(클라이언트에서 변경되지 않음)입니다.
`full_conversation`	`list[Message] \\| None`	체인에 대한 전체 대화 컨텍스트(이전 입력 + 에이전트 출력)입니다.

에이전트 실행기를 연결할 때 다운스트림 실행기는 AgentExecutorResponse를 처리기 from_response를 통해 수신합니다. 이 full_conversation 필드를 사용하여 전체 대화 기록을 유지하여 다운스트림 에이전트가 이전 컨텍스트를 잃지 않도록 합니다.

spam_detector = AgentExecutor(create_spam_detector_agent())
email_assistant = AgentExecutor(create_email_assistant_agent())

# The email_assistant receives the spam_detector's full conversation context
workflow = (
    WorkflowBuilder(start_executor=spam_detector)
    .add_edge(spam_detector, email_assistant)
    .build()
)

스트리밍 동작

워크플로 AgentExecutor 실행 모드에 자동으로 적용됩니다.

stream=True는 agent.run(stream=True)을 호출하고, 각 AgentResponseUpdate을 워크플로 출력 이벤트로 제공합니다. 스트리밍이 완료되면 업데이트가 다운스트림 디스패치를 위해 전체 AgentResponse 로 집계됩니다.
stream=False (기본값) — agent.run(stream=False)을 호출하고 단일 AgentResponse를 워크플로 출력 이벤트로 생성합니다.

# Streaming mode — receive incremental updates
events = workflow.run("Write a story about a cat.", stream=True)
async for event in events:
    if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
        print(event.data.text, end="", flush=True)

# Non-streaming mode — receive complete response
result = await workflow.run("Write a story about a cat.")

# Retrieve AgentResponse objects from the result
outputs = result.get_outputs()
for output in outputs:
    if isinstance(output, AgentResponse):
        print(output.text)

공유 세션

기본적으로 각 AgentExecutor 세션은 자체 세션을 만듭니다. 여러 에이전트 간에 세션을 공유하려면(예: 공통 대화 스레드를 유지 관리하기 위해) 세션을 명시적으로 만들고 각 실행자에 전달합니다.

from agent_framework import AgentExecutor

# Create a shared session from one agent
shared_session = writer_agent.create_session()

# Both executors share the same session
writer_executor = AgentExecutor(writer_agent, session=shared_session)
reviewer_executor = AgentExecutor(reviewer_agent, session=shared_session)

메모

모든 에이전트가 공유 세션을 지원하지는 않습니다. 일반적으로 동일한 공급자 유형의 에이전트만 세션을 공유할 수 있습니다.

검사점 설정

AgentExecutor는 장기 실행 작업흐름에서 상태를 저장하고 복원하기 위한 검사점을 지정하는 기능을 지원합니다. 검사점을 만들면 실행기는 다음을 직렬화합니다.

내부 메시지 캐시입니다.
전체 대화 기록입니다.
에이전트 세션 상태.
보류 중인 모든 사용자 입력 요청 및 응답입니다.

복원 시 실행기는 이 상태를 역직렬화하여 워크플로가 중단된 위치에서 다시 시작할 수 있도록 합니다.

경고

서버 쪽 세션(예: FoundryAgent)을 사용하는 에이전트를 사용한 검사점 지정에는 제한 사항이 있습니다. 서버 쪽 세션 상태는 검사점에서 캡처되지 않으며 후속 실행으로 수정할 수 있습니다. 서버 쪽 세션을 사용하여 신뢰할 수 있는 검사점이 필요한 경우 사용자 지정 실행기를 구현하는 것이 좋습니다.

다음 단계

워크플로의 에이전트

피드백

이 페이지가 도움이 되었나요?

Last updated on 2026-04-02

다음을 통해 공유

에이전트 실행기

개요

작동 방식

암시적 및 명시적 생성

사용자 지정 구성

입력 형식

출력 및 연결

스트리밍 동작

공유 세션

구성 옵션

검사점 설정

작동 방식

암시적 및 명시적 생성

명시적 생성

입력 형식

AgentExecutorRequest 사용

출력 및 연결

스트리밍 동작

공유 세션

검사점 설정

다음 단계

피드백

추가 리소스