Implementation:Microsoft Autogen TaskResult State

Knowledge Sources	Microsoft AutoGen
Domains	Multi-Agent Systems, State Management, Conversation Persistence, Human-in-the-Loop
Last Updated	2026-02-11 00:00 GMT

Overview

Concrete tools for capturing swarm conversation results and persisting/restoring full team state, provided by Microsoft AutoGen.

Description

State persistence in AutoGen swarms is handled by three complementary APIs:

TaskResult is a Pydantic BaseModel returned at the end of every swarm execution (both run() and run_stream()). It contains the complete sequence of messages produced during execution and an optional stop reason string. The messages field uses SerializeAsAny to preserve polymorphic types (TextMessage, HandoffMessage, ToolCallSummaryMessage, etc.) during serialization. TaskResult is a read-only snapshot; it does not provide restoration capabilities on its own.

save_state() is an async method on BaseGroupChat (and therefore on Swarm) that captures the complete internal state of the team. It iterates over each participant and the group chat manager, calling agent_save_state() on the runtime for each. The result is a nested dictionary with an agent_states key containing each agent's state keyed by agent name. This state includes model contexts, conversation threads, turn counters, and the current speaker.

load_state() is the inverse operation: it accepts a state dictionary (in the same format as produced by save_state()) and restores the internal state of each participant and the group chat manager. It validates that the state contains entries for all current participants and raises ValueError if any are missing. It also raises RuntimeError if called while the team is running.

Together, these three APIs enable the full pause-resume lifecycle for swarm workflows.

Usage

Use TaskResult to inspect the outcome of a swarm execution, including the message history and stop reason. Use save_state() and load_state() when you need to persist a swarm's state across sessions, transfer state between environments, or implement checkpoint-based recovery in long-running workflows.

Code Reference

Source Location

Repository: Microsoft AutoGen
File (TaskResult): python/packages/autogen-agentchat/src/autogen_agentchat/base/_task.py (Lines 9-16)
File (save_state/load_state): python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py (Lines 748-834)

Signature

# TaskResult
class TaskResult(BaseModel):
    messages: Sequence[SerializeAsAny[BaseAgentEvent | BaseChatMessage]]
    stop_reason: str | None = None

# save_state (on BaseGroupChat / Swarm)
async def save_state(self) -> Mapping[str, Any]: ...

# load_state (on BaseGroupChat / Swarm)
async def load_state(self, state: Mapping[str, Any]) -> None: ...

Import

from autogen_agentchat.base import TaskResult
from autogen_agentchat.teams import Swarm

I/O Contract

Inputs

Name	Type	Required	Description
messages (TaskResult)	BaseChatMessage]	Yes	The sequence of messages produced during swarm execution. Preserves polymorphic types via SerializeAsAny.
stop_reason (TaskResult)	None	No	The reason the swarm stopped. Set by the termination condition (e.g., "Handoff to user from Alice detected.") or None if stopped by max_turns.
state (load_state)	Mapping[str, Any]	Yes	A state dictionary produced by save_state(). Must contain an "agent_states" key with entries for all current participants and the group chat manager.

Outputs

Name	Type	Description
TaskResult	TaskResult	A Pydantic model containing the message history and stop reason. Returned as the final item in run_stream() or the return value of run().
save_state result	Mapping[str, Any]	A nested dictionary containing the serialized state of all agents and the group chat manager. Structure: {"agent_states": {"agent_name": {...}, "SwarmGroupChatManager": {...}}}.
load_state result	None	Restores internal state. Raises ValueError if agent names do not match. Raises RuntimeError if team is running.

Usage Examples

Basic Example

import asyncio
import json
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import Swarm
from autogen_agentchat.conditions import HandoffTermination
from autogen_agentchat.messages import HandoffMessage
from autogen_agentchat.base import TaskResult
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent = AssistantAgent(
        "assistant",
        model_client=model_client,
        handoffs=["user"],
        system_message="Help with tasks. Ask the user if you need more info.",
    )

    termination = HandoffTermination(target="user")
    team = Swarm([agent], termination_condition=termination)

    # Step 1: Run the swarm until it hands off to user
    result = await Console(team.run_stream(task="Plan a birthday party."))

    # Inspect the TaskResult
    print(f"Stop reason: {result.stop_reason}")
    print(f"Message count: {len(result.messages)}")
    for msg in result.messages:
        print(f"  [{msg.source}] {type(msg).__name__}: {msg.content[:80]}")

    # Step 2: Save the full team state
    state = await team.save_state()

    # Optionally persist to disk
    with open("swarm_state.json", "w") as f:
        json.dump(state, f)

    # Step 3: Later, restore the state (possibly in a new session)
    with open("swarm_state.json", "r") as f:
        restored_state = json.load(f)

    await team.load_state(restored_state)

    # Step 4: Resume with user input
    result = await Console(
        team.run_stream(
            task=HandoffMessage(
                source="user",
                target="assistant",
                content="The party is for 20 people, budget is $500.",
            )
        )
    )
    print(f"Final stop reason: {result.stop_reason}")


asyncio.run(main())

Related Pages

Implements Principle

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment