mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
TTL
This commit is contained in:
@@ -7,8 +7,10 @@ from ._client import DurableAIAgentClient
|
||||
from ._constants import (
|
||||
DEFAULT_MAX_POLL_RETRIES,
|
||||
DEFAULT_POLL_INTERVAL_SECONDS,
|
||||
DEFAULT_TIME_TO_LIVE_DAYS,
|
||||
MIMETYPE_APPLICATION_JSON,
|
||||
MIMETYPE_TEXT_PLAIN,
|
||||
MINIMUM_TTL_SIGNAL_DELAY_MINUTES,
|
||||
REQUEST_RESPONSE_FORMAT_JSON,
|
||||
REQUEST_RESPONSE_FORMAT_TEXT,
|
||||
THREAD_ID_FIELD,
|
||||
@@ -52,8 +54,10 @@ from ._worker import DurableAIAgentWorker
|
||||
__all__ = [
|
||||
"DEFAULT_MAX_POLL_RETRIES",
|
||||
"DEFAULT_POLL_INTERVAL_SECONDS",
|
||||
"DEFAULT_TIME_TO_LIVE_DAYS",
|
||||
"MIMETYPE_APPLICATION_JSON",
|
||||
"MIMETYPE_TEXT_PLAIN",
|
||||
"MINIMUM_TTL_SIGNAL_DELAY_MINUTES",
|
||||
"REQUEST_RESPONSE_FORMAT_JSON",
|
||||
"REQUEST_RESPONSE_FORMAT_TEXT",
|
||||
"THREAD_ID_FIELD",
|
||||
|
||||
@@ -28,6 +28,10 @@ WAIT_FOR_RESPONSE_HEADER: str = "x-ms-wait-for-response"
|
||||
DEFAULT_MAX_POLL_RETRIES: int = 30
|
||||
DEFAULT_POLL_INTERVAL_SECONDS: float = 1.0
|
||||
|
||||
# TTL configuration
|
||||
DEFAULT_TIME_TO_LIVE_DAYS: int = 14
|
||||
MINIMUM_TTL_SIGNAL_DELAY_MINUTES: int = 5
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# JSON Field Name Constants for Durable Agent State Serialization
|
||||
@@ -98,6 +102,9 @@ class DurableStateFields:
|
||||
# History field
|
||||
CONVERSATION_HISTORY: Final[str] = "conversationHistory"
|
||||
|
||||
# TTL field
|
||||
EXPIRATION_TIME_UTC: Final[str] = "expirationTimeUtc"
|
||||
|
||||
|
||||
class ContentTypes:
|
||||
"""Content type discriminator values for the $type field.
|
||||
|
||||
@@ -332,24 +332,29 @@ class DurableAgentStateData:
|
||||
Attributes:
|
||||
conversation_history: Ordered list of conversation entries (requests and responses)
|
||||
extension_data: Optional dictionary for custom metadata (not part of core schema)
|
||||
expiration_time_utc: Optional UTC datetime when the entity should expire (TTL feature)
|
||||
"""
|
||||
|
||||
conversation_history: list[DurableAgentStateEntry]
|
||||
extension_data: dict[str, Any] | None
|
||||
expiration_time_utc: datetime | None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
conversation_history: list[DurableAgentStateEntry] | None = None,
|
||||
extension_data: dict[str, Any] | None = None,
|
||||
expiration_time_utc: datetime | None = None,
|
||||
) -> None:
|
||||
"""Initialize the data container.
|
||||
|
||||
Args:
|
||||
conversation_history: Initial conversation history (defaults to empty list)
|
||||
extension_data: Optional custom metadata
|
||||
expiration_time_utc: Optional UTC datetime when the entity should expire
|
||||
"""
|
||||
self.conversation_history = conversation_history or []
|
||||
self.extension_data = extension_data
|
||||
self.expiration_time_utc = expiration_time_utc
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
result: dict[str, Any] = {
|
||||
@@ -357,13 +362,19 @@ class DurableAgentStateData:
|
||||
}
|
||||
if self.extension_data is not None:
|
||||
result[DurableStateFields.EXTENSION_DATA] = self.extension_data
|
||||
if self.expiration_time_utc is not None:
|
||||
result[DurableStateFields.EXPIRATION_TIME_UTC] = self.expiration_time_utc.isoformat()
|
||||
return result
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data_dict: dict[str, Any]) -> DurableAgentStateData:
|
||||
expiration_time_raw = data_dict.get(DurableStateFields.EXPIRATION_TIME_UTC)
|
||||
expiration_time_utc = _parse_created_at(expiration_time_raw) if expiration_time_raw else None
|
||||
|
||||
return cls(
|
||||
conversation_history=_parse_history_entries(data_dict),
|
||||
extension_data=data_dict.get(DurableStateFields.EXTENSION_DATA),
|
||||
expiration_time_utc=expiration_time_utc,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
from collections.abc import AsyncIterable
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any, cast
|
||||
|
||||
from agent_framework import (
|
||||
@@ -20,6 +21,7 @@ from agent_framework import (
|
||||
from durabletask.entities import DurableEntity
|
||||
|
||||
from ._callbacks import AgentCallbackContext, AgentResponseCallbackProtocol
|
||||
from ._constants import MINIMUM_TTL_SIGNAL_DELAY_MINUTES
|
||||
from ._durable_agent_state import (
|
||||
DurableAgentState,
|
||||
DurableAgentStateEntry,
|
||||
@@ -95,12 +97,20 @@ class AgentEntity:
|
||||
callback: AgentResponseCallbackProtocol | None = None,
|
||||
*,
|
||||
state_provider: AgentEntityStateProviderMixin,
|
||||
time_to_live: timedelta | None = None,
|
||||
minimum_ttl_signal_delay: timedelta = timedelta(minutes=MINIMUM_TTL_SIGNAL_DELAY_MINUTES),
|
||||
) -> None:
|
||||
self.agent = agent
|
||||
self.callback = callback
|
||||
self._state_provider = state_provider
|
||||
self._time_to_live = time_to_live
|
||||
self._minimum_ttl_signal_delay = minimum_ttl_signal_delay
|
||||
|
||||
logger.debug("[AgentEntity] Initialized with agent type: %s", type(agent).__name__)
|
||||
logger.debug(
|
||||
"[AgentEntity] Initialized with agent type: %s, TTL: %s",
|
||||
type(agent).__name__,
|
||||
time_to_live,
|
||||
)
|
||||
|
||||
@property
|
||||
def state(self) -> DurableAgentState:
|
||||
@@ -170,6 +180,10 @@ class AgentEntity:
|
||||
|
||||
state_response = DurableAgentStateResponse.from_run_response(correlation_id, agent_run_response)
|
||||
self.state.data.conversation_history.append(state_response)
|
||||
|
||||
# Update TTL expiration time
|
||||
self._update_ttl_expiration()
|
||||
|
||||
self.persist_state()
|
||||
|
||||
return agent_run_response
|
||||
@@ -327,6 +341,113 @@ class AgentEntity:
|
||||
request_message=request_message,
|
||||
)
|
||||
|
||||
# TTL (Time-To-Live) Methods
|
||||
def _update_ttl_expiration(self) -> None:
|
||||
"""Update the TTL expiration time after an interaction.
|
||||
|
||||
If TTL is configured:
|
||||
- Sets/updates the expiration time to now + TTL
|
||||
- On first interaction, schedules a deletion check signal
|
||||
|
||||
If TTL is disabled (None):
|
||||
- Clears any previously set expiration time
|
||||
"""
|
||||
if self._time_to_live is not None:
|
||||
current_time = datetime.now(timezone.utc)
|
||||
new_expiration_time = current_time + self._time_to_live
|
||||
is_first_interaction = self.state.data.expiration_time_utc is None
|
||||
|
||||
self.state.data.expiration_time_utc = new_expiration_time
|
||||
|
||||
logger.debug(
|
||||
"[AgentEntity] TTL expiration time updated to %s (first_interaction: %s)",
|
||||
new_expiration_time.isoformat(),
|
||||
is_first_interaction,
|
||||
)
|
||||
|
||||
# Only schedule deletion check on the first interaction when entity is created.
|
||||
# On subsequent interactions, we just update the expiration time. The scheduled
|
||||
# check_and_delete_if_expired will reschedule itself if the entity hasn't expired.
|
||||
if is_first_interaction:
|
||||
self._schedule_deletion_check()
|
||||
else:
|
||||
# TTL is disabled. Clear the expiration time if it was previously set.
|
||||
if self.state.data.expiration_time_utc is not None:
|
||||
logger.debug("[AgentEntity] TTL disabled, clearing expiration time")
|
||||
self.state.data.expiration_time_utc = None
|
||||
|
||||
def check_and_delete_if_expired(self) -> None:
|
||||
"""Check if the entity has expired and delete it if so, otherwise reschedule.
|
||||
|
||||
This method is called by a scheduled signal to check TTL expiration.
|
||||
If the entity has expired (current time >= expiration time), the state is deleted.
|
||||
If not expired, a new deletion check is scheduled for the updated expiration time.
|
||||
"""
|
||||
current_time = datetime.now(timezone.utc)
|
||||
expiration_time = self.state.data.expiration_time_utc
|
||||
|
||||
logger.debug(
|
||||
"[AgentEntity] TTL deletion check: expiration=%s, current=%s",
|
||||
expiration_time.isoformat() if expiration_time else "None",
|
||||
current_time.isoformat(),
|
||||
)
|
||||
|
||||
if expiration_time is not None:
|
||||
if current_time >= expiration_time:
|
||||
# Entity has expired, delete it by resetting state
|
||||
logger.info(
|
||||
"[AgentEntity] Entity expired at %s, deleting state",
|
||||
expiration_time.isoformat(),
|
||||
)
|
||||
self._delete_entity_state()
|
||||
else:
|
||||
# Entity hasn't expired yet, reschedule the deletion check
|
||||
if self._time_to_live is not None:
|
||||
logger.debug("[AgentEntity] Entity not expired, rescheduling deletion check")
|
||||
self._schedule_deletion_check()
|
||||
|
||||
def _schedule_deletion_check(self) -> None:
|
||||
"""Schedule a signal to self to check for expiration.
|
||||
|
||||
Note: The Python durabletask SDK does not currently support scheduled signals.
|
||||
This method logs a warning and skips scheduling. The expiration time is still
|
||||
tracked in state for potential external cleanup or future SDK support.
|
||||
|
||||
See: https://github.com/Azure/azure-functions-durable-extension/issues/1554
|
||||
"""
|
||||
if not isinstance(self._state_provider, DurableTaskEntityStateProvider):
|
||||
logger.warning(
|
||||
"[AgentEntity] Cannot schedule deletion check: state provider does not support entity signaling"
|
||||
)
|
||||
return
|
||||
|
||||
current_time = datetime.now(timezone.utc)
|
||||
expiration_time = self.state.data.expiration_time_utc
|
||||
|
||||
if expiration_time is None and self._time_to_live is not None:
|
||||
expiration_time = current_time + self._time_to_live
|
||||
|
||||
if expiration_time is None:
|
||||
return
|
||||
|
||||
# To avoid excessive scheduling, we schedule the deletion check for no less than the minimum delay
|
||||
minimum_scheduled_time = current_time + self._minimum_ttl_signal_delay
|
||||
scheduled_time = max(expiration_time, minimum_scheduled_time)
|
||||
|
||||
logger.warning(
|
||||
"[AgentEntity] TTL scheduled deletion is not yet supported in the Python durabletask SDK. "
|
||||
"Expiration time set to %s but automatic deletion will not occur. "
|
||||
"Consider implementing external cleanup based on expiration_time_utc in entity state.",
|
||||
scheduled_time.isoformat(),
|
||||
)
|
||||
|
||||
def _delete_entity_state(self) -> None:
|
||||
"""Delete the entity state to remove the entity."""
|
||||
# Reset the state cache to clear all data
|
||||
self._state_provider._state_cache = None
|
||||
# Set empty state to trigger deletion
|
||||
self._state_provider._set_state_dict({})
|
||||
|
||||
|
||||
class DurableTaskEntityStateProvider(DurableEntity, AgentEntityStateProviderMixin):
|
||||
"""DurableTask Durable Entity state provider for AgentEntity.
|
||||
|
||||
@@ -9,12 +9,14 @@ and enables registration of agents as durable entities.
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from datetime import timedelta
|
||||
from typing import Any
|
||||
|
||||
from agent_framework import AgentProtocol, get_logger
|
||||
from durabletask.worker import TaskHubGrpcWorker
|
||||
|
||||
from ._callbacks import AgentResponseCallbackProtocol
|
||||
from ._constants import DEFAULT_TIME_TO_LIVE_DAYS, MINIMUM_TTL_SIGNAL_DELAY_MINUTES
|
||||
from ._entities import AgentEntity, DurableTaskEntityStateProvider
|
||||
|
||||
logger = get_logger("agent_framework.durabletask.worker")
|
||||
@@ -51,22 +53,40 @@ class DurableAIAgentWorker:
|
||||
self,
|
||||
worker: TaskHubGrpcWorker,
|
||||
callback: AgentResponseCallbackProtocol | None = None,
|
||||
default_time_to_live: timedelta | None = timedelta(days=DEFAULT_TIME_TO_LIVE_DAYS),
|
||||
minimum_ttl_signal_delay: timedelta = timedelta(minutes=MINIMUM_TTL_SIGNAL_DELAY_MINUTES),
|
||||
):
|
||||
"""Initialize the worker wrapper.
|
||||
|
||||
Args:
|
||||
worker: The durabletask worker instance to wrap
|
||||
callback: Optional callback for agent response notifications
|
||||
default_time_to_live: Default TTL for agent entities. If an agent entity is idle
|
||||
for this duration, it will be automatically deleted. Defaults to 14 days.
|
||||
Set to None to disable TTL for agents without explicit TTL configuration.
|
||||
minimum_ttl_signal_delay: Minimum delay for scheduling TTL deletion signals.
|
||||
Defaults to 5 minutes. This prevents excessive scheduling overhead.
|
||||
|
||||
Raises:
|
||||
ValueError: If minimum_ttl_signal_delay exceeds 5 minutes.
|
||||
"""
|
||||
max_delay = timedelta(minutes=5)
|
||||
if minimum_ttl_signal_delay > max_delay:
|
||||
raise ValueError(f"minimum_ttl_signal_delay cannot exceed {max_delay}. Got: {minimum_ttl_signal_delay}")
|
||||
|
||||
self._worker = worker
|
||||
self._callback = callback
|
||||
self._default_time_to_live = default_time_to_live
|
||||
self._minimum_ttl_signal_delay = minimum_ttl_signal_delay
|
||||
self._registered_agents: dict[str, AgentProtocol] = {}
|
||||
self._agent_time_to_live: dict[str, timedelta | None] = {}
|
||||
logger.debug("[DurableAIAgentWorker] Initialized with worker type: %s", type(worker).__name__)
|
||||
|
||||
def add_agent(
|
||||
self,
|
||||
agent: AgentProtocol,
|
||||
callback: AgentResponseCallbackProtocol | None = None,
|
||||
time_to_live: timedelta | None = None,
|
||||
) -> None:
|
||||
"""Register an agent with the worker.
|
||||
|
||||
@@ -77,6 +97,10 @@ class DurableAIAgentWorker:
|
||||
Args:
|
||||
agent: The agent to register (must have a name)
|
||||
callback: Optional callback for this specific agent (overrides worker-level callback)
|
||||
time_to_live: Optional TTL for this agent's entities. If an entity is idle for
|
||||
this duration, it will be automatically deleted. If not specified, uses
|
||||
the worker's default_time_to_live. Pass a sentinel value to explicitly
|
||||
disable TTL for this agent.
|
||||
|
||||
Raises:
|
||||
ValueError: If the agent doesn't have a name or is already registered
|
||||
@@ -90,23 +114,29 @@ class DurableAIAgentWorker:
|
||||
|
||||
logger.info("[DurableAIAgentWorker] Registering agent: %s as entity: dafx-%s", agent_name, agent_name)
|
||||
|
||||
# Store the agent reference
|
||||
# Store the agent reference and TTL configuration
|
||||
self._registered_agents[agent_name] = agent
|
||||
if time_to_live is not None:
|
||||
self._agent_time_to_live[agent_name] = time_to_live
|
||||
|
||||
# Use agent-specific callback if provided, otherwise use worker-level callback
|
||||
effective_callback = callback or self._callback
|
||||
|
||||
# Get the effective TTL for this agent
|
||||
effective_ttl = self.get_time_to_live(agent_name)
|
||||
|
||||
# Create a configured entity class using the factory
|
||||
entity_class = self.__create_agent_entity(agent, effective_callback)
|
||||
entity_class = self.__create_agent_entity(agent, effective_callback, effective_ttl)
|
||||
|
||||
# Register the entity class with the worker
|
||||
# The worker.add_entity method takes a class
|
||||
entity_registered: str = self._worker.add_entity(entity_class) # pyright: ignore[reportUnknownMemberType]
|
||||
|
||||
logger.debug(
|
||||
"[DurableAIAgentWorker] Successfully registered entity class %s for agent: %s",
|
||||
"[DurableAIAgentWorker] Successfully registered entity class %s for agent: %s (TTL: %s)",
|
||||
entity_registered,
|
||||
agent_name,
|
||||
effective_ttl,
|
||||
)
|
||||
|
||||
def start(self) -> None:
|
||||
@@ -137,10 +167,30 @@ class DurableAIAgentWorker:
|
||||
"""
|
||||
return list(self._registered_agents.keys())
|
||||
|
||||
def get_time_to_live(self, agent_name: str) -> timedelta | None:
|
||||
"""Get the TTL for a specific agent.
|
||||
|
||||
Args:
|
||||
agent_name: The name of the agent
|
||||
|
||||
Returns:
|
||||
The TTL for the agent, or the default TTL if not specified.
|
||||
Returns None if TTL is disabled for this agent.
|
||||
"""
|
||||
if agent_name in self._agent_time_to_live:
|
||||
return self._agent_time_to_live[agent_name]
|
||||
return self._default_time_to_live
|
||||
|
||||
@property
|
||||
def minimum_ttl_signal_delay(self) -> timedelta:
|
||||
"""Get the minimum delay for TTL deletion signals."""
|
||||
return self._minimum_ttl_signal_delay
|
||||
|
||||
def __create_agent_entity(
|
||||
self,
|
||||
agent: AgentProtocol,
|
||||
callback: AgentResponseCallbackProtocol | None = None,
|
||||
time_to_live: timedelta | None = None,
|
||||
) -> type[DurableTaskEntityStateProvider]:
|
||||
"""Factory function to create a DurableEntity class configured with an agent.
|
||||
|
||||
@@ -150,12 +200,14 @@ class DurableAIAgentWorker:
|
||||
Args:
|
||||
agent: The agent instance to wrap
|
||||
callback: Optional callback for agent responses
|
||||
time_to_live: Optional TTL for this agent's entities
|
||||
|
||||
Returns:
|
||||
A new DurableEntity subclass configured for this agent
|
||||
"""
|
||||
agent_name = agent.name or type(agent).__name__
|
||||
entity_name = f"dafx-{agent_name}"
|
||||
minimum_signal_delay = self._minimum_ttl_signal_delay
|
||||
|
||||
class ConfiguredAgentEntity(DurableTaskEntityStateProvider):
|
||||
"""Durable entity configured with a specific agent instance."""
|
||||
@@ -167,11 +219,14 @@ class DurableAIAgentWorker:
|
||||
agent=agent,
|
||||
callback=callback,
|
||||
state_provider=self,
|
||||
time_to_live=time_to_live,
|
||||
minimum_ttl_signal_delay=minimum_signal_delay,
|
||||
)
|
||||
logger.debug(
|
||||
"[ConfiguredAgentEntity] Initialized entity for agent: %s (entity name: %s)",
|
||||
"[ConfiguredAgentEntity] Initialized entity for agent: %s (entity name: %s, TTL: %s)",
|
||||
agent_name,
|
||||
entity_name,
|
||||
time_to_live,
|
||||
)
|
||||
|
||||
def run(self, request: Any) -> Any:
|
||||
@@ -210,6 +265,16 @@ class DurableAIAgentWorker:
|
||||
logger.debug("[ConfiguredAgentEntity.reset] Resetting agent: %s", agent_name)
|
||||
self._agent_entity.reset()
|
||||
|
||||
def check_and_delete_if_expired(self) -> None:
|
||||
"""Check if the entity has expired and delete it if so.
|
||||
|
||||
This method is called by a scheduled signal to check TTL expiration.
|
||||
"""
|
||||
logger.debug(
|
||||
"[ConfiguredAgentEntity.check_and_delete_if_expired] Checking expiration for agent: %s", agent_name
|
||||
)
|
||||
self._agent_entity.check_and_delete_if_expired()
|
||||
|
||||
# Set the entity name to match the prefixed agent name
|
||||
# This is used by durabletask to register the entity
|
||||
ConfiguredAgentEntity.__name__ = entity_name
|
||||
|
||||
@@ -3,8 +3,8 @@
|
||||
This worker registers agents as durable entities and continuously listens for requests.
|
||||
The worker should run as a background service, processing incoming agent requests.
|
||||
|
||||
Prerequisites:
|
||||
- Set AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_CHAT_DEPLOYMENT_NAME
|
||||
Prerequisites:
|
||||
- Set AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_CHAT_DEPLOYMENT_NAME
|
||||
(plus AZURE_OPENAI_API_KEY or Azure CLI authentication)
|
||||
- Start a Durable Task Scheduler (e.g., using Docker)
|
||||
"""
|
||||
@@ -12,6 +12,7 @@ Prerequisites:
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
from datetime import timedelta
|
||||
|
||||
from agent_framework.azure import AzureOpenAIChatClient
|
||||
from agent_framework_durabletask import DurableAIAgentWorker
|
||||
@@ -69,24 +70,26 @@ def get_worker(
|
||||
|
||||
def setup_worker(worker: DurableTaskSchedulerWorker) -> DurableAIAgentWorker:
|
||||
"""Set up the worker with agents registered.
|
||||
|
||||
|
||||
Args:
|
||||
worker: The DurableTaskSchedulerWorker instance
|
||||
|
||||
|
||||
Returns:
|
||||
DurableAIAgentWorker with agents registered
|
||||
"""
|
||||
# Wrap it with the agent worker
|
||||
agent_worker = DurableAIAgentWorker(worker)
|
||||
|
||||
# Create and register the Joker agent
|
||||
|
||||
# Create and register the Joker agent with a 1-hour TTL
|
||||
# Agent sessions will be automatically deleted after 1 hour of inactivity
|
||||
logger.debug("Creating and registering Joker agent...")
|
||||
joker_agent = create_joker_agent()
|
||||
agent_worker.add_agent(joker_agent)
|
||||
|
||||
agent_worker.add_agent(joker_agent, time_to_live=timedelta(seconds=30))
|
||||
|
||||
logger.debug(f"✓ Registered agent: {joker_agent.name}")
|
||||
logger.debug(f" Entity name: dafx-{joker_agent.name}")
|
||||
|
||||
logger.debug(f" TTL: 1 hour")
|
||||
|
||||
return agent_worker
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user