Event-driven Agentic AI

Table of Contents

1. Foundations of event driven agent systems

• 1.1 From prompts to event-driven agents
• 1.2 Common event sources in production
• 1.3 Essential Agentics Use Cases

2. Architectural patterns for event driven agents

• 2.1 Event Buses and Pub/Sub topologies
• 2.2 State consistency and concurrency
• 2.3 Avoiding common pitfalls
• 2.4 Latency, throughput and backpressure

3. Frameworks and tooling for event driven agents

• 3.1 Messaging and streaming
• 3.2 Sustainable orchestration and workflows
• 3.3 Agent stacks and integrations
• 3.4 Cloud triggers and event routers

4. Design real time agent workflows

• 4.1 Design dagents event-native
• 4.2 Inter-agent protocols
• 4.3 Human-in-the-loop and SLAs

5. Making event driven agents robust

• 5.1 Failure Taxonomy
• 5.2 Well-done Retries
• 5.3 Idempotence strategies

6. Observability, monitoring and security

• 6.1 Traces, metrics and logs
• 6.2 Identity and access
• 6.3 Data security and privacy
• 6.4 Policies and guardrails

1. Foundations of event driven agent systems

1.1 From prompts to event-driven agents

Introduction

Generative AI tools wait for prompts before doing anything. Production systems don’t work like that — they talk in events. A purchase, a breakdown, an approval, a sudden rise of a sensor: these are all events that trigger immediate actions. Event-driven agents react to these signals in real time rather than waiting for explicit human instruction.

Fundamental Terminology

The five most fundamental concepts of an event-driven architecture are: events, commands, facts, streams, and sagas.

Event

An event is a record of something that has already happened, for example: payment declined. It is immutable — it cannot be changed once the action has taken place. We can’t go back in time. The events tell the story.

Command

An order is an instruction to do something, for example: refund order. Where events tell the story, commands write it. A command is an intention, an event is the observed result.

Fact

A fact is an event that we are willing to keep indefinitely as part of the truth of the system. Example: order 123 shipped at 10:42 UTC. Facts power auditability, training data, and agent memory.

Stream

A stream is a continuous sequence of events, ordered and append-only (you can only add, never modify). Like a physical stream, it continues to flow indefinitely — but instead of water, data flows.

Agents subscribe to streams, which can represent payments, tickets, IoT sensor readings, etc. They react to each new event that happens. Streams enable low latency processing and can be replayed for debugging or retraining.

Saga

A saga coordinates multi-step workflows across multiple services. If a step fails, the saga triggers compensatory actions. Example: A shipment is canceled if the payment is reversed. In agentic systems, sagas enable long and robust workflows.

Agentic event flow example

Here’s how these concepts fit together in a typical payment flow:

sequenceDiagram
    participant P as Producteur (Paiement)
    participant B as Event Bus / Stream
    participant A as Agent IA
    participant DB as Base de données
    participant H as Humain (si besoin)

    P->>B: Event: PaymentDeclined
    B->>A: Trigger agent
    A->>DB: Lire contexte client (fact store)
    A->>A: Raisonner (LLM)
    A->>B: Command: RefundOrder
    B->>A: Event: OrderRefunded (fact)
    A->>H: Escalade si nécessaire (human-in-the-loop)

Analogy: Events say what happened. The commands say what should happen. Facts are preserved truths. Streams are the data highways. Sagas are the conductors of complex workflows.

1.2 Common event sources in production

Agents live in a city of signals from a wide variety of sources. Here are the main sources of events encountered in production and when to use each.

Webhooks

A webhook is analogous to a doorbell that a web application calls. An event occurs, and it calls your endpoint with the details. Used when a payment is validated, a ticket is created, etc. It’s immediate and excellent for responding quickly.

Best practices for webhooks:

1. Record all incoming webhooks in the log with their delivery status — like a camera attached to your doorbell.
2. Allow only authorized clients to send webhooks to your system — like living in a secure neighborhood. Verify the signature of the webhook before it reaches your system.
3. Rate limit the number of webhooks so that your system does not receive more than it can handle.

Polling vs. Push

Rule: Push is optimal when latency is critical. Polling is optimal when producers cannot push, or as a safety net.

Queues

A queue is like a warehouse conveyor belt — messages align with the FIFO (First In, First Out) principle. Consumers read messages in order. This is a fundamental abstraction for the decoupling of producers and consumers.

flowchart LR
    P1[Producteur 1] --> Q["(Queue)"]
    P2[Producteur 2] --> Q
    Q --> C1[Consommateur 1]
    Q --> C2[Consommateur 2]
    style Q fill:#f9a825,color:#000

Pub/Sub (Publication / Subscription)

In the pub/sub pattern, producers publish events on topics without knowing who consumes them. Consumers subscribe to topics that interest them. Unlike queues, a pub/sub event can be delivered to multiple consumers simultaneously.

flowchart LR
    P[Producteur] --> T[Topic: payment.events]
    T --> A1[Agent Fraude]
    T --> A2[Agent Audit]
    T --> A3[Agent Notification]
    style T fill:#1565c0,color:#fff

Event Streaming (Kafka, Event Hubs)

Event streaming is like an infinite and lasting logbook. Unlike queues where messages are deleted after consumption, a stream retains the complete history. Consumers can replay the stream from any point.

IoT Sources and Sensors

IoT sensors generate continuous streams of telemetry readings. Each reading can trigger an agent if it exceeds a threshold. For example: temperature, pressure, vibrations of an industrial machine.

flowchart TD
    IoT[Capteur IoT] -->|Telemetry event| Stream[Event Stream]
    Stream --> Agent[Agent IA]
    Agent -->|Seuil dépassé ?| Decision{Décision}
    Decision -->|Oui| Alert[Alerte / Action]
    Decision -->|Non| Monitor[Continuer monitoring]

1.3 Essential agentics use cases

Here are the main use cases for event-driven agent systems.

Automated monitoring

This is your smoke detector. Agents subscribe to health and business metrics streams. They monitor component performance and trigger alerts when thresholds are crossed.

Like a detector that not only sounds but also:

• Identifies the part (affected component)
• Suggests nearest fire extinguisher (first action to take)
• Shows the probable cause of the problem
• Estimates likely impact radius
• In some cases, fixes the issue without human intervention

Incident triage

This is the emergency nurse from operations. Events arrive indicating a possible problem in your system (timeouts, user peaks, errors). The agent:

• Categorizes symptoms
• Retrieves the history of the “patient” (recent logs, recent deployments)
• Assigns a severity level
• Route to the right specialist (another agent or a human)

Automated approvals

Event-driven agents act like airport security: they verify identity, enforce rules, escalate exceptions, and maintain flow. They collect context, request human validation if necessary, set timers, and automatically advance or rewind if the approval window expires.

RAG Refresh (knowledge base update)

LLMs often rely on RAG ​​(Retrieval-Augmented Generation) technology. An event-driven agent can detect changes in data sources (new documents, policy updates) and automatically refresh the vector base used by the LLM.

sequenceDiagram
    participant DS as Source de données
    participant E as Event Stream
    participant A as Agent RAG Refresh
    participant VDB as Base vectorielle

    DS->>E: Event: DocumentUpdated
    E->>A: Trigger
    A->>DS: Récupérer nouveau document
    A->>VDB: Re-vectoriser et mettre à jour
    Note over A,VDB: Le LLM dispose maintenant<br/>de connaissances à jour

Commercial Fraud Detection

Agents monitor transaction streams in real time. They use historical context (stored facts) and LLM reasoning to identify suspicious patterns and trigger alerts or block transactions before they are finalized.

Operational Runbooks

Pre-defined workflows (runbooks) can be automatically triggered by specific events. The agent executes the runbook steps, reports the results, and escalates if anything falls outside the intended parameters.

2. Architectural patterns for event driven agents

2.1 Event Buses and Pub/Sub topologies

Centralized vs decentralized bus

Most agentics systems fail or succeed based on one fundamental decision: how to design the infrastructure.

Main tools

Kafka

Kafka is a popular tool for event-driven systems.

• Its logs are durable: events are available for replay and subsequent analysis.
• Supports consumer groups: different events can be sent to different parts of the business. Each team or department can be connected to its own consumer group.
• Ideal for: high volume streaming, replay, multi-teams.

NATS

If your main priority is to deliver and respond to events quickly and you don’t care about consumer groups, NATS may be more suitable than Kafka.

• NATS is like a high performance walkie-talkie: one service emits an event, another reacts immediately.
• By default, NATS does not persist events — this is a fire-and-forget pattern.
• Persistence can be added with JetStream if you want to examine events after they occur.

Azure Event Hubs, Service Bus, Event Grid

The Azure trio:

• Event Hubs: Kafka style streams
• Service Bus: enterprise messaging (queues, topics, sessions)
• Event Grid: serverless push, Software-as-a-Service routing and webhooks

graph TD
    subgraph Azure
        EH[Event Hubs<br/>Kafka-style streams]
        SB[Service Bus<br/>Enterprise messaging]
        EG[Event Grid<br/>Serverless routing]
    end

    Producer1[Producteur IoT] --> EH
    Producer2[Microservice] --> SB
    Producer3[Azure Resource] --> EG

    EH --> Agent1[Agent Analytics]
    SB --> Agent2[Agent Workflow]
    EG --> Agent3[Agent Webhook]

Topic design

Topic design is your street map. Names and routing keys should communicate intent and help distribute load. No topic should be a “trash” where everything ends up.

Example of naming by convention:

payments.v2.approved
payments.v2.declined
orders.v1.shipped
incidents.v1.created
incidents.v1.resolved

Partitioning based on tenant IDs allows each tenant to have a dedicated leader partition, which improves performance isolation.

graph LR
    subgraph Topic: payments.events
        P0[Partition 0<br/>tenant_A]
        P1[Partition 1<br/>tenant_B]
        P2[Partition 2<br/>tenant_C]
    end

    CG1[Consumer Group: FraudAgent] --> P0
    CG1 --> P1
    CG1 --> P2

    CG2[Consumer Group: AuditAgent] --> P0
    CG2 --> P1
    CG2 --> P2

2.2 State consistency and concurrency

The Saga pattern

A saga is a multi-stage story told through events and commands, with a plan for what to undo if something goes wrong. Saga is an excellent alternative to distributed transactions, less suited to robust event-driven systems.

Problem with distributed transactions:

• A single transaction across multiple resources may be simpler.
• It can simply be canceled if something goes wrong.
• However, it locks resources, which is problematic for high-volume systems.

sequenceDiagram
    participant O as Orchestrateur Saga
    participant Pay as Service Paiement
    participant Ship as Service Expédition
    participant Notif as Service Notification

    O->>Pay: Command: ChargeCustomer
    Pay-->>O: Event: CustomerCharged

    O->>Ship: Command: ShipOrder
    Ship-->>O: Event: OrderShipped

    O->>Notif: Command: NotifyCustomer
    Notif-->>O: Event: CustomerNotified

    Note over O: Succès !

    rect rgb(255, 200, 200)
        Note over O,Notif: Scénario d'échec
        O->>Pay: Command: ChargeCustomer
        Pay-->>O: Event: ChargeFailed
        O->>O: Action compensatoire:<br/>CancelOrder
    end

The Outbox pattern

The bug you never want to have: “we updated the state, but the event never left the building”. The outbox fixes that.

Principle:

1. The service writes both the business change AND the event to be published in the same database transaction.
2. A background publisher reads the outbox and outputs to the bus, marking each line as published.
3. With the outbox, state and event are either both committed or neither.

flowchart LR
    subgraph Transaction DB atomique
        BT[Changement business<br/>OrderUpdated] --> DB["(Base de données)"]
        EV[Event à publier<br/>OrderShipped] --> DB
    end

    DB --> BG[Background Publisher]
    BG --> Bus[Event Bus]
    BG --> DB
    Note[Marquer comme publié]
    BG -.-> Note

Exactly-once vs Effectively-once

Ideally, your system should process each event exactly once. For example, a payment should never be triggered twice.

However, exactly-once is not always easy to implement in practice.

A more pragmatic alternative is effectively-once, obtained by combining:

• At-least-once delivery (guaranteed but potentially repeated delivery)
• Idempotency keys
• Effect logs

How ​​it works:

• Assign each request or event a unique key.
• If you see the same key again, return the same result without redoing the work.
• Like pressing a button twice without feeling any effect the second time.

2.3 Avoiding common pitfalls

Real systems can behave strangely when you least expect it — at 2 a.m. It is impossible to build a system that never behaves strangely. The key is to expect it and prevent it from causing damage.

Event Storms

Problem: Sudden bursts from retries or cascading failures can overwhelm critical parts of the system.

Mitigation:

• Rate limits
• Prioritize essential consumers
• Be prepared to shed non-critical load under stress

Hot Partitions

Problem: Too much traffic is being sent to the same partition.

Solutions:

• Best partitioning strategy
• Use a more distributed key partition
• Hashing the partition key
• Add random numbers before hashing (technique called salting)
• Increase number of partitions

Duplicate messages

Problem: Duplicates are expected with intermittent failures, retries, and consumer rebalances.

Solution: They are harmless if your handlers are idempotent. For additional security, apply the effects log to verify that the duplicate delivery does not cause a duplicate effect.

Poison Messages

Problem: Inputs consistently fail due to an error in the system.

Solution:

• Don’t let them block the main queue
• Configure bounded retries
• Route to a dead-letter queue with enough context to debug
• Quarantine until root cause is known

Head-of-line Blocking

Problem: A slow or too large message blocks an entire batch.

Solution:

• Reduce batch size
• Configure timeouts
• Separate topics by SLA so that slow work does not block fast work

Fan-out Bottlenecks

Problem: Lots of subscribers or slow syncs turn the bus into a traffic jam.

Solutions:

• Stagnify topics
• Derive events later asynchronously
• Use async webhooks rather than waiting for downstream processing

Cold Starts

Problem: Additional latency during system startup (initializing containers, loading models, etc.).

Mitigation:

• Pre-warming
• Keep a minimum of active instances for critical workflows
• Monitor cold start metrics and integrate them into SLAs

2.4 Latency, throughput and backpressure

Performance rests on three pillars: response speed, volume processed, and the ability to gracefully say “not now.”

Batching

batching consists of grouping several messages together. It amortizes additional costs (network, serialization, authentication), but it trades latency for throughput: you process more, but slower.

Windowing

The windowing groups events by time or by account to aggregate them without drowning.

Examples:

• Fraud scoring over a 1 second window
• Inventory updates on 100 events

Consumer Groups as a scalability lever

Each consumer group reads the same stream independently, which allows your triage agent, your fraud agent, and your analytics job to not compete for the same events. You can scale them at their own pace and deploy them separately.

graph TD
    Stream[Event Stream<br/>payments.events] --> CG1[Consumer Group: FraudAgent<br/>3 instances]
    Stream --> CG2[Consumer Group: AuditAgent<br/>1 instance]
    Stream --> CG3[Consumer Group: Analytics<br/>5 instances]

Rate Limiting

Rate limiting protects downstream APIs from receiving more data than they can handle, both for security and to prevent your system from being slowed down.

Patterns:

• Token bucket: a token pool fills at a fixed rate; each request consumes a token
• Leaky bucket: requests are processed at a constant rate, regardless of the arrival rate

Apply where the pressure is greatest: at the edge, in API Gateways, or just before the fragile dependency.

Circuit Breaker

When a dependency deteriorates, the circuit breaker:

1. Fails fast
2. Keep retrying with exponentially increasing backoff
3. Offers graceful fallback to users rather than an infinite spinner

stateDiagram-v2
    [*] --> Fermé
    Fermé --> Ouvert : Taux d'erreur dépasse le seuil
    Ouvert --> Demi_ouvert : Après délai de backoff
    Demi_ouvert --> Fermé : Succès de la requête test
    Demi_ouvert --> Ouvert : Échec de la requête test
    Demi_ouvert: Demi-ouvert

Configuration in Azure Functions (Event Hub Trigger)

In an Azure Functions app, when integrating an Event Hub trigger:

• Configure BatchSize to control the number of messages per batch
• Configure maximumWaitTime to adjust latency/throughput trade-off

Performance recipe:

1. Batch + windowing to go faster
2. Scaling with consumer groups
3. Enforce rate limits to protect dependencies
4. Trigger circuit breakers to fail safely

3. Frameworks and tooling for event driven agents

3.1 Messaging and streaming

Different tools transmit events in different ways. Choose the wrong tool and everything afterward suffers. Here’s how to match the tool as needed.

Scenario 1: Sustainability and high volume

Use case: Management of interactions between users of a large social network, refresh of a large RAG database, complex agentic triage, etc.

Tools: Kafka or Azure Event Hubs

These tools act as a durable event log that stores the entire history. They excel at:

• Highest throughput fan-out: processing of numerous events in parallel
• Replay in case of problem
• Multiple independent consumers

Scenario 2: Speed ​​above all

Use case: Short-lived events, such as reacting to an IoT telemetry reading that can change in milliseconds. The replay does not count, nor the history.

Tools: NATS or Azure Service Bus

These tools offer:

• Pattern request/reply or subscriptions by topic
• Light topics
• Optional persistence if necessary

Scenario 3: Coordination of simple tasks

Use case: Simple coordination where big-data tools like Kafka would be oversized.

Tools: Redis Streams or Azure Queue Storage

• Redis Streams: lives where you already have Redis, lightweight, good for local queues and micro-streams with consumer groups and simple retention.
• Azure Queue Storage: if consumer groups are not important, simple work queue.

Neither of these tools is a Kafka replacement for massive multi-team replay, but both are great for limited scope pipelines.

graph TD
    subgraph "Haute volumétrie + durabilité"
        K[Kafka / Azure Event Hubs]
    end

    subgraph "Vitesse + coordination légère"
        N[NATS / Azure Service Bus]
    end

    subgraph "Tâches simples"
        R[Redis Streams / Azure Queue Storage]
    end

    Req{Besoin} -->|Big data, replay, multi-teams| K
    Req -->|Faible latence, IoT, agent-to-agent| N
    Req -->|Simple, local, micro-pipeline| R

Azure Event Hubs configuration — Number of partitions

The number of partitions determines the possible parallelism. The more partitions, the more horizontally you can scale consumers.

{
  "partitionCount": 32,
  "messageRetentionInDays": 7,
  "captureDescription": {
    "enabled": true,
    "destination": {
      "name": "EventHubArchive.AzureBlockBlob"
    }
  }
}

3.2 Sustainable orchestration and workflows

Agents don’t always end in one fell swoop: payments expire, people approve later. This is why we need an orchestration that coordinates the agents.

What is a durable orchestrator?

Sustainable orchestrators provide:

• A deterministic state
• Built-in retries
• timers
• external events (human approvals)

Azure Durable Functions

The primary durable orchestrator in Azure is Durable Functions. You write all your coordination logic in code. It is therefore generally the developers who build these workflows. These functions are often embedded in applications hosted in the cloud.

Key Features:

• The sustainable state keeps track of the steps already completed behind the scenes.
• If a problem on the host causes the orchestration function to restart, it will skip steps already completed. For example, if it has already processed the Charge and Ship stages, it will move directly to the next one, Notify.
• The function handles errors using deterministic state: it keeps track of the steps where it failed to choose the appropriate compensatory action.
• Supports the human-in-the-loop pattern if manual validation is required.

Order processing saga example:

# Azure Durable Function - Order Processing Saga
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    order_id = context.get_input()

    try:
        # Étape 1 : Facturer le client
        charge_result = yield context.call_activity("ChargeCustomer", order_id)

        # Étape 2 : Expédier la commande
        ship_result = yield context.call_activity("ShipOrder", order_id)

        # Étape 3 : Notifier le client
        yield context.call_activity("NotifyCustomer", order_id)

        return {"status": "success", "orderId": order_id}

    except Exception as e:
        # Action compensatoire en cas d'échec
        yield context.call_activity("CancelOrder", order_id)
        raise

main = df.Orchestrator.create(orchestrator_function)

AWS Step Functions

On AWS, the tool of choice is Step Functions. He uses a different philosophy:

• Operates in drag-and-drop mode, definition-first
• Usually built by platform engineers
• Can coordinate multiple services directly
• Span across services

Features common to all durable orchestrators

Regardless of the tool chosen, all durable orchestrators share:

1. Deterministic state: ability to resume where we left off
2. Compensation steps: cancellation actions in case of failure
3. Human-in-the-loop: support for manual validations

3.3 Agent stacks and integrations

LLM agents are only as intelligent as the tools and memory connected to them.

Agent Types

The integration of these two approaches allows agents that are both reactive to new events and capable of contextual and informed decision-making over time.

Semantic Kernel (Microsoft)

Semantic Kernel is a Microsoft tool that allows any arbitrary code to interact directly with an LLM.

• Perfect for .NET environments and Azure integration
• Also available in Python and other languages

Configuration example for a chatbot:

// Configuration de Semantic Kernel avec Ollama (local)
var builder = Kernel.CreateBuilder();

builder.AddOllamaChatCompletion(
    modelId: "llama3.2",
    endpoint: new Uri("http://localhost:11434")
);

var kernel = builder.Build();

// Peut aussi se connecter à un LLM frontier en ligne (GPT-4, Claude, etc.)
// builder.AddOpenAIChatCompletion("gpt-4o", apiKey);

// L'endpoint reçoit le message de l'utilisateur
app.MapPost("/chat", async (ChatMessage userMessage, Kernel kernel) =>
{
    chatHistory.AddUserMessage(userMessage.Content);

    // Streaming de la réponse vers le navigateur
    var response = kernel.InvokePromptStreamingAsync(
        chatHistory.ToString()
    );

    return Results.Stream(async stream => {
        await foreach (var chunk in response)
            await stream.WriteAsync(Encoding.UTF8.GetBytes(chunk.ToString()));
    });
});

LangChain and LangGraph

LangChain and LangGraph are used to orchestrate an agentic flow. They are similar to sustainable orchestration tools but designed specifically for multi-step agentic AI workflows.

• Python-first approach
• Can be integrated with sustainable orchestration tools
• Not designed for generic arbitrary workflows

Example of using LangChain for an incident triage agent:

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
import json

@tool
def query_recent_logs(service_name: str, time_window_minutes: int) -> str:
    """Récupère les logs récents d'un service."""
    # Requête vers Application Insights ou équivalent
    return get_logs_from_monitoring(service_name, time_window_minutes)

@tool
def check_recent_deployments(service_name: str) -> str:
    """Vérifie les déploiements récents pour un service."""
    return get_deployment_history(service_name)

@tool
def assign_severity(incident_description: str) -> dict:
    """Assigne un niveau de sévérité à un incident."""
    return classify_severity(incident_description)

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [query_recent_logs, check_recent_deployments, assign_severity]

agent = create_openai_tools_agent(llm, tools, system_prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# Déclenché par un webhook d'incident
@app.post("/incident-webhook")
async def handle_incident(incident: IncidentEvent):
    result = await agent_executor.ainvoke({
        "input": f"Incident reçu: {incident.description}"
    })
    return result

AutoGen (Microsoft)

AutoGen is excellent for multi-agent conversations and role orchestration. Different agents with different roles collaborate to solve complex problems.

Example of triage agent with AutoGen:

import autogen

# Agent coordinateur
coordinator = autogen.AssistantAgent(
    name="IncidentCoordinator",
    system_message="""Tu es le coordinateur d'incidents. 
    Tu analyses les incidents, délègues aux spécialistes, et prends les décisions finales.
    Tu dois toujours fournir un niveau de sévérité (P1/P2/P3/P4) et une action recommandée.""",
    llm_config={"model": "gpt-4o"}
)

# Agent spécialiste infrastructure
infra_specialist = autogen.AssistantAgent(
    name="InfraSpecialist",
    system_message="""Tu es un spécialiste en infrastructure. 
    Tu analyses les métriques système, les logs serveur, et les problèmes réseau.""",
    llm_config={"model": "gpt-4o"}
)

# Agent spécialiste applicatif
app_specialist = autogen.AssistantAgent(
    name="AppSpecialist",
    system_message="""Tu es un spécialiste en développement applicatif.
    Tu analyses les erreurs de code, les exceptions, et les problèmes de performance.""",
    llm_config={"model": "gpt-4o"}
)

# Proxy humain pour l'interaction
human_proxy = autogen.UserProxyAgent(
    name="HumanProxy",
    human_input_mode="NEVER",  # Automatique
    code_execution_config={"work_dir": "triage_workspace"}
)

# Chat de groupe pour la collaboration
groupchat = autogen.GroupChat(
    agents=[coordinator, infra_specialist, app_specialist, human_proxy],
    messages=[],
    max_round=10
)

manager = autogen.GroupChatManager(groupchat=groupchat)

# Déclenchement sur reception d'un event d'incident
async def process_incident_event(event: dict):
    await human_proxy.initiate_chat(
        manager,
        message=f"Incident à traiter: {json.dumps(event)}"
    )

3.4 Cloud triggers and event routers

Before your agents can think, they must hear. event routers connect the outside world to your workflows by managing subscriptions, filters, delivery and retries.

Azure Event Grid

Event Grid is Azure’s serverless event router.

• Routing with subject filters
• Dead-lettering (message queue failed)
• Interactions between Azure services and webhooks

AWS EventBridge

On AWS, the corresponding tool is EventBridge.

• Rich rule patterns
• Event nozzles by domain
• Software-as-a-Service Integrations
• Schema registry

Google Cloud Eventarc

On GCP, Eventarc routes cloud events through GCP services to Cloud Run and Functions.

Event Grid configuration example — topic retail-events

A retail-events topic addresses payments and refunds. Two subscriptions are configured:

1. payments-sub: Retrieves an event and sends it to an HTTP endpoint for further processing. Filter: subject begins with “payments”.

2. audit-sub: Sends events to a Storage Queue for durable auditing.

Dead-letter destination: configured on a storage account container to capture failed events.

{
  "topicName": "retail-events",
  "subscriptions": [
    {
      "name": "payments-sub",
      "destination": {
        "endpointType": "WebHook",
        "properties": {
          "endpointUrl": "https://func.azurewebsites.net/api/ProcessPayment"
        }
      },
      "filter": {
        "subjectBeginsWith": "payments"
      },
      "deadLetterDestination": {
        "endpointType": "StorageBlob",
        "properties": {
          "resourceId": "/subscriptions/.../storageAccounts/deadletter",
          "blobContainerName": "dead-letters"
        }
      }
    },
    {
      "name": "audit-sub",
      "destination": {
        "endpointType": "StorageQueue",
        "properties": {
          "resourceId": "/subscriptions/.../storageAccounts/audit",
          "queueName": "audit-queue"
        }
      }
    }
  ]
}

When to use a router: When you need to push from many producers to one or more agent entry points without having to manage your own consumers.

4. Design real time agent workflows

4.1 Event-native agent design

In event-driven systems, an event does something to an AI agent. Then some steps happen inside the agent.

The internal steps of an event-driven agent

flowchart TD
    EV[Event entrant] --> T[1. Trigger<br/>Réveille l'agent]
    T --> G[2. Guardrails<br/>Doit-il s'exécuter ?]
    G -->|Non| STOP[Rejet / Ignore]
    G -->|Oui| TS[3. Tool Selection<br/>Quelles fonctions utiliser ?]
    TS --> MA[4. Memory Attachment<br/>Contexte sémantique + épisodique]
    MA --> LLM[5. LLM Reasoning<br/>Décision]
    LLM --> ACTION[6. Action<br/>Appel d'outil, event, réponse]

1. Trigger

This is what wakes up the agent when an event is sent to it. It could be:

• From an HTTP endpoint if the event is sent via webhook
• From an Event Grid listener
• From a topic subscriber (Kafka, Service Bus, etc.)

2. Guardrails

Guardrails are the rules that decide whether the agent should run. For example, a guardrail can protect against a malicious prompt. In event-driven systems, this role can often be delegated to the event router by subscribing only to necessary events and applying subscription filters.

3. Tool Selection

An agent can perform different functions to resolve different problems. It can use various technologies to connect to external data sources. These elements are collectively known as tools in the context of an agent.

4. Memory Attachment

Different types of memory serve different purposes:

Memory optimization:

• Keep context windows tight
• Retrieve a minimum of semantic snippets
• Summarize episodic state rather than always skipping the full conversation history

Worked example: AutoGen-based incident triage agent

# Fonction Azure déclenchée par HTTP POST
# Ce POST endpoint est son trigger
@app.route(route="incident-triage", methods=["POST"])
async def incident_triage(req: func.HttpRequest) -> func.HttpResponse:
    # Validation du schéma de l'event entrant
    incident_data = req.get_json()

    # Guardrail : validation que les champs requis sont présents
    required_fields = ["service_name", "error_message", "timestamp", "severity_hint"]
    if not all(field in incident_data for field in required_fields):
        return func.HttpResponse(
            json.dumps({"error": "Missing required fields"}),
            status_code=400
        )

    # Récupérer la mémoire sémantique (runbooks, incidents passés similaires)
    semantic_context = await search_runbooks(incident_data["service_name"])

    # Initialiser les agents AutoGen avec leur contexte
    coordinator = autogen.AssistantAgent(
        name="Coordinator",
        system_message=f"""Tu es le coordinateur d'incidents.
        Runbooks disponibles: {semantic_context}
        Incident actuel: {json.dumps(incident_data)}
        """
    )

    # ... initialisation des autres agents ...

    result = await run_triage_chat(coordinator, incident_data)
    return func.HttpResponse(json.dumps(result), status_code=200)

4.2 Inter-agent protocols

When agents collaborate, they need a language, communication channels, and rules — it’s your protocol.

Topics

topics define where events go. They allow agents to subscribe to specific types of events. Multiple agents can listen to the same events. Additional filtering by agent can then be done based on:

• Event subject
• File extension
• Any custom event data

Roles

Roles define which agent does what. They are necessary for multi-agent interactions. Each agent thus has its own focused context on which to work.

Message Schema

The form of data that agents work with is defined by the message schema — a pre-agreed communication format that an agent expects.

Why schemas are important:

• Prevent hallucinated fields
• Allow deterministic validation during message ingestion
• Allow versioning and evolution of the agentic system
• Support replay and audit

Example of incident event diagram:

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "IncidentEvent",
  "type": "object",
  "required": ["event_id", "service_name", "error_message", "timestamp", "severity_hint"],
  "properties": {
    "event_id": {
      "type": "string",
      "format": "uuid",
      "description": "Identifiant unique de l'event"
    },
    "service_name": {
      "type": "string",
      "description": "Nom du service affecté"
    },
    "error_message": {
      "type": "string",
      "description": "Description de l'erreur"
    },
    "timestamp": {
      "type": "string",
      "format": "date-time"
    },
    "severity_hint": {
      "type": "string",
      "enum": ["P1", "P2", "P3", "P4"]
    },
    "metadata": {
      "type": "object",
      "additionalProperties": true
    }
  },
  "additionalProperties": false
}

Scratch Pad (shared dashboard)

Agents can share a scratch pad — what agents remember together. It may contain:

• Intermediate Reasoning Artifacts
• Partial plans
• Decisions already made – Facts discovered so far

Delegation Contracts

Once a decision has been made, delegation contracts allow it to be entrusted to other agents. A delegation contract defines:

• What is delegated
• The input schema with the expected output
• Constraints like time limits for the task
• The escalation path if the time limit is exceeded
• The authority limits that define what the receiving agent can and cannot do

Example of delegation contract:

class DelegationContract:
    task_description: str
    input_schema: dict          # Schéma JSON de l'input
    expected_output_schema: dict # Schéma JSON de l'output attendu
    time_limit_seconds: int     # Limite de temps
    escalation_path: str        # Chemin d'escalade si timeout
    authority_boundaries: list  # Actions autorisées/interdites

# Exemple : délégation au spécialiste DB
db_contract = DelegationContract(
    task_description="Analyser les performances de la base de données",
    input_schema={"service_name": "string", "query_logs": "array"},
    expected_output_schema={"root_cause": "string", "recommended_fix": "string", "confidence": "number"},
    time_limit_seconds=120,
    escalation_path="human_dba_team",
    authority_boundaries=["read_query_plans", "read_indexes", "NOT:execute_queries", "NOT:modify_schema"]
)

4.3 Human-in-the-loop and SLAs

Although AI can do many things, some decisions require a human — especially critical decisions where the consequences of making a mistake are severe.

How the Human-in-the-loop pattern works

The human-in-the-loop pattern allows agents to receive validation or information from a real person before the workflow can continue.

Example: Reimbursement approval process with 15 minute limit

sequenceDiagram
    participant O as Orchestrateur
    participant H as Humain (Équipe)
    participant PD as PagerDuty / Email
    participant T as Timer

    O->>O: Proposer remboursement
    O->>H: Demander approbation
    O->>T: Démarrer timer 15 minutes

    par Attente parallèle
        H->>O: approval_event (approuvé/refusé)
    and
        T->>O: Timeout après 15 min
    end

    alt approval_event reçu en premier
        O->>O: Traiter selon décision (approuvé/refusé)
    else Timeout atteint
        O->>PD: Escalade (PagerDuty, email, etc.)
        O->>T: Redémarrer timer (fenêtre d'escalade 30 min)
        alt Approbation reçue dans fenêtre d'escalade
            O->>O: Traiter selon décision
        else Timeout d'escalade
            O->>O: Exécuter fallback<br/>(remboursement partiel, pause fulfillment)
        end
    end

Implementation with Azure Durable Functions:

import azure.durable_functions as df
from datetime import timedelta

def refund_approval_orchestrator(context: df.DurableOrchestrationContext):
    order_data = context.get_input()

    APPROVAL_WINDOW = timedelta(minutes=15)
    ESCALATION_WINDOW = timedelta(minutes=30)

    # Proposer le remboursement
    yield context.call_activity("ProposeRefund", order_data)

    # Notifier l'équipe
    yield context.call_activity("NotifyApprovalRequired", {
        "order": order_data,
        "deadline": context.current_utc_datetime + APPROVAL_WINDOW
    })

    # Attente parallèle : approbation OU timeout (15 min)
    approval_event = context.wait_for_external_event("ApprovalDecision")
    timeout_task = context.create_timer(
        context.current_utc_datetime + APPROVAL_WINDOW
    )

    winner = yield context.task_any([approval_event, timeout_task])

    if winner == approval_event:
        # L'humain a répondu dans les temps
        decision = approval_event.result
        if decision["approved"]:
            yield context.call_activity("ProcessRefund", order_data)
        else:
            yield context.call_activity("RejectRefund", order_data)
    else:
        # Timeout : escalade
        yield context.call_activity("EscalateApproval", {
            "order": order_data,
            "channels": ["pagerduty", "email", "slack"]
        })

        # Deuxième fenêtre d'attente (30 min)
        approval_event2 = context.wait_for_external_event("ApprovalDecision")
        escalation_timeout = context.create_timer(
            context.current_utc_datetime + ESCALATION_WINDOW
        )

        winner2 = yield context.task_any([approval_event2, escalation_timeout])

        if winner2 == approval_event2:
            decision = approval_event2.result
            if decision["approved"]:
                yield context.call_activity("ProcessRefund", order_data)
            else:
                yield context.call_activity("RejectRefund", order_data)
        else:
            # Fallback automatique après expiration totale
            yield context.call_activity("ExecuteFallbackRefund", order_data)

main = df.Orchestrator.create(refund_approval_orchestrator)

Define SLAs for Human-in-the-loop

In the human-in-the-loop pattern, we typically define:

Rate caps per tenant: Rate caps can be added at a time so that a burst of approvals does not overwhelm reviewers.

RATE_LIMITS = {
    "tenant_small": {"approvals_per_hour": 10},
    "tenant_medium": {"approvals_per_hour": 50},
    "tenant_enterprise": {"approvals_per_hour": 200}
}

5. Making event driven agents robust

5.1 Taxonomy of failures

Not all failures are equal. Treating all errors the same way will either cause you to alert too often or miss the real problem.

Transit Errors

Features: Temporary issues — brief network outages, momentary throttling, slowness due to cold caches.

Solution: Generally recoverable with a simple retry and backoff.

Systemic Errors

Features: Widespread problems — regional crash, hot partitions, dependency crash.

Solution: Load shedding, circuit breakers, climbing. Blind retries will not work.

Logic Errors

Features: Bugs or bad assumptions — schema incompatibilities, null references, malformed payloads.

Solution: Do not correct themselves. Route to dead-letter and quarantine with enough context to debug, then resolve via patch or software update.

flowchart TD
    ERR[Erreur détectée] --> TYPE{Type d'erreur ?}
    TYPE -->|Transitoire| RETRY[Retry avec backoff]
    TYPE -->|Systémique| SHED[Load shedding + Circuit Breaker + Escalade]
    TYPE -->|Logique| DLQ[Dead-letter queue + Quarantaine]
    RETRY -->|Succès| OK[✓ Résolu]
    RETRY -->|Max retries| DLQ
    SHED --> ALERT[Alerte équipe]
    DLQ --> DEBUG[Déboguer + Patch]

Monitoring errors in Azure

The main tool for monitoring errors in Azure is Application Insights. It collects different types of application telemetry:

1. Logs: human-readable descriptions of individual events

• Different severity levels: Information (1), Warning (2), Error (3), Critical (4/5)
• In the raw view, the numbers represent these levels (ex: 3 = Error, 1 = Information)

2. Traces: multiple log entries linked by a correlation ID

• Examine complete traces via the “Search” blade of the “Investigate” panel

3. Metrics: simple measurements like counters and durations

• Easily visualized in graphs
• Can be linked to alert rules

KQL query to retrieve all log data:

union *
| where timestamp > ago(1h)
| order by timestamp desc

5.2 Retries done well

Retries can heal or worsen. Done well, they speed up recovery. Done badly, they start a storm.

Retry Strategy: Exponential Backoff

You should not simply retry at short, regular intervals. We use the exponential backoff to space out the retries:

Jitter

Adding jitter (random waiting time) between retries avoids thundering herd — preventing many clients from retrying simultaneously, which would amplify the problem.

Bounded Retries

Retries should not continue indefinitely. If the service is truly down, infinite retries will exhaust the hardware. It is therefore necessary:

• Limit maximum number of attempts
• Mark as poison if a message still fails after the max
• For routers like Event Grid, understand the built-in retry and add a dead-letter destination so as not to silently lose anything

Retries in Azure — Implementation options

Event Grid — Integrated Retry

Event Grid has its own built-in retry mechanism — nothing you need to do on your end. However, you must add a dead-letter destination (eg: a Storage Account) to examine messages that have exhausted all attempts.

Service Bus — Configuration by subscription

In Service Bus, the maximum number of delivery attempts can be configured on a per-subscription basis on a topic. Service Bus has a built-in dead-letter queue to which messages are sent when attempts are exhausted.

Azure Durable Functions — Retry in application code

import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    order_id = context.get_input()

    # Configuration du retry : 3 tentatives max, backoff exponentiel
    retry_options = df.RetryOptions(
        first_retry_interval_in_milliseconds=1000,  # 1 seconde
        max_number_of_attempts=3,
        backoff_coefficient=2.0,          # Exponentiel : 1s, 2s, 4s
        max_retry_interval_in_milliseconds=30000,    # Max 30 secondes
    )

    try:
        # Tentative de traitement de la commande avec retry
        result = yield context.call_activity_with_retry(
            "ProcessOrder",
            retry_options,
            order_id
        )
        return result

    except Exception as e:
        # Après épuisement des retries, passer à la compensation
        yield context.call_activity("CompensateOrder", order_id)
        raise

5.3 Idempotency strategies

The problem with retries is that they can cause duplicate messages to be sent to consumers. Idempotency strategies ensure that a duplicate event does not result in a repeated action — they ensure effectively-once delivery.

Idempotency Keys and Effect Logs

Principle:

1. Assign each request a unique idempotency key (per request or per event)
2. Store the result of the event in an effect log with this key
3. When an event with a duplicate key arrives, the action is not executed again because there is already an entry with the same key in the effect log

This is how we prevent, for example, double billing of a customer.

import hashlib
import json
from datetime import datetime

async def process_payment_idempotent(payment_event: dict, db) -> dict:
    # Générer ou utiliser l'idempotency key
    idempotency_key = payment_event.get("idempotency_key") or \
                      hashlib.sha256(json.dumps(payment_event, sort_keys=True).encode()).hexdigest()

    # Vérifier si cet event a déjà été traité
    existing_result = await db.effect_log.find_one({"key": idempotency_key})
    if existing_result:
        # Retourner le même résultat sans refaire le travail
        return existing_result["result"]

    # Traiter le paiement
    result = await charge_customer(payment_event)

    # Enregistrer dans le log d'effets
    await db.effect_log.insert_one({
        "key": idempotency_key,
        "result": result,
        "processed_at": datetime.utcnow(),
        "event_type": "payment_processed"
    })

    return result

Example in Azure Cosmos DB:

# Azure Cosmos DB - Upsert idempotent
async def upsert_idempotent(container, event: dict):
    item = {
        "id": event["idempotency_key"],
        "partition_key": event["tenant_id"],
        "event_data": event,
        "processed_at": datetime.utcnow().isoformat(),
        "status": "processed"
    }

    # Upsert : crée ou met à jour si la clé existe déjà
    # Si la clé existe, le résultat est le même — idempotent
    await container.upsert_item(item)

Semantic idempotence

Some actions are semantically idempotent in nature and do not require an idempotency key:

• A duplicate DELETE action is safe because the item is already deleted
• A duplicate UPDATE action is safe if all updated values are the same as those already stored

Semantic idempotence requires no additional storage, but it is not possible for operations with side effects like payments.

State Machines

A state machine advances only forward. It is impossible to execute an action that has already been executed. This is what sustainable orchestrators use to skip steps that have already been completed.

stateDiagram-v2
    [*] --> OrderCreated
    OrderCreated --> PaymentProcessing
    PaymentProcessing --> PaymentConfirmed
    PaymentProcessing --> PaymentFailed
    PaymentConfirmed --> Shipping
    Shipping --> Delivered
    PaymentFailed --> OrderCancelled
    Delivered --> [*]
    OrderCancelled --> [*]

    note right of PaymentProcessing
        Un paiement déjà confirmé
        ne peut pas repasser en
        PaymentProcessing
    end note

State machines are ideal for:

• The human-in-the-loop pattern
• Easy configuration of timeouts

Outbox Pattern (reminder)

The outbox pattern publishes an event and its results in the same transaction. Since both are in a transaction, either both are saved or neither. This pattern is often used in distributed systems and microservices where background workers publish events — the publisher is thus idempotent.

6. Observability, monitoring and security

6.1 Traces, metrics and logs

If you can’t see it, you can’t deliver it. You need to know how your agents operate and why they do what they do. Every agent decision and every event in the hub must be traceable.

The hierarchy: Logs → Traces → Metrics

graph TD
    subgraph Metrics
        M[Métriques<br/>Compteurs, durées<br/>Alerting, graphiques]
    end

    subgraph Traces
        T[Traces<br/>Opérations corrélées<br/>Root span + Child spans]
        L1[Log entry 1<br/>correlation_id: abc]
        L2[Log entry 2<br/>correlation_id: abc]
        L3[Log entry 3<br/>correlation_id: abc]
        T --> L1
        T --> L2
        T --> L3
    end

    M -->|"Qui regarder ?"| T

    style M fill:#1565c0,color:#fff
    style T fill:#2e7d32,color:#fff

Logs

An individual log entry has little meaning out of context. Logs are human-readable descriptions of individual events.

Features:

• Severity levels: Information, Warning, Error, Critical
• Can use templates and formatters to be easily searchable
• Custom attributes to enable linking between entries

import structlog
import logging

# Configuration d'un logger structuré
logger = structlog.get_logger()

# Log information normale
logger.info(
    "order_processed",
    order_id="ORD-12345",
    tenant_id="tenant_A",
    processing_time_ms=342,
    correlation_id="corr-abc-123"
)

# Log d'erreur avec contexte
logger.error(
    "payment_failed",
    order_id="ORD-12345",
    error_code="CARD_DECLINED",
    retry_count=2,
    correlation_id="corr-abc-123",
    severity=3  # 3 = Error dans Azure App Insights
)

Traces

A trace connects multiple log entries to tell a coherent story. These stories provide enough context to verify behavior and diagnose problems.

Structure of a trace:

• Root span: defines the limits of the operation (start → completion)
• Child spans: suboperations with their own traces
• Entries are linked via shared identifier (ID correlation)

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

tracer = trace.get_tracer("payment-service")

async def process_payment(payment_event: dict):
    # Root span pour l'opération complète
    with tracer.start_as_current_span("process_payment") as root_span:
        root_span.set_attribute("order.id", payment_event["order_id"])
        root_span.set_attribute("tenant.id", payment_event["tenant_id"])

        # Child span pour la validation
        with tracer.start_as_current_span("validate_payment") as validate_span:
            validation_result = await validate(payment_event)
            validate_span.set_attribute("validation.result", str(validation_result))

        # Child span pour l'appel à la passerelle de paiement
        with tracer.start_as_current_span("charge_gateway") as gateway_span:
            gateway_span.set_attribute("gateway.name", "stripe")
            result = await charge_via_gateway(payment_event)

        return result

Metrics

Metrics are high-frequency measurements that your system outputs continuously. They represent simple values ​​(counters, durations) that are easily visualized in graphs and linked to alert rules.

They indicate where to look in the traces.

from azure.monitor.opentelemetry import configure_azure_monitor
from opentelemetry import metrics

# Configuration des métriques Azure
configure_azure_monitor(connection_string=os.environ["APPLICATIONINSIGHTS_CONNECTION_STRING"])

meter = metrics.get_meter("agent-metrics")

# Compteur d'events traités
events_processed = meter.create_counter(
    "events.processed",
    description="Nombre d'events traités"
)

# Histogramme des temps de traitement
processing_duration = meter.create_histogram(
    "events.processing_duration_ms",
    description="Durée de traitement des events en millisecondes"
)

# Gauge du nombre d'agents actifs
active_agents = meter.create_up_down_counter(
    "agents.active_count",
    description="Nombre d'agents actifs"
)

# Utilisation dans le code
async def process_event(event: dict):
    start_time = time.time()
    try:
        result = await handle_event(event)
        events_processed.add(1, {"status": "success", "type": event["type"]})
        return result
    except Exception as e:
        events_processed.add(1, {"status": "error", "type": event["type"]})
        raise
    finally:
        duration = (time.time() - start_time) * 1000
        processing_duration.record(duration, {"type": event["type"]})

6.2 Identity and access

Your agents perform critical business tasks. You shouldn’t let just anyone trigger them. Authorized users and services must prove who they are and that they are authorized to perform the actions they are attempting.

Authentication vs Authorization

Managed Identity (Azure)

In Azure, each service can be assigned a managed identity. Other services can be explicitly configured to accept or deny access from this service. No explicit secrets are stored — this is configured at the level of each individual Azure resource.

from azure.identity import DefaultAzureCredential
from azure.servicebus import ServiceBusClient

# Authentification via Managed Identity — aucun secret en dur
credential = DefaultAzureCredential()

servicebus_client = ServiceBusClient(
    fully_qualified_namespace="my-namespace.servicebus.windows.net",
    credential=credential  # Utilise la Managed Identity automatiquement
)

Mutual TLS (mTLS)

In device-to-device communication, mutual TLS is often used. For example, it is very common in the MQTT protocol used in IoT. Each device has a valid X.509 certificate. The certificate key is validated on the server side — that’s how the server knows it’s dealing with a real device that’s supposed to be part of its network.

Role-Based Access Control (RBAC)

One of the most common ways to enforce authorization permissions is RBAC. Roles define who is allowed to do what. Each user or agent can have multiple roles combined, allowing fine-grained access control where specific resources and operations are restricted by specific roles.

from enum import Enum
from functools import wraps

class Role(Enum):
    READ_ONLY = "read_only"
    OPERATOR = "operator"
    ADMIN = "admin"
    AGENT_EXECUTOR = "agent_executor"

# Décorateur d'autorisation RBAC
def require_role(*required_roles: Role):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, token: str, **kwargs):
            user_roles = await get_user_roles_from_token(token)

            if not any(role in user_roles for role in required_roles):
                raise PermissionError(
                    f"Accès refusé. Rôles requis: {required_roles}, "
                    f"rôles actuels: {user_roles}"
                )
            return await func(*args, **kwargs)
        return wrapper
    return decorator

# Utilisation
@require_role(Role.OPERATOR, Role.ADMIN)
async def trigger_triage_agent(incident: dict, token: str):
    # Seuls les operators et admins peuvent déclencher l'agent
    await process_incident(incident)

6.3 Data security and privacy

Securing access to your data is important. However, it is equally important to ensure that you do not accidentally leak sensitive data to the outside world.

Fundamentals

1. Never log raw PII (Personally Identifiable Information) in places where they are not relevant, such as application logs. If they are part of a loggable payload, redact them.

Not sure what constitutes PII? Assume all user information is in it. Use a whitelist to explicitly allow certain safe information to be published.

2. Encrypt the payload when transferring sensitive data. Do not transfer raw data in the clear. Ensure encryption keys are securely stored, revoked if compromised, and rotated/refreshed regularly.

3. Limit the lifetime of access tokens — sometimes it should be long, but most of the time it shouldn’t be.

Redaction of PII in logs

import re
from typing import Any

class PIIRedactor:
    """Utilitaire pour redacter les PII des données de log."""

    @staticmethod
    def redact_email(email: str) -> str:
        """Garde seulement le domaine de l'email."""
        if "@" in email:
            domain = email.split("@")[1]
            return f"***@{domain}"
        return "***REDACTED_EMAIL***"

    @staticmethod
    def redact_card(card_number: str) -> str:
        """Garde seulement les 4 derniers chiffres."""
        cleaned = re.sub(r'\D', '', card_number)
        if len(cleaned) >= 4:
            return f"****-****-****-{cleaned[-4:]}"
        return "****REDACTED_CARD****"

    @staticmethod
    def redact_name(name: str) -> str:
        """Remplace par une valeur redactée."""
        return "***REDACTED_NAME***"

    @staticmethod
    def redact_address(address: str) -> str:
        """Remplace par une valeur redactée."""
        return "***REDACTED_ADDRESS***"

    @classmethod
    def redact_order_log(cls, order_data: dict) -> dict:
        """Redacte les PII d'un objet de commande avant logging."""
        safe_data = order_data.copy()

        if "customer_email" in safe_data:
            safe_data["customer_email"] = cls.redact_email(safe_data["customer_email"])

        if "customer_name" in safe_data:
            safe_data["customer_name"] = cls.redact_name(safe_data["customer_name"])

        if "payment_card" in safe_data:
            safe_data["payment_card"] = cls.redact_card(safe_data["payment_card"])

        if "shipping_address" in safe_data:
            safe_data["shipping_address"] = cls.redact_address(safe_data["shipping_address"])

        return safe_data

# Utilisation dans la saga de traitement de commande
redactor = PIIRedactor()

async def process_order_saga(order: dict):
    # Logger les informations de commande sans PII
    safe_order = redactor.redact_order_log(order)
    logger.info("order_processing_started", order=safe_order)

    # Traiter la commande (avec données complètes)
    result = await charge_customer(order)

    # Logger le résultat sans PII
    logger.info("order_processed", order_id=order["id"], status=result["status"])

Resulting log (without sensitive PII):

{
  "event": "order_processing_started",
  "order": {
    "id": "ORD-12345",
    "customer_email": "***@gmail.com",
    "customer_name": "***REDACTED_NAME***",
    "payment_card": "****-****-****-4242",
    "shipping_address": "***REDACTED_ADDRESS***",
    "amount": 150.00
  }
}

Summary:

• Redact before storing
• Encrypt across borders
• Treat model tool tokens as production secrets: scoped, rotated, and monitored

6.4 Policies and guardrails

Agents are powerful and sometimes too powerful to be safe by default. guardrails make them safe, explainable and auditable.

Types of guardrails

Content Filters

You don’t want just any user to send just any prompt to your agents. Your agents should be specialized and only do what they were designed to do. Two reasons:

1. LLM tokens are expensive if overused
2. Allowing everything is a security risk

content filters ensure that irrelevant queries are rejected.

from openai import OpenAI
import json

class ContentFilter:
    """Guardrail pour filtrer les requêtes inappropriées."""

    ALLOWED_INTENTS = [
        "incident_triage",
        "refund_approval",
        "fraud_detection",
        "monitoring_alert"
    ]

    def __init__(self, llm_client: OpenAI):
        self.client = llm_client

    async def classify_intent(self, user_input: str) -> dict:
        """Classifie l'intention de la requête."""
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",  # Modèle léger pour le filtrage
            messages=[{
                "role": "system",
                "content": f"""Classifie l'intention de cette requête.
                Intensions autorisées: {self.ALLOWED_INTENTS}
                Réponds en JSON: {{"intent": "...", "is_allowed": true/false, "confidence": 0.0-1.0}}"""
            }, {
                "role": "user",
                "content": user_input
            }],
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)

    async def filter(self, user_input: str) -> bool:
        """Retourne True si la requête est autorisée."""
        classification = await self.classify_intent(user_input)
        return (
            classification["is_allowed"] and
            classification["confidence"] > 0.8
        )

Approval Gates

If you deal with reimbursements, you don’t want your team to be bombarded with approval requests for all amounts. For example, only involve a human if the amount is greater than $200 — anything lower is processed automatically.

APPROVAL_THRESHOLD = 200.0  # $200

async def refund_processor_with_guardrail(refund_request: dict, context):
    amount = refund_request["amount"]

    # Guardrail : approval gate basé sur le montant
    if amount <= APPROVAL_THRESHOLD:
        # Remboursement automatique — aucun humain n'est dérangé
        result = await process_refund_automatically(refund_request)

        # Decision log : enregistrer la décision automatique
        await log_decision({
            "decision_type": "auto_approved",
            "reason": f"Amount {amount} <= threshold {APPROVAL_THRESHOLD}",
            "refund_id": refund_request["id"],
            "amount": amount,
            "timestamp": datetime.utcnow().isoformat()
        })

        return result
    else:
        # Montant élevé : déclencher human-in-the-loop
        await log_decision({
            "decision_type": "human_approval_required",
            "reason": f"Amount {amount} > threshold {APPROVAL_THRESHOLD}",
            "refund_id": refund_request["id"],
            "amount": amount
        })

        return await trigger_human_approval_workflow(refund_request, context)

Decision Logs

All types of guardrails must be accompanied by decision logs which record:

• What decision was made
• Why it was taken
• When

These logs are very useful for:

• Verify guardrails are working as expected
• Identify if some prompts can still bypass your guardrails
• Check if your guardrails are too strict and preventing your agents from doing their normal work

async def log_decision(decision: dict):
    """Enregistre une décision de guardrail dans le decision log."""
    decision_record = {
        "id": str(uuid.uuid4()),
        "timestamp": datetime.utcnow().isoformat(),
        "guardrail_type": decision.get("decision_type"),
        "reason": decision.get("reason"),
        "context": decision,
        "agent_version": os.environ.get("AGENT_VERSION", "unknown")
    }

    # Stocker dans une collection dédiée (non supprimable pour l'audit)
    await db.decision_logs.insert_one(decision_record)

    # Logger aussi pour Application Insights
    logger.info(
        "guardrail_decision",
        decision_id=decision_record["id"],
        guardrail_type=decision_record["guardrail_type"],
        reason=decision_record["reason"]
    )

Guardrail Summary

Filter borderline cases, require approvals when the risk increases, write a decision log each time to be able to prove what happened.

7. General Summary

Overall architecture of an event-driven agent system

graph TD
    subgraph Sources d'events
        W[Webhooks]
        IoT[Capteurs IoT]
        API[APIs / Polling]
        SVC[Microservices]
    end

    subgraph Infrastructure d'events
        EB[Event Bus<br/>Kafka / Event Hubs]
        PB[Pub/Sub<br/>Event Grid / NATS]
        Q[Queues<br/>Service Bus / Redis]
    end

    subgraph Agents IA
        TA[Agent Triage]
        FA[Agent Fraude]
        AA[Agent Approbation]
        RA[Agent RAG Refresh]
    end

    subgraph Orchestration
        DO[Durable Orchestrator<br/>Azure Durable Functions]
        HITL[Human-in-the-Loop]
    end

    subgraph Mémoire
        VM[Mémoire Vectorielle<br/>Semantic Memory]
        EM[Mémoire Épisodique<br/>Effect Log / State]
    end

    subgraph Observabilité
        AI[Application Insights]
        LOG[Logs + Traces]
        MET[Métriques + Alertes]
    end

    W --> PB
    IoT --> EB
    API --> Q
    SVC --> EB

    EB --> TA
    PB --> FA
    Q --> AA
    EB --> RA

    TA --> DO
    AA --> DO
    DO --> HITL

    TA <--> VM
    TA <--> EM

    TA --> LOG
    FA --> MET
    DO --> AI

Summary table of patterns

Tools per need

Search Terms

event-driven · agentic · ai · agents · orchestration · artificial · intelligence · generative · event · azure · agent · logs · configuration · driven · durable · errors · functions · grid · pattern · bus · design · guardrails · hubs · human-in-the-loop