Se lavori nell’infrastruttura enterprise nel 2026, avrai notato che gli AI agent non sono pi\u00f9 chatbot intelligenti: sono sistemi autonomi che pianificano task, orchestrano workflow multi-step e invocano API senza supervisione umana costante. Nella mia esperienza, il passaggio da \u00abchatbot che risponde\u00bb a \u00abagent che agisce autonomamente\u00bb cambia completamente il paradigma di deployment, governance e monitoring.<\/p>\n
Il problema centrale che affronto nei miei progetti \u00e8 questo: i singoli agent sono ormai affidabili, ma coordinarli senza fallimenti \u00e8 come orchestrare microservizi distribuiti<\/strong>. Gli agent devono comunicare, gestire stato condiviso, gestire fallimenti, mantenere budget di token e API, e rispettare policy di governance in tempo reale. Senza un’architettura corretta, il 86\u201389% dei pilot agentic fallisce prima di raggiungere production.<\/p>\n In questo articolo vi mostro come ho implementato orchestrazione multi-agent production-ready, tool integration sicura, model chaining robusto e monitoring che cattura ogni decision loop. Parler\u00f2 di framework, architetture, code reale e come evitare i failure mode pi\u00f9 comuni che ho visto fallare nei data center europei.<\/p>\n Gartner predice che il 40% delle applicazioni enterprise conterr\u00e0 AI agent task-specific entro la fine del 2026, rispetto a meno del 5% nel 2025<\/cite>. Ma il trend \u00e8 chiaro: il settore agentic AI si sta spostando da prototipi sperimentali a sistemi autonomi production-ready<\/cite>.<\/p>\n Ho osservato direttamente questa transizione nei miei clienti financial services e manufacturing: gli agent isolati funzionano bene per task single-goal, ma le operazioni reali\u2014supply chain orchestration, customer support multi-tier, financial auditing automatizzato\u2014richiedono teams di agent specializzati che si coordinano<\/strong>.<\/p>\n Come i monoliti hanno ceduto a microservizi, cos\u00ec gli agent single-purpose vengono rimpiazzati da team orchestrati di agent specializzati<\/cite>. Questo significa:<\/p>\n Questo \u00e8 quello che ho dovuto costruire per i miei clienti quando i loro AI agent hanno iniziato a fare operazioni critiche.<\/p>\n Ho strutturato ogni deployment di orchestrazione agentica in tre layer chiaramente separati:<\/p>\n Questo \u00e8 il decision engine<\/strong> della vostra architettura. Le funzioni core che un sistema di orchestrazione production-grade deve fornire sono: decomposizione di goal (spezzare un obiettivo di alto livello in sub-task distribuiti tra agent specializzati), gestione dello stato (mantenere il contesto condiviso e la cronologia di esecuzione che consente agli agent di coordinarsi senza processing ridondante), comunicazione inter-agent (canali autenticati e strutturati attraverso cui gli agent passano contesto, istruzioni e output), gestione dei fallimenti (rilevare fallimenti e eseguire recovery path definiti), e strumentazione di governance<\/cite>.<\/p>\n Nel mio deployment attuale ho usato LangGraph<\/strong> come orchestrator di riferimento. Ecco come ho strutturato il graph:<\/p>\n La key insight qui \u00e8 che ogni nodo \u00e8 atomico e idempotente<\/strong>. Se un agent fallisce, posso riprovare senza effetti collaterali perch\u00e9 lo stato \u00e8 immutabile e versionato.<\/p>\n Gli agent devono essere built con ruoli e confini chiaramente definiti\u2014cosa possono fare, a quali tool possono accedere, quali decisioni possono prendere autonomamente. Inoltre richiedono protocolli di comunicazione standardizzati come il Model Context Protocol (MCP) per l’accesso ai tool e il protocollo Agent-to-Agent (A2A) per la coordinazione peer<\/cite>.<\/p>\n Nel mio setup, ho implementato ogni specialist agent con accesso controllato ai tool via MCP. Ecco la struttura:<\/p>\n La struttura con MCPToolSpec garantisce che ogni agent conosce esattamente quali tool pu\u00f2 usare, con quali rate limit, e quale timeout. Se un agent cerca di chiamare un tool non autorizzato, fallisce gracefully e logga la violazione.<\/p>\n Qui va tutto quello che mantiene il sistema funzionante in production:<\/p>\n Ho strutturato il state management cos\u00ec:<\/p>\n Questo setup consente di:<\/p>\n I pattern di orchestrazione sono le regole che governano come gli agent si passano il lavoro, e scegliere il pattern sbagliato per il vostro workload \u00e8 un failure point pi\u00f9 comune che scegliere il modello sbagliato. Sei pattern coprono quasi ogni sistema production costruito nel 2026<\/cite>.<\/p>\n Nel mio deployment uso tre pattern principali:<\/p>\n Gli agent eseguono in una catena fissa e deterministica, ogni agent consuma l’output dell’agent precedente<\/cite>.<\/p>\n Caso d’uso: elaborazione di ticket di customer support dove ricerca \u2192 validazione \u2192 esecuzione \u2192 follow-up deve avvenire in ordine.<\/p>\n Quando pi\u00f9 agent possono lavorare in parallelo su sub-task indipendenti, uso fan-out. Esempio: ricerca simultanea su 3 source diverse.<\/p>\n Quando serve un agent orchestrator che supervisiona sub-agent specializzati. Google ADK usa una struttura gerarchica ad albero dove un orchestrator radice delega task a sub-agent, e ogni sub-agent pu\u00f2 coordinare agent di livello inferiore. Ogni agent ha un solo parent, mantenendo il command flow e data flow pi\u00f9 facili da seguire<\/cite>.<\/p>\n Ho imparato a mie spese che monitorare agent autonomi \u00e8 diverso dal monitorare LLM semplici<\/strong>. Non puoi limitarti a tracciare request-response: devi catturare l’intera decision chain, i tool call, le violazioni di policy e i failure mode.<\/p>\n Uno strumento di monitoring di AI agent fornisce visibilit\u00e0 e controllo sui workflow autonomi multi-step in production. A differenza della observability LLM standard che traccia cicli singoli request-response, il monitoring di agent cattura session-level execution<\/cite>.<\/p>\n I monitoring delivery capabilities critiche per team in production sono: distributed tracing che cattura ogni operazione nei workflow multi-agent, incluse le LLM call, tool invocation, e comunicazioni inter-agent<\/cite>.<\/p>\n Ecco come ho implementato il monitoring layer:<\/p>\n Ho integrato questo system con Datadog<\/strong> (per infrastructure) ma funziona anche con Jaeger, Grafana Loki, o altri backend OpenTelemetry-compatibili.<\/p>\n Nel primo anno di deployment, ho assistito a vari failure mode che ora monitoro specificamente:<\/p>\n Un agent genera informazione falsa, l’agent successivo la assume come vera e compie azioni basate su falsi dato.<\/p>\n Mitigation<\/strong>: Validation agent obbligatorio tra ogni coppia di agent, con fact-checking contro knowledge base.<\/p>\n Conversation history diventa troppo lunga, token si esauriscono, agent non ha contesto sufficiente.<\/p>\n Mitigation<\/strong>: Automatic summarization ogni 50 message, rollup to Qdrant, mantenere solo ultimi 10 message in context.<\/p>\n Agent rimane intrappolato in loop di retry\/retry senza fare progresso.<\/p>\n Mitigation<\/strong>: Max iteration counter (es. max 5 tentativi per task), timeout globale su execution.<\/p>\n Agent chiama tool con parametri sbagliati, corrotto database, o triggera rate limit.<\/p>\n Mitigation<\/strong>: Schema validation rigida, parameter sanitization, circuit breaker su rate limit.<\/p>\n Un agent scade il timeout, blocca il supervisor che scade, che fa scadere l’orchestrator.<\/p>\n Mitigation<\/strong>: Per-agent timeout (30-60s), per-workflow timeout (300-600s), per-task timeout con fallback immediato.<\/p>\n L’AI agentico espande i requisiti di governance perch\u00e9 gli agent non solo generano linguaggio\u2014pianificano, invocano tool, chiamano API, coordinano workflow e attivano azioni operative. Il problema di governance si sposta da review di output a controllo di azione. Il controllo di esecuzione agentica dovrebbe includere allowlist e denylist di tool, regole di autorit\u00e0 delegata, checkpoint di approvazione, limiti di budget o transazione, constraint di contesto, provenance di azione, meccanismi di interrupt, path di escalation, procedure di rollback e audit trail post-azione<\/cite>.<\/p>\n Ho implementato questo come policy engine centralistico:<\/p>\n Nel mio deployment: Ho implementato una Kafka queue di approval con SLA tracking. Quando un agent fa un’azione che richiede approval: Ogni LLM call viene tracciata con tokens count e costo stimato. Al livello di execution_context, sommo i costi di tutti gli agent e genero un report. Per multi-tenant scenarios, attribuo i costi al tenant specifico tramite tag nel traccing. Ho anche implementato budget limit per workflow e agent\u2014se il budget si esaurisce, il workflow fallisce gracefully.<\/p>\n Ho visto questo succedere quando due agent specializzati arrivano a conclusioni diverse. La soluzione \u00e8 un evaluator agent<\/strong> che media le decisioni basato su evidence. Se still discrepancy, escalation a human.<\/p>\n Max iteration counter (5-10 max tentativi per task), timeout globale su execution (5-10 minuti), e circuit breaker su tool call (se uno stesso tool fallisce 3 volte, stop). Se un agent entra in loop, fallisce gracefully con messaggio chiaro.<\/p>\n Il settore agentic AI si sta muovendo da prototipi sperimentali a sistemi autonomi production-ready<\/cite>. Nel 2026, orchestrare team di AI agent non \u00e8 pi\u00f9 science fiction\u2014\u00e8 requirement aziendale. Ho visto sistemi Salesforce Agentforce gestire migliaia di AI agent in production con riduzioni dell’84% nei tempi di risoluzione case e risparmi annuali superiori ai $100 milioni<\/cite>.<\/p>\n Ma il gap tra pilot e production rimane enorme: solo l’11-14% dei pilot agentic enterprise ha raggiunto production at scale, mentre l’86-89% non realizza valore durevole<\/cite>. La differenza? Architettura, governance, monitoring e failure recovery sono costruiti dall’inizio, non aggiunti dopo.<\/p>\n Se state iniziando il vostro journey agentico, partite da:<\/p>\nPerch\u00e9 l’orchestrazione agentica \u00e8 diventata essenziale nel 2026<\/h2>\n
\n
L’architettura production di orchestrazione agentica: tre layer<\/h2>\n
Layer 1: Orchestration\/Coordination<\/h3>\n
from langgraph.graph import StateGraph, START, END\nfrom typing import TypedDict\n\nclass AgentState(TypedDict):\n user_goal: str\n sub_tasks: list[dict]\n current_agent: str\n execution_history: list\n shared_context: dict\n policy_violations: list\n\ndef orchestrator_node(state: AgentState) -> AgentState:\n \"\"\"Decompose goal into sub-tasks and route to specialist agents.\"\"\"\n # LLM chiamato con constraint di policy\n tasks = call_llm_with_governance(\n prompt=f\"Decompose: {state['user_goal']}\",\n constraints=load_policy_rules(),\n budget_remaining=state.get('token_budget', 50000)\n )\n state['sub_tasks'] = tasks\n # Route al primo specialist\n state['current_agent'] = tasks[0]['assigned_agent']\n return state\n\ndef route_to_specialist(state: AgentState):\n \"\"\"Route a specialist agent basato su task type.\"\"\"\n task_type = state['sub_tasks'][0]['type']\n routing_map = {\n 'research': 'researcher_agent',\n 'validation': 'validator_agent',\n 'execution': 'executor_agent'\n }\n return routing_map.get(task_type, 'researcher_agent')\n\n# Build graph\nworkflow = StateGraph(AgentState)\n\n# Nodi\nworkflow.add_node(\"orchestrator\", orchestrator_node)\nworkflow.add_node(\"researcher_agent\", researcher_agent_executor)\nworkflow.add_node(\"validator_agent\", validator_agent_executor)\nworkflow.add_node(\"executor_agent\", executor_agent_executor)\nworkflow.add_node(\"failure_handler\", handle_agent_failure)\n\n# Edge con conditional routing\nworkflow.add_edge(START, \"orchestrator\")\nworkflow.add_conditional_edges(\n \"orchestrator\",\n route_to_specialist,\n {\n \"researcher_agent\": \"researcher_agent\",\n \"validator_agent\": \"validator_agent\",\n \"executor_agent\": \"executor_agent\"\n }\n)\n\n# Fallback su failure\nworkflow.add_edge(\"researcher_agent\", \"validator_agent\")\nworkflow.add_conditional_edges(\n \"validator_agent\",\n lambda state: \"executor_agent\" if state.get('validation_passed') else \"failure_handler\",\n {\"executor_agent\": \"executor_agent\", \"failure_handler\": \"failure_handler\"}\n)\nworkflow.add_edge(\"executor_agent\", END)\nworkflow.add_edge(\"failure_handler\", END)\n\nagent = workflow.compile()\n<\/code><\/pre>\nLayer 2: Specialist Agents con MCP Tool Access<\/h3>\n
from typing import Any\nfrom pydantic import BaseModel\n\nclass MCPToolSpec(BaseModel):\n name: str\n description: str\n input_schema: dict\n required_permissions: list[str] # \"read:database\", \"call:external_api\"\n rate_limit: int = 100 # calls per minute\n timeout_seconds: int = 30\n\nclass SpecialistAgent:\n def __init__(\n self,\n name: str,\n role: str,\n available_tools: list[MCPToolSpec],\n llm_model: str,\n execution_policy: dict\n ):\n self.name = name\n self.role = role\n self.available_tools = available_tools\n self.llm_model = llm_model\n self.execution_policy = execution_policy\n self.tool_call_log = [] # Per audit\n self.token_counter = 0\n\n def execute(\n self,\n task: dict,\n shared_context: dict,\n execution_constraints: dict\n ) -> dict:\n \"\"\"Execute task with tool access control.\"\"\"\n \n # Construct prompt with context e accessible tools\n system_prompt = f\"\"\"\nYou are a {self.role} agent.\nYour name: {self.name}\nYour goal: {task['goal']}\n\nAvailable tools (MCP-enabled):\n{json.dumps([{\"name\": t.name, \"description\": t.description} for t in self.available_tools], indent=2)}\n\nConstraints:\n- Total token budget: {execution_constraints.get('token_budget', 5000)}\n- Tool call budget: {execution_constraints.get('tool_call_budget', 20)}\n- Execution timeout: {execution_constraints.get('timeout_seconds', 300)}s\n- Policy rules: {json.dumps(self.execution_policy, indent=2)}\n\nCurrent shared context:\n{json.dumps(shared_context, indent=2)}\n\"\"\"\n \n # LLM call con tool_choice constraint\n response = llm_with_tools(\n model=self.llm_model,\n messages=[\n {\"role\": \"system\", \"content\": system_prompt},\n {\"role\": \"user\", \"content\": task['description']}\n ],\n tools=[\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": tool.name,\n \"description\": tool.description,\n \"parameters\": tool.input_schema\n }\n }\n for tool in self.available_tools\n ],\n tool_choice=\"auto\"\n )\n \n # Esegui tool calls con validation\n execution_result = {\n \"agent\": self.name,\n \"task\": task['id'],\n \"tool_calls\": [],\n \"reasoning\": response.content,\n \"success\": True\n }\n \n for tool_call in response.tool_calls:\n # Validate che il tool sia nella allowlist\n tool_spec = next(\n (t for t in self.available_tools if t.name == tool_call.function.name),\n None\n )\n \n if not tool_spec:\n # Tool non autorizzato\n execution_result['success'] = False\n execution_result['error'] = f\"Tool {tool_call.function.name} not in allowlist\"\n self._log_violation(\"unauthorized_tool_call\", tool_call.function.name)\n break\n \n # Validate permissions\n agent_permissions = self.execution_policy.get('allowed_tools', {})\n if tool_call.function.name not in agent_permissions:\n execution_result['success'] = False\n execution_result['error'] = f\"Permission denied: {tool_call.function.name}\"\n self._log_violation(\"insufficient_permissions\", tool_call.function.name)\n break\n \n # Execute tool con timeout\n try:\n tool_result = execute_mcp_tool(\n tool_name=tool_call.function.name,\n args=tool_call.function.arguments,\n timeout_seconds=tool_spec.timeout_seconds\n )\n \n execution_result['tool_calls'].append({\n \"tool\": tool_call.function.name,\n \"input\": tool_call.function.arguments,\n \"output\": tool_result,\n \"timestamp\": datetime.utcnow().isoformat()\n })\n \n except TimeoutError:\n execution_result['success'] = False\n execution_result['error'] = f\"Tool {tool_call.function.name} timeout\"\n break\n except Exception as e:\n execution_result['success'] = False\n execution_result['error'] = f\"Tool execution failed: {str(e)}\"\n break\n \n self.token_counter += count_tokens(response.content)\n return execution_result\n \n def _log_violation(self, violation_type: str, details: str):\n \"\"\"Log governance violation per audit.\"\"\"\n log_entry = {\n \"agent\": self.name,\n \"violation_type\": violation_type,\n \"details\": details,\n \"timestamp\": datetime.utcnow().isoformat()\n }\n self.tool_call_log.append(log_entry)\n # Push to centralized audit system\n send_to_siem(log_entry)\n<\/code><\/pre>\nLayer 3: Infrastructure\u2014Memory, State e Observability<\/h3>\n
\n
from redis import Redis\nfrom qdrant_client import QdrantClient\nfrom opentelemetry import trace\n\nclass AgentExecutionContext:\n def __init__(\n self,\n redis_client: Redis,\n qdrant_client: QdrantClient,\n tracer: trace.Tracer\n ):\n self.redis = redis_client # L1 cache\n self.qdrant = qdrant_client # L2 semantic memory\n self.tracer = tracer\n self.execution_id = None\n\n def initialize_execution(self, user_goal: str, metadata: dict):\n \"\"\"Init new execution session.\"\"\"\n self.execution_id = str(uuid.uuid4())\n \n # Initialize L1 state in Redis\n session_state = {\n \"execution_id\": self.execution_id,\n \"user_goal\": user_goal,\n \"start_time\": datetime.utcnow().isoformat(),\n \"agent_decisions\": [],\n \"tool_calls\": [],\n \"errors\": [],\n \"policy_checks\": [],\n \"metadata\": metadata\n }\n \n with self.tracer.start_as_current_span(\"execution_init\") as span:\n span.set_attribute(\"execution_id\", self.execution_id)\n span.set_attribute(\"user_goal\", user_goal)\n \n # Store in Redis with TTL (24h)\n self.redis.setex(\n f\"execution:{self.execution_id}\",\n 86400,\n json.dumps(session_state)\n )\n \n return self.execution_id\n \n def add_agent_decision(self, agent_name: str, decision: dict):\n \"\"\"Log agent decision to execution context.\"\"\"\n with self.tracer.start_as_current_span(\"agent_decision\") as span:\n span.set_attribute(\"agent\", agent_name)\n span.set_attribute(\"decision_type\", decision.get('type'))\n \n # Fetch current state\n state_json = self.redis.get(f\"execution:{self.execution_id}\")\n state = json.loads(state_json)\n \n # Append decision\n state['agent_decisions'].append({\n \"agent\": agent_name,\n \"decision\": decision,\n \"timestamp\": datetime.utcnow().isoformat()\n })\n \n # Update Redis\n self.redis.setex(\n f\"execution:{self.execution_id}\",\n 86400,\n json.dumps(state)\n )\n \n def add_semantic_context(self, key: str, text: str, metadata: dict):\n \"\"\"Index e retrieval su Qdrant per similarit\u00e0.\"\"\"\n with self.tracer.start_as_current_span(\"semantic_indexing\") as span:\n # Generate embedding\n embedding = generate_embedding(text)\n \n # Upsert a Qdrant\n self.qdrant.upsert(\n collection_name=\"agent_context\",\n points=[\n {\n \"id\": hash(key),\n \"vector\": embedding,\n \"payload\": {\n \"key\": key,\n \"text\": text,\n \"execution_id\": self.execution_id,\n \"metadata\": metadata,\n \"timestamp\": datetime.utcnow().isoformat()\n }\n }\n ]\n )\n \n def retrieve_similar_context(self, query: str, limit: int = 5) -> list:\n \"\"\"Semantic search su previous context.\"\"\"\n with self.tracer.start_as_current_span(\"semantic_retrieval\") as span:\n query_embedding = generate_embedding(query)\n \n results = self.qdrant.search(\n collection_name=\"agent_context\",\n query_vector=query_embedding,\n query_filter={\n \"must\": [\n {\"key\": \"execution_id\", \"match\": {\"value\": self.execution_id}}\n ]\n },\n limit=limit\n )\n \n return [r.payload for r in results]\n \n def get_execution_trace(self) -> dict:\n \"\"\"Fetch complete execution trace da Redis.\"\"\"\n state_json = self.redis.get(f\"execution:{self.execution_id}\")\n return json.loads(state_json)\n<\/code><\/pre>\n\n
Model Chaining: sequenziale vs parallelo vs gerarchico<\/h2>\n
1. Sequential (Pipeline)<\/h3>\n
def sequential_chain(\n agents: list[SpecialistAgent],\n initial_input: dict,\n execution_context: AgentExecutionContext\n) -> dict:\n \"\"\"Execute agents in fixed sequence.\"\"\"\n \n current_output = initial_input\n \n for agent in agents:\n with execution_context.tracer.start_as_current_span(f\"agent_{agent.name}\") as span:\n span.set_attribute(\"pattern\", \"sequential\")\n span.set_attribute(\"agent\", agent.name)\n \n # Execute agent con output precedente\n result = agent.execute(\n task={\n 'id': execution_context.execution_id,\n 'goal': current_output.get('next_goal'),\n 'description': current_output.get('context')\n },\n shared_context=current_output.get('shared_context', {}),\n execution_constraints={\n 'token_budget': 10000,\n 'tool_call_budget': 10,\n 'timeout_seconds': 60\n }\n )\n \n if not result['success']:\n # Sequential failure = stop\n span.add_event(f\"Sequential pipeline failed at {agent.name}\")\n return {\n \"success\": False,\n \"failed_at\": agent.name,\n \"error\": result['error'],\n \"execution_id\": execution_context.execution_id\n }\n \n # Pass output to next agent\n current_output = {\n \"context\": result.get('reasoning'),\n \"tool_results\": result.get('tool_calls'),\n \"next_goal\": result.get('next_task'),\n \"shared_context\": result.get('shared_context', {})\n }\n \n # Log decision\n execution_context.add_agent_decision(agent.name, result)\n \n return {\n \"success\": True,\n \"final_output\": current_output,\n \"execution_id\": execution_context.execution_id\n }\n<\/code><\/pre>\n2. Parallel Fan-Out<\/h3>\n
import asyncio\n\nasync def parallel_fan_out(\n agents: list[SpecialistAgent],\n tasks: list[dict],\n execution_context: AgentExecutionContext\n) -> dict:\n \"\"\"Execute agents in parallel, synchronize results.\"\"\"\n \n if len(agents) != len(tasks):\n raise ValueError(\"agents e tasks devono avere uguale lunghezza\")\n \n # Launch all agents in parallel\n async def execute_agent_task(agent: SpecialistAgent, task: dict):\n with execution_context.tracer.start_as_current_span(f\"parallel_{agent.name}\") as span:\n span.set_attribute(\"pattern\", \"parallel\")\n \n return agent.execute(\n task=task,\n shared_context={},\n execution_constraints={\n 'token_budget': 8000,\n 'tool_call_budget': 8,\n 'timeout_seconds': 45\n }\n )\n \n # Gather results\n results = await asyncio.gather(\n *[execute_agent_task(a, t) for a, t in zip(agents, tasks)],\n return_exceptions=True\n )\n \n # Check for failures\n failures = [r for r in results if isinstance(r, Exception) or not r.get('success')]\n if failures:\n return {\n \"success\": False,\n \"failed_count\": len(failures),\n \"errors\": [str(f) for f in failures],\n \"execution_id\": execution_context.execution_id\n }\n \n # Merge results\n merged_output = {\n \"success\": True,\n \"parallel_results\": results,\n \"execution_id\": execution_context.execution_id\n }\n \n # Log all decisions\n for agent, result in zip(agents, results):\n execution_context.add_agent_decision(agent.name, result)\n \n return merged_output\n<\/code><\/pre>\n3. Hierarchical Supervisor<\/h3>\n
class HierarchicalSupervisor:\n def __init__(\n self,\n coordinator_agent: SpecialistAgent,\n specialist_groups: dict, # {domain: [agents]}\n execution_context: AgentExecutionContext\n ):\n self.coordinator = coordinator_agent\n self.specialist_groups = specialist_groups\n self.execution_context = execution_context\n \n def execute(self, user_goal: str) -> dict:\n \"\"\"Hierarchical orchestration.\"\"\"\n \n with self.execution_context.tracer.start_as_current_span(\"hierarchical_orchestration\") as span:\n # Level 1: Coordinator decomposes goal\n decomposition = self.coordinator.execute(\n task={\n 'id': self.execution_context.execution_id,\n 'goal': f\"Decompose: {user_goal}\",\n 'description': \"Break down into domain-specific sub-tasks\"\n },\n shared_context={},\n execution_constraints={'token_budget': 5000, 'tool_call_budget': 5}\n )\n \n if not decomposition['success']:\n return {'success': False, 'error': decomposition['error']}\n \n # Parse sub-tasks from coordinator\n sub_tasks = parse_subtasks_from_reasoning(decomposition['reasoning'])\n \n # Level 2: Route to specialist groups\n group_results = {}\n \n for group_name, agents in self.specialist_groups.items():\n group_tasks = [t for t in sub_tasks if t.get('domain') == group_name]\n \n if not group_tasks:\n continue\n \n # Assign specialist agents\n assigned_agents = agents[:len(group_tasks)]\n \n # Execute in parallel within group\n group_result = asyncio.run(\n parallel_fan_out(\n agents=assigned_agents,\n tasks=group_tasks,\n execution_context=self.execution_context\n )\n )\n \n group_results[group_name] = group_result\n \n # Level 3: Synthesize results\n synthesis_result = self.coordinator.execute(\n task={\n 'id': self.execution_context.execution_id,\n 'goal': \"Synthesize results from all domains\",\n 'description': f\"Integrate: {json.dumps(group_results)}\"\n },\n shared_context=group_results,\n execution_constraints={'token_budget': 4000, 'tool_call_budget': 3}\n )\n \n return {\n \"success\": True,\n \"group_results\": group_results,\n \"final_output\": synthesis_result.get('reasoning'),\n \"execution_id\": self.execution_context.execution_id\n }\n<\/code><\/pre>\nExecution Monitoring: tracing, observability e failure detection<\/h2>\n
from opentelemetry import trace, metrics\nfrom opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter\nfrom opentelemetry.sdk.trace import TracerProvider\nfrom opentelemetry.sdk.trace.export import BatchSpanProcessor\n\nclass AgentMonitoringSystem:\n def __init__(self, otel_endpoint: str):\n # Setup OTLP exporter\n otlp_exporter = OTLPSpanExporter(endpoint=otel_endpoint)\n trace_provider = TracerProvider()\n trace_provider.add_span_processor(BatchSpanProcessor(otlp_exporter))\n trace.set_tracer_provider(trace_provider)\n \n self.tracer = trace.get_tracer(__name__)\n self.metrics_recorder = MetricsRecorder()\n \n def trace_llm_call(\n self,\n agent_name: str,\n model: str,\n messages: list,\n tools_available: list,\n response: dict\n ):\n \"\"\"Trace every LLM call with full context.\"\"\"\n \n with self.tracer.start_as_current_span(\"llm_call\") as span:\n span.set_attribute(\"agent\", agent_name)\n span.set_attribute(\"model\", model)\n span.set_attribute(\"num_messages\", len(messages))\n span.set_attribute(\"num_tools\", len(tools_available))\n span.set_attribute(\"input_tokens\", response.get('usage', {}).get('prompt_tokens', 0))\n span.set_attribute(\"output_tokens\", response.get('usage', {}).get('completion_tokens', 0))\n span.set_attribute(\"num_tool_calls\", len(response.get('tool_calls', [])))\n \n # Cost tracking\n total_tokens = (\n response.get('usage', {}).get('prompt_tokens', 0) +\n response.get('usage', {}).get('completion_tokens', 0)\n )\n span.set_attribute(\"total_tokens\", total_tokens)\n span.set_attribute(\"estimated_cost_usd\", self._calculate_cost(model, total_tokens))\n \n # Log messages (first 500 chars per message)\n span.add_event(\"llm_messages\", {\n \"messages_summary\": json.dumps([\n {\"role\": m['role'], \"content\": m.get('content', '')[:500]}\n for m in messages\n ])\n })\n \n def trace_tool_call(\n self,\n agent_name: str,\n tool_name: str,\n input_params: dict,\n result: dict,\n duration_ms: float,\n status: str = \"success\"\n ):\n \"\"\"Trace tool invocation e result.\"\"\"\n \n with self.tracer.start_as_current_span(\"tool_call\") as span:\n span.set_attribute(\"agent\", agent_name)\n span.set_attribute(\"tool\", tool_name)\n span.set_attribute(\"status\", status)\n span.set_attribute(\"duration_ms\", duration_ms)\n \n # Mask sensitive data in params\n safe_params = self._mask_sensitive_data(input_params)\n span.set_attribute(\"input_params\", json.dumps(safe_params))\n span.set_attribute(\"output_size_bytes\", len(json.dumps(result)))\n \n # Track metrics\n self.metrics_recorder.record(\n metric=\"tool_call_latency_ms\",\n value=duration_ms,\n attributes={\"tool\": tool_name, \"status\": status}\n )\n \n def trace_policy_check(\n self,\n agent_name: str,\n policy_name: str,\n check_type: str,\n result: bool,\n details: dict\n ):\n \"\"\"Trace policy evaluation.\"\"\"\n \n with self.tracer.start_as_current_span(\"policy_check\") as span:\n span.set_attribute(\"agent\", agent_name)\n span.set_attribute(\"policy\", policy_name)\n span.set_attribute(\"check_type\", check_type)\n span.set_attribute(\"passed\", result)\n \n if not result:\n span.set_attribute(\"violation_details\", json.dumps(details))\n span.add_event(\"policy_violation\", {\n \"severity\": details.get('severity', 'unknown')\n })\n \n def trace_failure(\n self,\n agent_name: str,\n error_type: str,\n error_message: str,\n recovery_action: str,\n traceback: str = None\n ):\n \"\"\"Trace failures con recovery.\"\"\"\n \n with self.tracer.start_as_current_span(\"agent_failure\") as span:\n span.set_attribute(\"agent\", agent_name)\n span.set_attribute(\"error_type\", error_type)\n span.set_attribute(\"error_message\", error_message)\n span.set_attribute(\"recovery_action\", recovery_action)\n \n if traceback:\n span.add_event(\"exception_details\", {\"traceback\": traceback})\n \n # Increment failure counter\n self.metrics_recorder.increment(\n metric=\"agent_failures\",\n attributes={\"agent\": agent_name, \"error_type\": error_type}\n )\n \n def get_execution_summary(self, execution_id: str) -> dict:\n \"\"\"Fetch complete execution summary.\"\"\"\n # Query observability backend (Jaeger, DataDog, etc.)\n traces = query_traces(\n filter=f\"execution_id={execution_id}\",\n backend=\"datadog\" # o Jaeger, Lightstep, etc.\n )\n \n return {\n \"execution_id\": execution_id,\n \"total_spans\": len(traces),\n \"llm_calls\": len([t for t in traces if t['span_name'] == 'llm_call']),\n \"tool_calls\": len([t for t in traces if t['span_name'] == 'tool_call']),\n \"failures\": len([t for t in traces if t['span_name'] == 'agent_failure']),\n \"policy_violations\": len([t for t in traces if t['attributes'].get('violation_details')]),\n \"total_tokens\": sum(int(t['attributes'].get('total_tokens', 0)) for t in traces),\n \"total_duration_ms\": traces[-1]['end_time'] - traces[0]['start_time'] if traces else 0,\n \"estimated_cost_usd\": sum(\n float(t['attributes'].get('estimated_cost_usd', 0))\n for t in traces\n )\n }\n \n def _calculate_cost(self, model: str, tokens: int) -> float:\n \"\"\"Estimate cost da model e token count.\"\"\"\n # Pricing per model (aggiorna con prezzi attuali)\n pricing = {\n \"gpt-4o\": {\"input\": 0.005, \"output\": 0.015},\n \"gpt-4-turbo\": {\"input\": 0.01, \"output\": 0.03},\n \"claude-opus\": {\"input\": 0.015, \"output\": 0.075},\n \"gemini-2.0-pro\": {\"input\": 0.0075, \"output\": 0.03}\n }\n rates = pricing.get(model, {\"input\": 0.005, \"output\": 0.015})\n # Rough estimate: 70% input, 30% output\n return (tokens * 0.7 * rates['input'] + tokens * 0.3 * rates['output']) \/ 1000\n \n def _mask_sensitive_data(self, params: dict) -> dict:\n \"\"\"Mask PII e sensitive data in tool params.\"\"\"\n sensitive_keys = ['password', 'api_key', 'token', 'secret', 'email', 'ssn', 'credit_card']\n masked = {}\n for key, value in params.items():\n if any(s in key.lower() for s in sensitive_keys):\n masked[key] = \"[REDACTED]\"\n else:\n masked[key] = value\n return masked\n<\/code><\/pre>\nFailure Modes e Recovery: cosa ho visto fallare in production<\/h2>\n
1. Hallucination Cascades<\/h3>\n
2. Context Overflow<\/h3>\n
3. Unbounded Loops<\/h3>\n
4. Tool Misuse<\/h3>\n
5. Cascading Timeouts<\/h3>\n
Governance e Compliance: il cuore della production agentic<\/h2>\n
class AgentPolicyEngine:\n def __init__(self, policy_store: dict):\n \"\"\"\n policy_store structure:\n {\n \"policies\": [\n {\n \"name\": \"financial_transfer\",\n \"condition\": \"tool == 'bank_api' AND action == 'transfer'\",\n \"rules\": [\n {\"rule\": \"amount 10000 AND amount 50000\", \"action\": \"deny\"}\n ]\n }\n ]\n }\n \"\"\"\n self.policies = policy_store\n self.approval_queue = []\n \n def evaluate_action(\n self,\n agent_name: str,\n tool_name: str,\n action: str,\n parameters: dict,\n execution_context: dict\n ) -> dict:\n \"\"\"Evaluate se un'azione \u00e8 allowed, denied, o richiede approval.\"\"\"\n \n # Find matching policies\n matching_policies = [\n p for p in self.policies['policies']\n if self._evaluate_condition(p['condition'], {\n 'agent': agent_name,\n 'tool': tool_name,\n 'action': action,\n 'params': parameters,\n 'context': execution_context\n })\n ]\n \n if not matching_policies:\n # No policy matched = allow per default\n return {'decision': 'allow', 'reason': 'no_matching_policy'}\n \n # Evaluate all matching policies\n decisions = []\n \n for policy in matching_policies:\n for rule in policy.get('rules', []):\n if self._evaluate_rule(\n rule['rule'],\n {'params': parameters, 'context': execution_context}\n ):\n decisions.append({\n 'policy': policy['name'],\n 'rule': rule['rule'],\n 'action': rule['action']\n })\n break\n \n # Aggregate decisions: deny > require_approval > allow\n if any(d['action'] == 'deny' for d in decisions):\n return {\n 'decision': 'deny',\n 'reason': 'policy_violation',\n 'violating_policies': [d for d in decisions if d['action'] == 'deny']\n }\n elif any(d['action'] == 'require_approval' for d in decisions):\n approval_id = str(uuid.uuid4())\n self.approval_queue.append({\n 'approval_id': approval_id,\n 'agent': agent_name,\n 'tool': tool_name,\n 'action': action,\n 'parameters': parameters,\n 'required_by_policies': [d for d in decisions if d['action'] == 'require_approval'],\n 'timestamp': datetime.utcnow().isoformat(),\n 'status': 'pending'\n })\n return {\n 'decision': 'pending_approval',\n 'approval_id': approval_id,\n 'policies_requiring_approval': [d for d in decisions if d['action'] == 'require_approval']\n }\n else:\n return {'decision': 'allow', 'reason': 'all_policies_passed'}\n \n def _evaluate_condition(self, condition: str, context: dict) -> bool:\n \"\"\"Evaluate condition string with context.\"\"\"\n # Simple expression evaluator (in production usa Python AST o safer option)\n # Example: \"tool == 'bank_api' AND action == 'transfer'\"\n try:\n # Replace variables\n expr = condition\n for var, val in context.items():\n if isinstance(val, str):\n expr = expr.replace(f\"{var}\", f\"'{val}'\")\n else:\n expr = expr.replace(f\"{var}\", str(val))\n \n # Evaluate (UNSAFE in production\u2014use safer evaluator)\n return eval(expr, {\"__builtins__\": {}}, {})\n except Exception as e:\n # Log error and default to deny\n log_error(f\"Policy condition evaluation failed: {e}\")\n return False\n \n def _evaluate_rule(self, rule: str, context: dict) -> bool:\n \"\"\"Evaluate rule condition.\"\"\"\n # Similar to _evaluate_condition but with parameter context\n try:\n # Replace variables from params e context\n expr = rule\n for param, val in context.get('params', {}).items():\n if isinstance(val, (int, float)):\n expr = expr.replace(param, str(val))\n elif isinstance(val, str):\n expr = expr.replace(param, f\"'{val}'\")\n \n return eval(expr, {\"__builtins__\": {}}, {})\n except Exception as e:\n log_error(f\"Policy rule evaluation failed: {e}\")\n return False\n \n def approve_pending(\n self,\n approval_id: str,\n approver: str,\n decision: str, # 'approved' o 'rejected'\n notes: str = None\n ) -> dict:\n \"\"\"Handle approval workflow.\"\"\"\n \n approval = next(\n (a for a in self.approval_queue if a['approval_id'] == approval_id),\n None\n )\n \n if not approval:\n return {'success': False, 'error': 'approval_id_not_found'}\n \n # Update approval status\n approval['status'] = decision\n approval['approver'] = approver\n approval['approval_timestamp'] = datetime.utcnow().isoformat()\n approval['notes'] = notes\n \n # Log to audit\n log_approval(approval)\n \n return {\n 'success': True,\n 'approval_id': approval_id,\n 'decision': decision\n }\n<\/code><\/pre>\nFAQ<\/h2>\n
Come scelgo tra LangGraph, CrewAI e OpenAI Agents SDK?<\/h3>\n
\n– LangGraph<\/strong> quando ho workflow complessi con loops e conditional routing espliciti. Ha il miglior support per checkpointing e recovery.
\n– CrewAI<\/strong> quando voglio rapid prototyping e orchestrazione role-based. Attenzione al token overhead\u2014prepende role\/goal a ogni call.
\n– OpenAI Agents SDK<\/strong> quando target esclusivamente OpenAI models e voglio semplicit\u00e0. Ha native handoff e guardrail.
\nLangGraph guida per workflow auditable in production; CrewAI guida per speed di prototyping; AutoGen\/AG2 guida per task iterative e conversazionali; cloud platforms (AWS AgentCore, Google ADK, Microsoft Agent Framework) sempre pi\u00f9 spesso offrono runtime managed agnostic-to-framework su top di tutti e tre<\/cite>.<\/p>\nCome gestisco l’escalation quando un agent richiede human approval?<\/h3>\n
\n1. L’agent mette in pausa l’esecuzione (non chiama il tool)
\n2. Spinge il request in Kafka topic “pending_approvals”
\n3. Dashboard umano mostra request con context completo
\n4. Approver approva o rifiuta
\n5. Agent continua o fallback a azione alternativa
\nTutto \u00e8 auditable e loggato a fini compliance.<\/p>\nCome monitoro i costi token se ogni workflow ha LLM call multiple?<\/h3>\n
Come gestisco agent che litigano (contradictory decisions)?<\/h3>\n
Come assicuro che gli agent non creino loop infiniti?<\/h3>\n
Conclusione: il presente della produzione agentica \u00e8 qui<\/h2>\n
\n