Agent de support 100 % AWS : Bedrock, Knowledge Bases, Guardrails & AgentCore

Provisionner la Knowledge Base

On crée une Knowledge Base pointant vers un bucket S3 (la documentation), avec OpenSearch Serverless en backend. Cela se fait en IaC (CDK/Terraform) ou en console ; on retient l’knowledgeBaseId. L’ingestion, le chunking, les embeddings et l’index hybride sont gérés par AWS — c’est tout l’intérêt.

# La Knowledge Base est provisionnée hors application (IaC). On ne garde que son id.
KB_ID = "KB123456"        # identifiant de la Knowledge Base
REGION = "eu-west-1"

Exposer la récupération managée comme outil

On branche la Knowledge Base via AmazonKnowledgeBasesRetriever, en mode hybride, et on l’expose comme outil — l’agent décide quand chercher.

from langchain_aws import AmazonKnowledgeBasesRetriever
from langchain_core.tools import tool

retriever = AmazonKnowledgeBasesRetriever(
    knowledge_base_id=KB_ID, region_name=REGION,
    retrieval_config={"vectorSearchConfiguration": {
        "numberOfResults": 5, "overrideSearchType": "HYBRID"}},
)

@tool
def base_de_connaissances(requete: str) -> str:
    """Recherche dans la base de connaissances produit (Bedrock Knowledge Base)."""
    docs = retriever.invoke(requete)
    if not docs:
        return "AUCUN_RESULTAT"
    return "\n\n---\n\n".join(
        f"[{d.metadata.get('source_metadata', {}).get('x-amz-bedrock-kb-source-uri', 'doc')}]\n"
        f"{d.page_content}" for d in docs)

Le modèle Bedrock, protégé par un Guardrail

Le raisonnement est confié à Claude via la Converse API, avec un Guardrail attaché : PII caviardée, sujets interdits, vérification d’ancrage — appliqués à chaque appel (cf. Guardrails en profondeur).

from langchain_aws import ChatBedrockConverse

modele = ChatBedrockConverse(
    model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name=REGION,
    temperature=0.1,
    guardrail_config={"guardrailIdentifier": "abcd1234",
                      "guardrailVersion": "1", "trace": "enabled"},
)

Câbler l’escalade

Comme dans le cas d’origine, « escalader » est un outil concret — ici, il crée un ticket (par exemple via une Lambda exposée).

@tool
def escalader(motif: str, resume: str) -> str:
    """Transfère la demande à un humain (hors base, réclamation, action sur le compte)."""
    ticket_id = creer_ticket_helpdesk(motif, resume)   # Lambda / API interne
    return f"Demande transférée à un conseiller (ticket {ticket_id})."

Assembler l’agent avec persistance DynamoDB

On assemble l’agent ReAct et on lui donne un checkpointer DynamoDB : l’état survit aux redéploiements, et le human-in-the-loop devient possible (cf. archi de production).

from langgraph.prebuilt import create_react_agent
from langgraph_checkpoint_aws.saver import BedrockSessionSaver

SYSTEME = """Tu es l'assistant de support de niveau 1 de DEEP-5.
- Réponds UNIQUEMENT à partir des documents de `base_de_connaissances`, cite la source.
- Hors périmètre, réclamation ou action sur le compte : appelle `escalader`.
- N'invente jamais. Réponds en français, concis et actionnable."""

checkpointer = BedrockSessionSaver(region_name=REGION)
agent = create_react_agent(
    modele, [base_de_connaissances, escalader],
    prompt=SYSTEME, checkpointer=checkpointer,
)

Déployer sur AgentCore Runtime

On emballe l’agent derrière le contrat d’entrée d’AgentCore Runtime : exécution serverless, isolée, à sessions longues, sans gérer de conteneurs.

from langfuse.langchain import CallbackHandler
_handler = CallbackHandler()

def handler(requete: dict) -> dict:
    """Point d'entrée invoqué par AgentCore Runtime pour chaque requête."""
    res = agent.invoke(
        {"messages": [{"role": "user", "content": requete["prompt"]}]},
        config={
            "callbacks": [_handler],
            "configurable": {"thread_id": requete["session_id"]},
            "recursion_limit": 8,
            "metadata": {"langfuse_session_id": requete["session_id"],
                         "langfuse_tags": ["support-aws"]},
        },
    )
    return {"reponse": res["messages"][-1].content}