워크플로 상태(State)
Microsoft Agent Framework 워크플로 - 상태(State)
이 문서는 Microsoft Agent Framework 워크플로 시스템의 **상태(State)**에 대한 개요를 제공합니다.
개요
상태는 워크플로 안의 여러 실행기(executor)가 공통 데이터에 접근하고 수정할 수 있게 해줍니다. 워크플로의 서로 다른 부분이 정보를 공유해야 하는데 직접 메시지를 주고받기 어렵거나 비효율적인 시나리오에서 이 기능은 필수적이에요.
상태 가시성과 범위 동작
::: zone pivot="programming-language-csharp"
QueueStateUpdateAsync와 ReadStateAsync는 모두 범위(scope)를 인식합니다.
scopeName이null이면 실행기의 개인 기본 범위가 사용됩니다.scopeName이 설정되면(예:"SharedResponse") 같은 범위 이름을 쓰는 어떤 실행기든 읽을 수 있는 공유 범위에 값이 기록됩니다.
가시성 타이밍은 슈퍼스텝 규칙을 따릅니다.
QueueStateUpdateAsync를 호출한 실행기는 같은 핸들러 안에서 갱신된 값을 즉시 읽을 수 있어요.- 다른 실행기는 다음 슈퍼스텝부터 그 갱신을 볼 수 있습니다.
실행기들 간에 상태를 공유하려면 쓰기와 읽기 호출에서 같은 null이 아닌 범위 이름을 사용하세요.
private const string SharedScope = "SharedResponse";
await context.QueueStateUpdateAsync("Response", blanketResponse, scopeName: SharedScope, cancellationToken);
var finalResponse = await context.ReadStateAsync<string>("Response", scopeName: SharedScope, cancellationToken);
::: zone-end
::: zone pivot="programming-language-python"
WorkflowContext.set_state()와 WorkflowContext.get_state()는 워크플로 실행 중 하위 실행기들이 사용할 수 있는 워크플로 상태를 다룹니다.
실행기들 사이에서 같은 값을 쓰고 읽으려면 일관된 키를 사용하세요.
ctx.set_state("response", blanket_response)
final_response = ctx.get_state("response")
::: zone-end
상태에 쓰기
::: zone pivot="programming-language-csharp"
using Microsoft.Agents.AI.Workflows;
internal sealed class FileReadExecutor() : Executor<string, string>("FileReadExecutor")
{
public override async ValueTask<string> HandleAsync(
string message,
IWorkflowContext context,
CancellationToken cancellationToken = default)
{
// Read file content from embedded resource
string fileContent = File.ReadAllText(message);
// Store file content in a shared state for access by other executors
string fileID = Guid.NewGuid().ToString("N");
await context.QueueStateUpdateAsync(fileID, fileContent, scopeName: "FileContent", cancellationToken);
return fileID;
}
}
::: zone-end
::: zone pivot="programming-language-python"
import uuid
from agent_framework import (
Executor,
WorkflowContext,
handler,
)
class FileReadExecutor(Executor):
@handler
async def handle(self, file_path: str, ctx: WorkflowContext[str]):
# Read file content from embedded resource
with open(file_path, 'r') as file:
file_content = file.read()
# Store file content in state for access by other executors
file_id = str(uuid.uuid4())
ctx.set_state(file_id, file_content)
await ctx.send_message(file_id)
::: zone-end
::: zone pivot="programming-language-go"
fileRead := workflow.NewExecutor("FileReadExecutor", func(ctx *workflow.Context, path string) (string, error) {
fileContent, err := os.ReadFile(path)
if err != nil {
return "", err
}
fileID := uuid.NewString()
if err := ctx.QueueStateUpdate(fileID, "FileContent", string(fileContent)); err != nil {
return "", err
}
return fileID, nil
}).Bind()
::: zone-end
상태 접근하기
::: zone pivot="programming-language-csharp"
using Microsoft.Agents.AI.Workflows;
internal sealed class WordCountingExecutor() : Executor<string, int>("WordCountingExecutor")
{
public override async ValueTask<int> HandleAsync(
string message,
IWorkflowContext context,
CancellationToken cancellationToken = default)
{
// Retrieve the file content from the shared state
var fileContent = await context.ReadStateAsync<string>(message, scopeName: "FileContent", cancellationToken)
?? throw new InvalidOperationException("File content state not found");
return fileContent.Split([' ', '\n', '\r'], StringSplitOptions.RemoveEmptyEntries).Length;
}
}
::: zone-end
::: zone pivot="programming-language-python"
from agent_framework import (
Executor,
WorkflowContext,
handler,
)
class WordCountingExecutor(Executor):
@handler
async def handle(self, file_id: str, ctx: WorkflowContext[int]):
# Retrieve the file content from state
file_content = ctx.get_state(file_id)
if file_content is None:
raise ValueError("File content state not found")
await ctx.send_message(len(file_content.split()))
::: zone-end
::: zone pivot="programming-language-go"
fileProcess := workflow.NewExecutor("FileProcessExecutor", func(ctx *workflow.Context, fileID string) (FileSummary, error) {
value, err := ctx.ReadState(fileID, "FileContent")
if err != nil {
return FileSummary{}, err
}
fileContent, ok := value.(string)
if !ok {
return FileSummary{}, fmt.Errorf("file content %q was not found", fileID)
}
return FileSummary{
FileID: fileID,
Summary: summarize(fileContent),
}, nil
}).Bind()
::: zone-end
워크플로 범위 런타임 kwargs
에이전트와 도구로 흘러가야 하지만 공유 워크플로 상태가 되지는 않아야 하는 값이라면, workflow.run()에 function_invocation_kwargs=나 client_kwargs=로 넘겨주세요.
- 최상위 키 중 아무것도 실행기 ID와 일치하지 않으면 매핑은 전역으로 취급되고, 일치하는 모든 에이전트 실행기가 같은 dict를 받습니다.
- 최상위 키 중 하나 이상이 실행기 ID와 일치하면 매핑 전체가 실행기별 타게팅으로 취급되어, 각 실행기는 자기 항목만 받아요.
function_invocation_kwargs와client_kwargs모두에 같은 전역/타게팅 규칙이 적용됩니다.
await workflow.run(
"Create the report",
function_invocation_kwargs={
"tenant": "contoso",
"request_id": "req-42",
},
)
await workflow.run(
"Create the report",
function_invocation_kwargs={
"researcher": {
"db_config": {"connection_string": "..."},
},
"writer": {
"user_preferences": {"format": "markdown"},
},
},
)
[!TIP] 실행기 타게팅 kwargs는 워크플로 실행기 ID를 사용합니다. 래핑된 에이전트의 경우 기본적으로 에이전트 이름이거나,
AgentExecutor(...)에 넘긴 명시적id가 사용됩니다.
상태 격리
실제 애플리케이션에서 여러 작업이나 요청을 처리할 때 상태를 제대로 관리하는 건 아주 중요합니다. 제대로 격리하지 않으면 서로 다른 워크플로 실행 간의 공유 상태가 예상치 못한 동작, 데이터 손상, 레이스 컨디션을 일으킬 수 있어요. 이 섹션에서는 Microsoft Agent Framework 워크플로 안에서 상태 격리를 보장하는 방법을 설명하고, 모범 사례와 흔한 함정을 다룹니다.
변경 가능한 워크플로 빌더 vs 변경 불가능한 워크플로
워크플로는 워크플로 빌더로 만들어집니다. 워크플로 빌더는 일반적으로 변경 가능한데, 빌더를 만든 뒤 또는 워크플로가 빌드된 뒤에도 시작 실행기를 추가·수정하거나 다른 구성을 바꿀 수 있기 때문입니다. 반면 워크플로는 빌드되고 나면 수정할 수 없어요(워크플로를 수정하는 공용 API가 없습니다).
이 구분이 중요한 이유는 서로 다른 워크플로 실행에서 상태가 어떻게 관리되는지에 영향을 주기 때문입니다. 단일 워크플로 인스턴스를 여러 작업이나 요청에 재사용하는 건 권장하지 않아요. 의도하지 않은 상태 공유로 이어질 수 있기 때문입니다. 대신 작업이나 요청마다 빌더에서 새 워크플로 인스턴스를 만들어, 상태 격리와 스레드 안전성을 확보하세요.
헬퍼 메서드로 상태 격리 보장하기
실행기 인스턴스를 한 번 만들어 여러 워크플로 빌드에서 공유하면, 그 내부 상태가 모든 워크플로 실행에 걸쳐 공유됩니다. 실행기가 워크플로마다 격리되어야 하는 변경 가능한 상태를 담고 있다면 문제가 될 수 있어요. 상태 격리와 스레드 안전성을 확보하려면 실행기 인스턴스 생성과 워크플로 빌드를 헬퍼 메서드 안에 감싸서, 호출할 때마다 새롭고 독립적인 인스턴스를 만들게 하세요.
::: zone pivot="programming-language-csharp"
준비 중...
::: zone-end
::: zone pivot="programming-language-python"
격리되지 않은 예시(상태 공유):
executor_a = CustomExecutorA()
executor_b = CustomExecutorB()
# executor_a and executor_b are shared across all workflows built from this builder
workflow_builder = WorkflowBuilder(start_executor=executor_a).add_edge(executor_a, executor_b)
workflow_a = workflow_builder.build()
workflow_b = workflow_builder.build()
# workflow_a and workflow_b share the same executor instances and their mutable state
격리된 예시(헬퍼 메서드):
def create_workflow() -> Workflow:
"""Create a fresh workflow with isolated state.
Each call produces independent executor instances, ensuring no state
leaks between workflow runs.
"""
executor_a = CustomExecutorA()
executor_b = CustomExecutorB()
return WorkflowBuilder(start_executor=executor_a).add_edge(executor_a, executor_b).build()
# Each workflow has its own executor instances with independent state
workflow_a = create_workflow()
workflow_b = create_workflow()
::: zone-end
::: zone pivot="programming-language-go"
격리되지 않은 예시(상태 공유):
executorA := workflow.NewExecutor("ExecutorA", func(_ *workflow.Context, input string) (string, error) {
return input, nil
}).Bind()
executorB := workflow.NewExecutor("ExecutorB", func(_ *workflow.Context, input string) (string, error) {
return input, nil
}).Bind()
builder := workflow.NewBuilder(executorA).AddEdge(executorA, executorB)
workflowA, err := builder.Build()
if err != nil {
return err
}
workflowB, err := builder.Build()
if err != nil {
return err
}
격리된 예시(헬퍼 메서드):
func createWorkflow() (*workflow.Workflow, error) {
executorA := workflow.NewExecutor("ExecutorA", func(_ *workflow.Context, input string) (string, error) {
return input, nil
}).Bind()
executorB := workflow.NewExecutor("ExecutorB", func(_ *workflow.Context, input string) (string, error) {
return input, nil
}).Bind()
return workflow.NewBuilder(executorA).AddEdge(executorA, executorB).Build()
}
workflowA, err := createWorkflow()
if err != nil {
return err
}
workflowB, err := createWorkflow()
if err != nil {
return err
}
::: zone-end
[!TIP] 상태 격리와 스레드 안전성을 제대로 확보하려면, 헬퍼 메서드 안에서 만든 실행기 인스턴스가 외부 변경 가능 상태를 공유하지 않도록 해야 합니다.
::: zone pivot="programming-language-csharp"
공유 실행기 리셋하기
워크플로 실행 간에 실행기 인스턴스를 공유해야 한다면 — 예를 들어 실행기 생성 비용이 크거나 워크플로가 에이전트로 노출되는 경우 — 상태를 가진 실행기는 IResettableExecutor를 구현해야 합니다. 이 인터페이스는 워크플로 런타임이 실행 사이에 자동으로 호출해 낡은 상태를 지우는 ResetAsync() 메서드를 제공해요.
IResettableExecutor를 언제 어떻게 구현하는지에 대한 자세한 내용은 Resettable Executors를 참고하세요.
::: zone-end
::: zone pivot="programming-language-go"
공유 실행기 리셋하기
Go 실행기 바인딩은 ResetFunc로 공유 실행기 상태를 리셋할 수 있습니다. BindNewExecutorFunc로 만든 바인딩은 워크플로 세션마다 새 실행기를 만들기 때문에 리셋 훅이 필요 없어요.
자세한 내용은 Resettable Executors를 참고하세요.
::: zone-end
에이전트 상태 관리
에이전트 컨텍스트는 에이전트 스레드를 통해 관리됩니다. 기본적으로 워크플로 안의 각 에이전트는 커스텀 실행기가 관리하지 않는 한 자신만의 스레드를 가져요. 자세한 내용은 Agents 다루기를 참고하세요.
에이전트 스레드는 워크플로 실행을 가로질러 유지됩니다. 즉 어떤 에이전트가 워크플로의 첫 실행에서 호출됐다면, 그 에이전트가 생성한 콘텐츠는 같은 워크플로 인스턴스의 이후 실행에서도 사용할 수 있어요. 단일 작업 안에서 연속성을 유지하는 데는 유용하지만, 같은 워크플로 인스턴스를 서로 다른 작업이나 요청에 재사용하면 의도하지 않은 상태 공유로 이어질 수 있습니다. 각 작업이 격리된 에이전트 상태를 갖게 하려면, 에이전트와 워크플로 생성을 헬퍼 메서드 안에 감싸서 호출할 때마다 자신의 스레드를 가진 새 에이전트 인스턴스를 만들게 하세요.
::: zone pivot="programming-language-csharp"
준비 중...
::: zone-end
::: zone pivot="programming-language-python"
격리되지 않은 예시(에이전트 상태 공유):
writer_agent = FoundryChatClient(
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
credential=AzureCliCredential(),
).as_agent(
instructions=(
"You are an excellent content writer. You create new content and edit contents based on the feedback."
),
name="writer_agent",
)
reviewer_agent = FoundryChatClient(
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
credential=AzureCliCredential(),
).as_agent(
instructions=(
"You are an excellent content reviewer. "
"Provide actionable feedback to the writer about the provided content. "
"Provide the feedback in the most concise manner possible."
),
name="reviewer_agent",
)
# writer_agent and reviewer_agent are shared across all workflows
workflow = WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()
격리된 예시(헬퍼 메서드):
def create_workflow() -> Workflow:
"""Create a fresh workflow with isolated agent state.
Each call produces new agent instances with their own threads,
ensuring no conversation history leaks between workflow runs.
"""
writer_agent = FoundryChatClient(
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
credential=AzureCliCredential(),
).as_agent(
instructions=(
"You are an excellent content writer. You create new content and edit contents based on the feedback."
),
name="writer_agent",
)
reviewer_agent = FoundryChatClient(
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
credential=AzureCliCredential(),
).as_agent(
instructions=(
"You are an excellent content reviewer. "
"Provide actionable feedback to the writer about the provided content. "
"Provide the feedback in the most concise manner possible."
),
name="reviewer_agent",
)
return WorkflowBuilder(start_executor=writer_agent).add_edge(writer_agent, reviewer_agent).build()
# Each workflow has its own agent instances and threads
workflow_a = create_workflow()
workflow_b = create_workflow()
::: zone-end
::: zone pivot="programming-language-go"
Go 에이전트 상태는 agent.Session을 통해 관리됩니다. 워크플로 안의 에이전트는 새 에이전트·워크플로·세션이 만들어지지 않는 한 자기 세션을 턴 사이에 유지해요.
session, err := writerAgent.CreateSession(ctx)
if err != nil {
return err
}
_, err = writerAgent.RunText(ctx, "first request", agent.WithSession(session)).Collect()
if err != nil {
return err
}
_, err = writerAgent.RunText(ctx, "follow-up request", agent.WithSession(session)).Collect()
if err != nil {
return err
}
agentworkflow.New로 만든 호스팅 에이전트 실행기는 agentworkflow.ResetSignal{}을 보내 새 에이전트 세션을 시작할 수도 있습니다.
::: zone-end
요약
Microsoft Agent Framework 워크플로에서 상태 격리는, 실행기와 에이전트 인스턴스 생성 그리고 워크플로 빌드를 헬퍼 메서드 안에 감싸는 것으로 효과적으로 관리할 수 있습니다. 새 워크플로가 필요할 때마다 그 헬퍼 메서드를 호출하면 각 인스턴스가 새롭고 독립적인 상태를 갖게 되어, 서로 다른 워크플로 실행 사이의 의도치 않은 상태 공유를 피할 수 있어요.