AI Agent with Short-Term Memory on Azure

Creating AI agents capable of handling complex tasks and maintaining context requires implementing short-term memory, often referred to as “scratchpad” or working memory. This allows agents to temporarily store and process information relevant to their immediate goals. Microsoft Azure offers a range of services that can be utilized to build effective short-term memory for AI agents.

Why Short-Term Memory is Crucial for AI Agents:

• Contextual Understanding: Enables agents to remember previous parts of a conversation or steps in a task.
• Task Decomposition: Helps in breaking down complex tasks into smaller steps and tracking progress.
• Tool Integration: Facilitates the management of inputs and outputs when using external tools or APIs.
• Dynamic Planning: Allows agents to adjust their actions based on immediate feedback or new information.
• Improved Reasoning: Provides a space for temporary storage of intermediate thoughts and calculations.

Azure Services for Implementing Short-Term Memory (with Conceptual Examples):

Azure provides various services suitable for building short-term memory for AI agents:

1. In-Memory Storage within Compute Services:

• Azure Virtual Machines: RAM within the VM provides direct and fast short-term memory for processes.
• Azure Functions: Variables within the function’s execution scope offer ephemeral short-term memory per invocation. Local file storage can be used for very short-term needs within an invocation.
• Azure Container Instances (ACI) / Azure Kubernetes Service (AKS): Container memory can serve as an in-memory scratchpad.

Use Case: Simple, localized state management within a single session or task execution.

import time

class InMemoryAgent:
    def __init__(self):
        self.scratchpad = {}

    def process_data(self, session_id, data_point):
        if session_id not in self.scratchpad:
            self.scratchpad[session_id] = []
        self.scratchpad[session_id].append(data_point)
        print(f"Session {session_id}'s data: {self.scratchpad[session_id]}")

# Example usage (within an Azure VM or Function execution)
agent = InMemoryAgent()
agent.process_data("session123", {"step": 1, "value": "initial"})
agent.process_data("session123", {"step": 2, "value": "intermediate"})
    

2. Managed In-Memory Data Stores:

• Azure Cache for Redis: A fully managed, high-performance in-memory data store service based on the popular open-source Redis. Offers various data structures suitable for session management, caching, and temporary data storage with low latency.

Use Case: Maintaining user session data for conversational agents, caching intermediate API responses, storing temporary task states.

import redis
import json
import time

# Configuration (replace with your actual Azure Cache for Redis configuration)
REDIS_HOST = "your-redis-instance.redis.cache.windows.net"
REDIS_PORT = 6379
REDIS_PASSWORD = "your-redis-password"

try:
    r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD, db=0)

    r.set('chat:user789', json.dumps({'last_message': 'Hello', 'timestamp': time.time()}))
    user_chat_data = json.loads(r.get('chat:user789'))
    print(f"User 789 Chat Data from Redis: {user_chat_data}")

    r.delete('chat:user789')

except redis.exceptions.ConnectionError as e:
    print(f"Error connecting to Azure Cache for Redis: {e}")
    

Remember to replace the placeholder configuration with your actual Azure Cache for Redis details.

3. NoSQL Databases with Time-to-Live (TTL):

• Azure Cosmos DB with TTL: A globally distributed, multi-model database service that offers TTL functionality. You can set a time-to-live for documents, after which they are automatically expired and deleted. Suitable for temporary data that needs persistence with automatic cleanup.

Use Case: Storing temporary authentication tokens, managing the state of short-lived workflows, caching data with an expiration policy.

import azure.cosmos.cosmos_client as cosmos_client
import azure.cosmos.exceptions as exceptions
import datetime
import json

# Configuration (replace with your actual Azure Cosmos DB configuration)
COSMOS_ENDPOINT = "your-cosmosdb-endpoint"
COSMOS_KEY = "your-cosmosdb-key"
DATABASE_NAME = "temp_data_db"
CONTAINER_NAME = "temp_items"

try:
    client = cosmos_client.CosmosClient(COSMOS_ENDPOINT, COSMOS_KEY)
    database = client.get_database_client(DATABASE_NAME)
    container = database.get_container_client(CONTAINER_NAME)

    item = {
        'id': 'temp_item_1',
        'data': {'value': 'some temporary info'},
        'ttl': 3600  # Expire in 1 hour (in seconds)
    }
    container.upsert_item(body=item)
    print(f"Temporary data stored in Cosmos DB with TTL.")

    # Note: TTL deletion happens in the background and might not be immediate.

except exceptions.CosmosHttpResponseError as e:
    print(f"Error interacting with Azure Cosmos DB: {e}")
    

Remember to replace the placeholder configuration with your actual Azure Cosmos DB details and ensure TTL is enabled on the container.

4. Leveraging Language Model Context Windows (for Azure OpenAI Service):

• Azure OpenAI Service: Provides access to powerful language models. The context window acts as short-term memory during an interaction.
• Azure Functions / Azure Container Apps: Compute environments for interacting with Azure OpenAI.
• Vector Search in Azure AI Search: For Retrieval-Augmented Generation (RAG), extending the model’s short-term memory with external knowledge.

Use Case: Building conversational AI, question answering systems, and agents that require reasoning over specific documents within a single interaction.

# Conceptual Python example using Azure OpenAI Service (requires openai library)
"""
import openai

# Configuration (replace with your actual Azure OpenAI configuration)
openai.api_type = "azure"
openai.api_base = "your-azure-openai-endpoint"
openai.api_version = "2023-05-15v1"
openai.api_key = "your-azure-openai-api-key"
deployment_name = "your-deployment-name" # The name of your deployed model

try:
    response = openai.ChatCompletion.create(
        engine=deployment_name,
        messages=[
            {"role": "system", "content": "You are a helpful AI assistant."},
            {"role": "user", "content": "What is the capital of France?"},
            {"role": "assistant", "content": "The capital of France is Paris."},
            {"role": "user", "content": "What is the population of that city?"} # Relying on the short-term context
        ]
    )
    print(response.choices[0].message.content)

except openai.error.OpenAIError as e:
    print(f"Error with Azure OpenAI Service: {e}")
"""
# To run this, uncomment the code and ensure you have the openai library installed and Azure OpenAI configured.
    

Key Aspects: Careful prompt design and managing the token limit are essential. Vector Search can enhance the context.

5. Durable Functions (Azure Functions):

• Durable Functions: An extension of Azure Functions for stateful workflows. The orchestration state acts as short-term memory for the workflow instance.

Use Case: Implementing complex, multi-step processes where the agent needs to remember the state across different function executions.

# Conceptual Python example for Durable Functions (Illustrative - requires Durable Functions setup)
"""
import azure.durable_functions as df

@df.orchestrator()
def orchestrator_function(context):
    input_data = context.get_input()
    step1_result = yield context.call_activity('process_step_1', input_data)
    step2_input = {'previous_result': step1_result, 'original_input': input_data}
    step2_result = yield context.call_activity('process_step_2', step2_input)
    return step2_result

@df.activity_trigger(input_type=str)
def process_step_1(input: str):
    print(f"Processing step 1 with input: {input}")
    return f"Result from step 1 based on {input}"

@df.activity_trigger(input_type=dict)
def process_step_2(input: dict):
    print(f"Processing step 2 with input: {input}")
    return f"Final result based on {input}"

# The orchestration context implicitly manages the short-term state of the workflow.
"""
# To run this, you would need to define these functions within an Azure Functions project with the Durable Functions extension.
    

Key Aspects: Durable Functions manage workflow state, enabling complex, stateful AI agent logic.

Live Use Cases in Summary:

• Contextual Customer Support Bot: Azure Bot Service with Azure Cache for Redis for storing conversation history.
• Personalized Product Recommendations: Azure App Service or Azure Cache for Redis tracking user browsing within a session.
• Workflow Automation Agent: Azure Durable Functions managing state across multi-step processes.
• Real-time Anomaly Detection: Azure Stream Analytics and Azure Cache for Redis maintaining recent metrics.

Choosing the Right Approach:

The selection of Azure services depends on factors such as latency requirements, data volume, persistence needs across sessions, complexity of the data structures, and the overall architecture of your AI agent on Azure.

Developers have access to a robust set of Azure services to build AI agents with effective short-term memory capabilities, enabling more interactive and intelligent applications.