Sviluppo Agenti AI con Amazon Bedrock e Claude

Sviluppo Agenti AI con Amazon Bedrock e Claude: Guida Tecnica per Sviluppatori

Introduzione: L’Era degli Agenti AI Autonomi

L’intelligenza artificiale generativa sta trasformando radicalmente il modo in cui le aziende IT operano e competono sul mercato. Non parliamo più solo di chatbot che rispondono a domande predefinite, ma di agenti AI autonomi capaci di ragionare, pianificare e agire per raggiungere obiettivi complessi. Secondo recenti analisi di settore, il 70% dei CEO globali prevede che l’AI generativa trasformerà la creazione di valore nei prossimi tre anni, mentre le aziende che integrano l’AI con i propri dati aziendali registrano già aumenti di fatturato del 51% e miglioramenti dei profitti del 49%.

Amazon Bedrock rappresenta la piattaforma serverless ideale per sviluppare queste applicazioni avanzate, offrendo accesso ai modelli Claude di Anthropic (tra i più potenti per coding, reasoning e costruzione di agenti) attraverso un’unica API gestita. In questa guida tecnica, esploreremo come costruire agenti AI enterprise-ready integrando dati privati tramite RAG (Retrieval-Augmented Generation) e abilitando l’utilizzo di API esterne.

Architettura Serverless con Amazon Bedrock

Amazon Bedrock è un servizio fully managed che elimina la complessità dell’infrastruttura AI. Gli sviluppatori possono accedere ai modelli Claude (Sonnet 4.5, Opus 4.1, Haiku 4.5) senza dover gestire server, scaling o aggiornamenti.

Configurazione Iniziale con Python

import boto3
import json

# Inizializzazione del client Bedrock
bedrock_runtime = boto3.client(
    service_name='bedrock-runtime',
    region_name='us-east-1'
)

# Definizione del model ID per Claude Sonnet 4.5
MODEL_ID = "anthropic.claude-sonnet-4-5-20250929-v1:0"

def invoke_claude(prompt, max_tokens=1000):
    """Invoca Claude tramite Bedrock Runtime"""
    body = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": max_tokens,
        "messages": [
            {
                "role": "user",
                "content": prompt
            }
        ]
    })
    
    response = bedrock_runtime.invoke_model(
        modelId=MODEL_ID,
        body=body
    )
    
    response_body = json.loads(response['body'].read())
    return response_body['content'][0]['text']

# Esempio di utilizzo
result = invoke_claude("Analizza i rischi di sicurezza in questo codice Python")
print(result)

Utilizzo della Converse API

Per applicazioni più complesse con conversazioni multi-turno, Bedrock offre la Converse API che mantiene il contesto:

def converse_with_claude(messages, system_prompt=None):
    """Conversazione con Claude mantenendo il contesto"""
    request_params = {
        "modelId": MODEL_ID,
        "messages": messages
    }
    
    if system_prompt:
        request_params["system"] = [{"text": system_prompt}]
    
    response = bedrock_runtime.converse(**request_params)
    return response['output']['message']['content'][0]['text']

# Esempio conversazione
conversation = [
    {"role": "user", "content": [{"text": "Qual è la capitale della Francia?"}]},
    {"role": "assistant", "content": [{"text": "La capitale della Francia è Parigi."}]},
    {"role": "user", "content": [{"text": "Quanti abitanti ha?"}]}
]

response = converse_with_claude(
    messages=conversation,
    system_prompt="Sei un assistente geografico esperto"
)

Prompt Engineering: Massimizzare le Prestazioni di Claude – Sviluppo Agenti AI con Amazon Bedrock e Claude

I modelli Claude 4.x richiedono istruzioni precise e strutturate. Ecco le best practice fondamentali:

1. Chiarezza e Specificità

#  Prompt vago
prompt_vago = "Scrivi del codice"

#  Prompt specifico
prompt_specifico = """
Scrivi una funzione Python che:
1. Accetti una lista di dizionari con chiavi 'nome' e 'età'
2. Filtri le persone maggiorenni (età >= 18)
3. Ordini il risultato per età decrescente
4. Restituisca una lista di nomi

Includi docstring e type hints.
"""

2. Utilizzo di Tag XML

Claude risponde particolarmente bene ai tag XML per strutturare le istruzioni:

prompt_strutturato = """
<task>
Analizza il seguente log di errore e identifica la causa principale.
</task>

<log>
ERROR: Connection timeout after 30s
Failed to connect to database: postgresql://prod-db:5432
Last successful connection: 2025-11-06 14:23:45
</log>

<instructions>
1. Identifica il tipo di errore
2. Suggerisci 3 possibili cause
3. Proponi soluzioni concrete
</instructions>

<output_format>
Rispondi in formato JSON con campi: error_type, causes[], solutions[]
</output_format>
"""

3. Few-Shot Learning con Esempi

prompt_esempi = """
Classifica il sentiment delle recensioni clienti.

<examples>
<example>
Input: "Prodotto eccellente, consegna veloce!"
Output: {"sentiment": "positivo", "confidence": 0.95}
</example>

<example>
Input: "Qualità scadente, deluso dall'acquisto"
Output: {"sentiment": "negativo", "confidence": 0.92}
</example>
</examples>

<review>
"Il servizio clienti è stato disponibile ma il prodotto non soddisfa le aspettative"
</review>
"""

Retrieval-Augmented Generation (RAG) con Knowledge Bases – Sviluppo Agenti AI con Amazon Bedrock e Claude

Il RAG permette agli agenti AI di accedere a dati aziendali privati senza riaddestramento del modello. Amazon Bedrock Knowledge Bases gestisce automaticamente tutto il workflow.

Architettura RAG

1. Ingestione: I documenti vengono caricati da S3, Confluence, SharePoint
2. Chunking: Il testo viene suddiviso in blocchi semanticamente coerenti
3. Embedding: Ogni chunk viene convertito in vettori numerici
4. Indexing: I vettori sono salvati in un vector database (OpenSearch, Aurora, Pinecone)
5. Retrieval: Al runtime, la query viene convertita in vettore e confrontata con l’indice
6. Augmentation: I chunk rilevanti arricchiscono il prompt al modello

Implementazione Knowledge Base

import boto3

bedrock_agent = boto3.client('bedrock-agent', region_name='us-east-1')

# Creazione Knowledge Base
kb_response = bedrock_agent.create_knowledge_base(
    name='DocumentazioneAziendale',
    description='Knowledge base con policy e procedure aziendali',
    roleArn='arn:aws:iam::123456789:role/BedrockKBRole',
    knowledgeBaseConfiguration={
        'type': 'VECTOR',
        'vectorKnowledgeBaseConfiguration': {
            'embeddingModelArn': 'arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v1'
        }
    },
    storageConfiguration={
        'type': 'OPENSEARCH_SERVERLESS',
        'opensearchServerlessConfiguration': {
            'collectionArn': 'arn:aws:aoss:us-east-1:123456789:collection/kb-collection',
            'vectorIndexName': 'bedrock-knowledge-base-index',
            'fieldMapping': {
                'vectorField': 'embedding',
                'textField': 'text',
                'metadataField': 'metadata'
            }
        }
    }
)

kb_id = kb_response['knowledgeBase']['knowledgeBaseId']

Query con RAG

bedrock_agent_runtime = boto3.client('bedrock-agent-runtime')

def query_with_rag(question, kb_id):
    """Query la knowledge base e genera risposta con contesto"""
    response = bedrock_agent_runtime.retrieve_and_generate(
        input={'text': question},
        retrieveAndGenerateConfiguration={
            'type': 'KNOWLEDGE_BASE',
            'knowledgeBaseConfiguration': {
                'knowledgeBaseId': kb_id,
                'modelArn': f'arn:aws:bedrock:us-east-1::foundation-model/{MODEL_ID}',
                'retrievalConfiguration': {
                    'vectorSearchConfiguration': {
                        'numberOfResults': 5
                    }
                }
            }
        }
    )
    
    return {
        'answer': response['output']['text'],
        'sources': [citation['retrievedReferences'] 
                   for citation in response.get('citations', [])]
    }

# Esempio
result = query_with_rag(
    "Qual è la policy aziendale per il lavoro remoto?",
    kb_id
)
print(f"Risposta: {result['answer']}")
print(f"Fonti: {result['sources']}")

Function Calling: Connettere Agenti a Sistemi Esterni – Sviluppo Agenti AI con Amazon Bedrock e Claude

Il function calling (tool use) permette agli agenti di interagire con API, database e servizi esterni.

Definizione degli Strumenti

def define_tools():
    """Definisce gli strumenti disponibili per l'agente"""
    return [
        {
            "name": "get_weather",
            "description": "Ottiene le previsioni meteo per una città",
            "input_schema": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "Nome della città"
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Unità di temperatura"
                    }
                },
                "required": ["city"]
            }
        },
        {
            "name": "search_database",
            "description": "Cerca informazioni nel database aziendale",
            "input_schema": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Query SQL da eseguire"
                    },
                    "table": {
                        "type": "string",
                        "description": "Nome della tabella"
                    }
                },
                "required": ["query", "table"]
            }
        }
    ]

def execute_tool(tool_name, tool_input):
    """Esegue la funzione richiesta dall'agente"""
    if tool_name == "get_weather":
        # Simulazione chiamata API meteo
        return f"Temperatura a {tool_input['city']}: 22°C, soleggiato"
    
    elif tool_name == "search_database":
        # Simulazione query database
        return f"Risultati per {tool_input['query']}: 15 record trovati"
    
    return "Strumento non disponibile"

Loop Agente con Tool Use

def agent_loop(user_message):
    """Loop principale dell'agente con supporto tool use"""
    messages = [{"role": "user", "content": [{"text": user_message}]}]
    tools = define_tools()
    
    while True:
        response = bedrock_runtime.converse(
            modelId=MODEL_ID,
            messages=messages,
            toolConfig={'tools': tools}
        )
        
        stop_reason = response['stopReason']
        
        if stop_reason == 'end_turn':
            # L'agente ha completato il task
            final_text = response['output']['message']['content'][0]['text']
            return final_text
        
        elif stop_reason == 'tool_use':
            # L'agente richiede l'uso di uno strumento
            messages.append(response['output']['message'])
            
            tool_results = []
            for content_block in response['output']['message']['content']:
                if 'toolUse' in content_block:
                    tool_use = content_block['toolUse']
                    result = execute_tool(
                        tool_use['name'],
                        tool_use['input']
                    )
                    tool_results.append({
                        "toolResult": {
                            "toolUseId": tool_use['toolUseId'],
                            "content": [{"text": result}]
                        }
                    })
            
            # Invia i risultati al modello
            messages.append({
                "role": "user",
                "content": tool_results
            })

# Esempio utilizzo
response = agent_loop("Che tempo fa a Milano e cerca nel database i clienti attivi")
print(response)

Orchestrazione Multi-Agente

Per task complessi, è possibile coordinare più agenti specializzati tramite un supervisor agent.

Architettura Multi-Agente

# Configurazione agenti specializzati
SPECIALIST_AGENTS = {
    "data_analyst": {
        "model": MODEL_ID,
        "role": "Analizza dati e genera report statistici",
        "tools": ["query_database", "generate_chart"]
    },
    "content_writer": {
        "model": MODEL_ID,
        "role": "Crea contenuti marketing e comunicazioni",
        "tools": ["check_grammar", "optimize_seo"]
    },
    "code_reviewer": {
        "model": MODEL_ID,
        "role": "Revisiona codice per sicurezza e best practices",
        "tools": ["run_static_analysis", "check_vulnerabilities"]
    }
}

def supervisor_agent(user_request):
    """Agente supervisor che coordina gli specialisti"""
    
    # 1. Analizza la richiesta e determina quali agenti servono
    analysis_prompt = f"""
    <request>{user_request}</request>
    
    Analizza questa richiesta e determina quali agenti specializzati servono.
    Agenti disponibili: {list(SPECIALIST_AGENTS.keys())}
    
    Rispondi in JSON:
    {{
        "agents_needed": ["agent1", "agent2"],
        "execution_plan": "descrizione piano",
        "parallel_execution": true/false
    }}
    """
    
    plan = invoke_claude(analysis_prompt)
    plan_data = json.loads(plan)
    
    # 2. Delega task agli agenti specializzati
    results = {}
    for agent_name in plan_data['agents_needed']:
        agent_config = SPECIALIST_AGENTS[agent_name]
        
        agent_prompt = f"""
        {agent_config['role']}
        
        Task: {user_request}
        Esegui il tuo compito specifico.
        """
        
        results[agent_name] = invoke_claude(agent_prompt)
    
    # 3. Consolida i risultati
    consolidation_prompt = f"""
    Consolida i seguenti risultati in una risposta finale coerente:
    
    {json.dumps(results, indent=2)}
    
    Richiesta originale: {user_request}
    """
    
    return invoke_claude(consolidation_prompt)

# Esempio
final_result = supervisor_agent(
    "Analizza i dati di vendita Q4 e crea un report esecutivo con grafici"
)

Monitoring e Logging

Monitorare gli agenti AI è fondamentale per debugging, ottimizzazione e compliance.

CloudWatch Integration

import logging
from datetime import datetime

# Configurazione logging CloudWatch
logger = logging.getLogger('BedrockAgent')
logger.setLevel(logging.INFO)

def invoke_with_monitoring(prompt, metadata=None):
    """Invoca Claude con monitoring completo"""
    start_time = datetime.now()
    
    try:
        # Log richiesta
        logger.info({
            'event': 'agent_invocation_start',
            'timestamp': start_time.isoformat(),
            'prompt_length': len(prompt),
            'metadata': metadata
        })
        
        # Invocazione
        response = invoke_claude(prompt)
        
        # Calcola metriche
        duration = (datetime.now() - start_time).total_seconds()
        
        # Log successo
        logger.info({
            'event': 'agent_invocation_success',
            'duration_seconds': duration,
            'response_length': len(response),
            'tokens_estimated': len(response.split()) * 1.3
        })
        
        return response
        
    except Exception as e:
        # Log errore
        logger.error({
            'event': 'agent_invocation_error',
            'error': str(e),
            'duration_seconds': (datetime.now() - start_time).total_seconds()
        })
        raise

Trace Visualization

Per agenti complessi con tool use, Amazon Bedrock fornisce trace dettagliate:

def invoke_agent_with_trace(agent_id, session_id, prompt):
    """Invoca agente Bedrock con trace abilitato"""
    bedrock_agent_runtime = boto3.client('bedrock-agent-runtime')
    
    response = bedrock_agent_runtime.invoke_agent(
        agentId=agent_id,
        agentAliasId='LATEST',
        sessionId=session_id,
        inputText=prompt,
        enableTrace=True  # Abilita trace dettagliato
    )
    
    # Processa stream e trace
    traces = []
    output = ""
    
    for event in response['completion']:
        if 'chunk' in event:
            output += event['chunk']['bytes'].decode()
        
        if 'trace' in event:
            trace = event['trace']['trace']
            traces.append(trace)
            
            # Log trace specifici
            if 'orchestrationTrace' in trace:
                logger.info(f"Orchestration: {trace['orchestrationTrace']}")
            if 'preProcessingTrace' in trace:
                logger.info(f"PreProcessing: {trace['preProcessingTrace']}")
    
    return {
        'output': output,
        'traces': traces
    }

Caso Pratico: Agente di Customer Support Enterprise – Sviluppo Agenti AI con Amazon Bedrock e Claude

Implementiamo un agente completo per supporto clienti con:

• Accesso a knowledge base (documentazione prodotti)
• Tool use (gestione ticket, query CRM)
• Multi-agente (escalation a specialisti)

class CustomerSupportAgent:
    def __init__(self, kb_id, crm_api_key):
        self.kb_id = kb_id
        self.crm_api_key = crm_api_key
        self.bedrock = boto3.client('bedrock-agent-runtime')
    
    def handle_request(self, customer_query, customer_id):
        """Gestisce richiesta cliente end-to-end"""
        
        # 1. Classifica intento
        intent = self._classify_intent(customer_query)
        
        # 2. Recupera contesto cliente da CRM
        customer_context = self._get_customer_context(customer_id)
        
        # 3. Query knowledge base se necessario
        kb_context = ""
        if intent in ['technical_issue', 'product_info']:
            kb_response = self._query_knowledge_base(customer_query)
            kb_context = kb_response['answer']
        
        # 4. Genera risposta personalizzata
        response = self._generate_response(
            query=customer_query,
            intent=intent,
            customer_context=customer_context,
            kb_context=kb_context
        )
        
        # 5. Crea ticket se richiesto
        if intent == 'complaint' or 'urgente' in customer_query.lower():
            ticket_id = self._create_ticket(customer_id, customer_query, response)
            response += f"\n\nTicket #{ticket_id} creato per follow-up."
        
        return response
    
    def _classify_intent(self, query):
        """Classifica l'intento della richiesta"""
        prompt = f"""
        Classifica l'intento di questa richiesta cliente.
        Opzioni: technical_issue, product_info, complaint, billing, general
        
        Richiesta: {query}
        
        Rispondi solo con l'intento (una parola).
        """
        return invoke_claude(prompt).strip().lower()
    
    def _query_knowledge_base(self, query):
        """Query RAG knowledge base"""
        return query_with_rag(query, self.kb_id)
    
    def _get_customer_context(self, customer_id):
        """Recupera dati cliente da CRM (simulato)"""
        return {
            'tier': 'premium',
            'active_subscriptions': ['Product A', 'Product B'],
            'last_interaction': '2025-10-15',
            'satisfaction_score': 4.5
        }
    
    def _generate_response(self, query, intent, customer_context, kb_context):
        """Genera risposta personalizzata"""
        prompt = f"""
        Sei un agente di customer support professionale.
        
        <customer_context>
        Cliente: {customer_context['tier']} tier
        Prodotti attivi: {', '.join(customer_context['active_subscriptions'])}
        Soddisfazione: {customer_context['satisfaction_score']}/5
        </customer_context>
        
        <knowledge_base_info>
        {kb_context}
        </knowledge_base_info>
        
        <request>
        Tipo: {intent}
        Messaggio: {query}
        </request>
        
        Genera una risposta:
        1. Empatica e professionale
        2. Basata sulle informazioni della knowledge base
        3. Personalizzata per cliente {customer_context['tier']}
        4. Con passi concreti da seguire
        """
        return invoke_claude(prompt)
    
    def _create_ticket(self, customer_id, issue, initial_response):
        """Crea ticket nel sistema (simulato)"""
        ticket_id = f"TKT-{int(datetime.now().timestamp())}"
        logger.info(f"Ticket {ticket_id} creato per customer {customer_id}")
        return ticket_id

# Esempio utilizzo
agent = CustomerSupportAgent(
    kb_id='kb-123456',
    crm_api_key='crm-api-key-secret'
)

response = agent.handle_request(
    customer_query="Il mio Product A non si avvia dopo l'ultimo aggiornamento",
    customer_id="CUST-789"
)
print(response)

Perché Investire in Agenti AI Oggi – Sviluppo Agenti AI con Amazon Bedrock e Claude

Le aziende IT che adottano agenti AI ottengono vantaggi competitivi significativi:

1. Automazione Intelligente: Riduzione del 70% del tempo su task ripetitivi
2. Scalabilità: Gestione di volumi 10x superiori senza aumento di personale
3. Costi Ottimizzati: Architettura serverless con pricing pay-per-use
4. Time-to-Market: Sviluppo di MVP in giorni anziché mesi
5. Compliance e Sicurezza: Dati privati rimangono nel perimetro AWS
6. Adattabilità: Facile integrazione con sistemi legacy

Formazione Continua: Investire nel Team

Le tecnologie AI evolvono rapidamente. È fondamentale mantenere il team aggiornato attraverso formazione strutturata e certificazioni.

Innovaformazione eroga formazione specializzata per aziende IT con:

• Corsi Personalizzati: Calendario e contenuti adattati alle esigenze aziendali
• Modalità Online: Classi virtuali interattive con docenti certificati
• Finanziamenti Disponibili: Possibilità di finanziare la formazione tramite Fondimpresa o altri fondi interprofessionali
• Gestione Completa: Innovaformazione gestisce l’intero piano formativo finanziato

Corsi Disponibili

Corso Amazon Bedrock e Claude: Attivabile su richiesta, copre sviluppo agenti, RAG, orchestrazione multi-agente e deployment enterprise.

Per altri corsi su AI Generativa, visita:

Catalogo completo corsi aziendali QUI.

Contatti

Email: info@innovaformazione.net
Telefono: 3471012275 (Dario Carrassi)

Contattateci per un preventivo personalizzato e scoprite come possiamo supportare la crescita tecnica del vostro team con formazione finanziata.

Conclusioni – Sviluppo Agenti AI con Amazon Bedrock e Claude

Lo sviluppo di agenti AI con Amazon Bedrock e Claude rappresenta il futuro dell’automazione enterprise. Le architetture serverless, combinate con tecniche avanzate come RAG e orchestrazione multi-agente, permettono di costruire applicazioni intelligenti, scalabili e sicure.

Gli esempi di codice presentati forniscono una base solida per iniziare, ma la chiave del successo sta nell’apprendimento continuo e nell’iterazione rapida. Investire nella formazione del team è essenziale per restare competitivi in un mercato in rapida evoluzione.

Iniziate con piccoli proof-of-concept, misurate i risultati, e scalate gradualmente verso soluzioni enterprise. Le tecnologie sono mature, l’ecosistema è ricco, e le opportunità sono enormi.

(fonte) (fonte) (fonte) (fonte)