Source code for autogen_agentchat.base._task

from dataclasses import dataclass
from typing import AsyncGenerator, Protocol, Sequence

from autogen_core import CancellationToken

from ..messages import AgentEvent, ChatMessage



[docs]
@dataclass
class TaskResult:
    """Result of running a task."""

    messages: Sequence[AgentEvent | ChatMessage]
    """Messages produced by the task."""

    stop_reason: str | None = None
    """The reason the task stopped."""




[docs]
class TaskRunner(Protocol):
    """A task runner."""


[docs]
    async def run(
        self,
        *,
        task: str | ChatMessage | Sequence[ChatMessage] | None = None,
        cancellation_token: CancellationToken | None = None,
    ) -> TaskResult:
        """Run the task and return the result.

        The task can be a string, a single message, or a sequence of messages.

        The runner is stateful and a subsequent call to this method will continue
        from where the previous call left off. If the task is not specified,
        the runner will continue with the current task."""
        ...



[docs]
    def run_stream(
        self,
        *,
        task: str | ChatMessage | Sequence[ChatMessage] | None = None,
        cancellation_token: CancellationToken | None = None,
    ) -> AsyncGenerator[AgentEvent | ChatMessage | TaskResult, None]:
        """Run the task and produces a stream of messages and the final result
        :class:`TaskResult` as the last item in the stream.

        The task can be a string, a single message, or a sequence of messages.

        The runner is stateful and a subsequent call to this method will continue
        from where the previous call left off. If the task is not specified,
        the runner will continue with the current task."""
        ...