Tool use con LangChain: l'agente AI che usa strumenti

Il tool use è il passaggio che trasforma un LLM in un agente: invece di limitarsi a rispondere, il modello sceglie uno strumento, lo chiama, legge il risultato e decide il passo successivo. In questo capitolo costruisci gli strumenti (ricerca web, file, HTTP, shell, calcolatrice), colleghi un agente a LangChain e imposti i limiti di sicurezza.

Tool Use: da LLM ad Agente

┌─────────────────────────────────────────────────────────────────────────────┐
│                            AGENT LOOP (ReAct)                                │
│                                                                              │
│   ┌─────────┐     ┌─────────┐     ┌─────────┐     ┌─────────┐              │
│   │ Thought │────►│ Action  │────►│Observat.│────►│ Thought │  ...         │
│   │(Reason) │     │(Tool)   │     │(Result) │     │(Reason) │              │
│   └─────────┘     └─────────┘     └─────────┘     └─────────┘              │
│        ▲                                           │                        │
│        │                                           │                        │
│        └─────────────── Final Answer ◄──────────────┘                        │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

ReAct (Reasoning + Acting): il LLM alterna pensiero (catena di ragionamento) e azione (chiamata di un tool), osservando il risultato per decidere il passo successivo.

Function Calling (nativo): i modelli che lo supportano (per esempio Llama 3.1, Mistral, Qwen 2.5) restituiscono tool_calls strutturati, senza parsing del testo.

Dipendenze

# requirements.txt (aggiungi a Cap. 5)
langchain>=0.3,<1.0
langchain-community>=0.3,<1.0
langchain-experimental>=0.3,<0.4  # PythonREPLTool
duckduckgo-search>=5.0.0
wikipedia>=1.4.0
requests>=2.32.0
aiohttp>=3.9.0  # per async HTTP tool

pip install -r requirements.txt

1. Built-in Tools: Pronti all'uso

# builtin_tools.py
from langchain_community.tools import (
    DuckDuckGoSearchRun,
    ShellTool,
)
from langchain_experimental.tools import PythonREPLTool
from langchain_community.utilities import (
    DuckDuckGoSearchAPIWrapper,
    WikipediaAPIWrapper,
)
from langchain_core.tools import Tool
from typing import Dict, Any


# === Web Search ===
def get_duckduckgo_search(
    max_results: int = 5,
    region: str = "it-it",
    safesearch: str = "moderate",
) -> Tool:
    """DuckDuckGo search senza API key."""
    wrapper = DuckDuckGoSearchAPIWrapper(
        max_results=max_results,
        region=region,
        safesearch=safesearch,
    )
    return DuckDuckGoSearchRun(api_wrapper=wrapper)


# === Python REPL (Esecuzione codice) ===
def get_python_repl(
    globals_dict: Dict[str, Any] = None,
    locals_dict: Dict[str, Any] = None,
) -> PythonREPLTool:
    """REPL Python persistente tra chiamate (mantiene stato variabili)."""
    repl = PythonREPLTool()
    if globals_dict:
        repl.python_repl.globals.update(globals_dict)
    if locals_dict:
        repl.python_repl.locals.update(locals_dict)
    return repl


# === Shell Command ===
def get_shell_tool() -> ShellTool:
    """Shell tool di LangChain: esegue QUALSIASI comando, senza allowlist né timeout."""
    # Usalo solo in un ambiente isolato. Per una versione con allowlist vedi safe_shell.
    return ShellTool()


# === Wikipedia ===
def get_wikipedia_tool(
    lang: str = "it",
    top_k_results: int = 3,
) -> Tool:
    from langchain_community.tools import WikipediaQueryRun
    return WikipediaQueryRun(
        api_wrapper=WikipediaAPIWrapper(
            lang=lang,
            top_k_results=top_k_results,
        )
    )


# Demo
if __name__ == "__main__":
    search = get_duckduckgo_search()
    print(search.invoke("Ollama local LLM installation"))
    
    repl = get_python_repl()
    print(repl.invoke("x = 42\nprint(f'Risposta: {x}')"))
    
    shell = get_shell_tool()
    print(shell.invoke("echo 'Hello from shell' && pwd"))

2. Custom Tools: @tool Decorator

# custom_tools.py
from langchain_core.tools import tool
from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Optional, Type, List, Dict, Any
import requests
import json
import os
import subprocess
import shlex
from pathlib import Path


# === File I/O Tools ===

class ReadFileInput(BaseModel):
    path: str = Field(description="Percorso file da leggere (relativo a WORK_DIR)")
    encoding: str = Field(default="utf-8", description="Encoding file")


@tool("read_file", args_schema=ReadFileInput, return_direct=False)
def read_file(path: str, encoding: str = "utf-8") -> str:
    """Legge contenuto di un file di testo. Sicuro: solo WORK_DIR."""
    work_dir = Path(os.getenv("AGENT_WORK_DIR", ".")).resolve()
    file_path = (work_dir / path).resolve()
    
    # Security: path traversal protection
    try:
        file_path.relative_to(work_dir)
    except ValueError:
        return f"Errore: Accesso negato - path fuori WORK_DIR: {path}"
    
    if not file_path.exists():
        return f"Errore: File non trovato: {path}"
    
    if not file_path.is_file():
        return f"Errore: Non è un file: {path}"
    
    try:
        content = file_path.read_text(encoding=encoding)
        return f"Contenuto di {path}:\n```\n{content}\n```"
    except Exception as e:
        return f"Errore lettura: {e}"


class WriteFileInput(BaseModel):
    path: str = Field(description="Percorso file da scrivere (relativo a WORK_DIR)")
    content: str = Field(description="Contenuto da scrivere")
    mode: str = Field(default="w", description="Modalità: 'w' sovrascrive, 'a' appende")


@tool("write_file", args_schema=WriteFileInput, return_direct=False)
def write_file(path: str, content: str, mode: str = "w") -> str:
    """Scrive contenuto su file. Sicuro: solo WORK_DIR, crea directory se serve."""
    work_dir = Path(os.getenv("AGENT_WORK_DIR", ".")).resolve()
    file_path = (work_dir / path).resolve()
    
    try:
        file_path.relative_to(work_dir)
    except ValueError:
        return f"Errore: Accesso negato - path fuori WORK_DIR: {path}"
    
    # Crea directory parent
    file_path.parent.mkdir(parents=True, exist_ok=True)
    
    try:
        if mode == "a":
            with file_path.open("a", encoding="utf-8") as fh:
                fh.write(content)
            return f"Appeso a {path} ({len(content)} caratteri)"
        else:
            file_path.write_text(content, encoding="utf-8")
            return f"Scritto {path} ({len(content)} caratteri)"
    except Exception as e:
        return f"Errore scrittura: {e}"


class ListFilesInput(BaseModel):
    path: str = Field(default=".", description="Directory da listare")
    pattern: str = Field(default="*", description="Glob pattern (es. '*.py', '**/*.md')")


@tool("list_files", args_schema=ListFilesInput, return_direct=False)
def list_files(path: str = ".", pattern: str = "*") -> str:
    """Lista file in directory con pattern glob."""
    work_dir = Path(os.getenv("AGENT_WORK_DIR", ".")).resolve()
    dir_path = (work_dir / path).resolve()
    
    try:
        dir_path.relative_to(work_dir)
    except ValueError:
        return f"Errore: Accesso negato - path fuori WORK_DIR: {path}"
    
    if not dir_path.exists():
        return f"Errore: Directory non trovata: {path}"
    
    files = list(dir_path.glob(pattern))
    if not files:
        return f"Nessun file trovato in {path} con pattern {pattern}"
    
    lines = []
    for f in sorted(files):
        rel = f.relative_to(work_dir)
        size = f.stat().st_size if f.is_file() else 0
        lines.append(f"{'DIR' if f.is_dir() else 'FILE'}  {size:>10}  {rel}")
    
    return "\n".join(lines)


# === HTTP Request Tool ===

class HTTPRequestInput(BaseModel):
    url: str = Field(description="URL completo (http/https)")
    method: str = Field(default="GET", description="Metodo HTTP: GET, POST, PUT, DELETE")
    headers: Optional[Dict[str, str]] = Field(default=None, description="Headers opzionali")
    json_body: Optional[Dict[str, Any]] = Field(default=None, description="Body JSON per POST/PUT")
    params: Optional[Dict[str, str]] = Field(default=None, description="Query parameters")
    timeout: int = Field(default=30, description="Timeout secondi")


@tool("http_request", args_schema=HTTPRequestInput, return_direct=False)
def http_request(
    url: str,
    method: str = "GET",
    headers: Optional[Dict[str, str]] = None,
    json_body: Optional[Dict[str, Any]] = None,
    params: Optional[Dict[str, str]] = None,
    timeout: int = 30,
) -> str:
    """Esegue richiesta HTTP/REST. Allowlist domini via env AGENT_ALLOWED_DOMAINS."""
    # Security: domain allowlist
    allowed = os.getenv("AGENT_ALLOWED_DOMAINS", "").split(",")
    allowed = [d.strip() for d in allowed if d.strip()]
    
    from urllib.parse import urlparse
    domain = urlparse(url).netloc
    if allowed and domain not in allowed:
        return f"Errore: Dominio non in allowlist: {domain}. Allowlist: {allowed}"
    
    try:
        resp = requests.request(
            method=method.upper(),
            url=url,
            headers=headers or {},
            json=json_body,
            params=params,
            timeout=timeout,
        )
        resp.raise_for_status()
        
        # Prova JSON, fallback text
        try:
            return json.dumps(resp.json(), indent=2, ensure_ascii=False)
        except ValueError:
            return resp.text[:5000]  # Trunca output lungo
    except requests.RequestException as e:
        return f"Errore HTTP: {e}"


# === Safe Shell Tool (con allowlist) ===

class SafeShellInput(BaseModel):
    command: str = Field(description="Comando da eseguire")
    timeout: int = Field(default=30, description="Timeout secondi")
    cwd: Optional[str] = Field(default=None, description="Working directory (relativo a WORK_DIR)")


# Allowlist comandi sicuri
SAFE_COMMANDS = {
    "ls", "cat", "head", "tail", "grep", "find", "wc",
    "python3", "python", "pip", "uv", "poetry",
    "git", "cargo", "go", "node", "npm", "bun",
    "curl", "wget", "jq", "yq",
    "mkdir", "touch", "cp", "mv", "rm",
    "echo", "printf", "date", "whoami", "pwd",
}

@tool("safe_shell", args_schema=SafeShellInput, return_direct=False)
def safe_shell(command: str, timeout: int = 30, cwd: Optional[str] = None) -> str:
    """Esegue comando shell SOLO se il primo comando è in allowlist. Limitato, non a prova di abuso: vedi la sezione Sicurezza."""
    # Parse comando
    try:
        parts = shlex.split(command)
    except ValueError as e:
        return f"Errore parsing comando: {e}"
    
    if not parts:
        return "Errore: Comando vuoto"
    
    base_cmd = parts[0]
    if base_cmd not in SAFE_COMMANDS:
        return f"Errore: Comando non permesso: {base_cmd}. Permessi: {sorted(SAFE_COMMANDS)}"
    
    # Working directory
    work_dir = Path(os.getenv("AGENT_WORK_DIR", ".")).resolve()
    if cwd:
        exec_dir = (work_dir / cwd).resolve()
        try:
            exec_dir.relative_to(work_dir)
        except ValueError:
            return f"Errore: cwd fuori WORK_DIR: {cwd}"
    else:
        exec_dir = work_dir
    
    try:
        result = subprocess.run(
            parts,
            cwd=exec_dir,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        output = result.stdout
        if result.stderr:
            output += f"\n[stderr]\n{result.stderr}"
        if result.returncode != 0:
            output += f"\n[exit code: {result.returncode}]"
        return output[:10000]  # Trunca
    except subprocess.TimeoutExpired:
        return f"Errore: Timeout ({timeout}s)"
    except Exception as e:
        return f"Errore esecuzione: {e}"


# === Calculator Tool ===

class CalculatorInput(BaseModel):
    expression: str = Field(description="Espressione matematica (es. '2 * (3 + 4) / 5')")


@tool("calculator", args_schema=CalculatorInput, return_direct=False)
def calculator(expression: str) -> str:
    """Calcola espressione matematica (solo operatori base, niente potenze)."""
    # Whitelist caratteri permessi
    allowed_chars = set("0123456789+-*/().% ")
    if not all(c in allowed_chars for c in expression):
        return "Errore: Caratteri non permessi nell'espressione"
    if "**" in expression:
        return "Errore: potenze non permesse (rischio di calcoli senza fine)"
    
    try:
        # Safe eval: solo math builtin
        result = eval(expression, {"__builtins__": {}}, {})
        return f"Risultato: {result}"
    except Exception as e:
        return f"Errore calcolo: {e}"


# === Registry Tools ===

def get_all_tools() -> List[BaseTool]:
    """Restituisce tutti i tool custom registrati."""
    return [
        read_file,
        write_file,
        list_files,
        http_request,
        safe_shell,
        calculator,
    ]


def get_tool_by_name(name: str) -> BaseTool:
    for tool in get_all_tools():
        if tool.name == name:
            return tool
    raise ValueError(f"Tool non trovato: {name}")

3. Agent Initialization: ReAct vs Function Calling

# agent_setup.py
from langchain.agents import (
    initialize_agent,
    AgentType,
    AgentExecutor,
)
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.language_models import BaseChatModel
from typing import List, Any, Dict
from llm_setup import get_llm
from custom_tools import get_all_tools
import logging


# === Callback per logging/debug ===
class AgentCallbackHandler(BaseCallbackHandler):
    def on_agent_action(self, action: AgentAction, **kwargs):
        logging.info(f"[AGENT] Action: {action.tool} | Input: {action.tool_input}")
    
    def on_tool_end(self, output: str, **kwargs):
        logging.info(f"[TOOL] Output: {output[:200]}...")
    
    def on_agent_finish(self, finish: AgentFinish, **kwargs):
        logging.info(f"[AGENT] Finish: {finish.return_values}")


# === ReAct Agent (Text-based parsing) ===
def create_react_agent(
    llm: BaseChatModel = None,
    tools: List = None,
    max_iterations: int = 10,
    max_execution_time: float = 60.0,
    verbose: bool = True,
) -> AgentExecutor:
    """
    Agente ReAct classico: LLM genera testo "Thought/Action/Observation",
    parser estrae tool e input. Funziona con QUALSIASI modello.
    """
    llm = llm or get_llm()
    tools = tools or get_all_tools()
    
    agent = initialize_agent(
        tools=tools,
        llm=llm,
        agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
        max_iterations=max_iterations,
        max_execution_time=max_execution_time,
        verbose=verbose,
        handle_parsing_errors=True,  # Tenta recovery se parsing fallisce
        callbacks=[AgentCallbackHandler()],
    )
    return agent


# === Structured Chat Agent (JSON in testo, NON function calling nativo) ===
def create_structured_chat_agent(
    llm: BaseChatModel = None,
    tools: List = None,
    max_iterations: int = 10,
    verbose: bool = True,
) -> AgentExecutor:
    """
    Agente structured chat: il modello risponde con un blocco JSON (action, action_input)
    che LangChain interpreta. Gestisce tool con più argomenti, ma il parsing resta testuale.
    Per il function calling nativo vedi create_tool_calling_agent nella sezione dedicata.
    Modelli con supporto ai tool: Llama 3.1, Mistral 7B v0.3, Qwen 2.5.
    """
    llm = llm or get_llm()
    tools = tools or get_all_tools()
    
    # Verifica supporto function calling
    model_name = getattr(llm, "model", "").lower()
    supports_fc = any(m in model_name for m in ["llama3.1", "mistral", "qwen2.5"])
    
    if not supports_fc:
        logging.warning(f"Modello {model_name} potrebbe non supportare function calling nativo. Fallback a ReAct.")
        return create_react_agent(llm, tools, verbose=verbose)
    
    agent = initialize_agent(
        tools=tools,
        llm=llm,
        agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
        max_iterations=max_iterations,
        verbose=verbose,
        handle_parsing_errors=True,
        callbacks=[AgentCallbackHandler()],
    )
    return agent


# === Auto-detect best agent ===
def create_agent(
    llm: BaseChatModel = None,
    tools: List = None,
    prefer_function_calling: bool = True,
    **kwargs,
) -> AgentExecutor:
    """Sceglie automaticamente il tipo agente migliore per il modello."""
    llm = llm or get_llm()
    model_name = getattr(llm, "model", "").lower()
    
    supports_fc = any(m in model_name for m in ["llama3.1", "mistral", "qwen2.5"])
    
    if prefer_function_calling and supports_fc:
        logging.info(f"Usando Structured Chat Agent (function calling) per {model_name}")
        return create_structured_chat_agent(llm, tools, **kwargs)
    else:
        logging.info(f"Usando ReAct Agent (text parsing) per {model_name}")
        return create_react_agent(llm, tools, **kwargs)

4. Tool Descriptions: Prompt Engineering per LLM

# tool_descriptions.py
"""
Le description dei tool SONO il prompt per l'LLM.
Devono essere: chiare, specifiche, con esempi input/output.
"""

TOOL_DESCRIPTIONS = {
    "read_file": """Legge il contenuto di un file di testo.
Input: {"path": "percorso/relativo/file.txt", "encoding": "utf-8"}
Output: Contenuto del file o messaggio errore.
Usa per: leggere config, codice, documenti, log.
Nota: Path relativo a WORK_DIR. Non usa path assoluti.""",
    
    "write_file": """Scrive contenuto su file (crea directory se serve).
Input: {"path": "percorso/relativo/file.txt", "content": "testo da scrivere", "mode": "w"}
Mode: "w" = sovrascrive, "a" = appende in fondo.
Output: Conferma scrittura o errore.
Usa per: salvare report, generare codice, creare config.""",
    
    "list_files": """Lista file in directory con pattern glob.
Input: {"path": ".", "pattern": "**/*.py"}
Output: Lista tipo 'FILE  1234  src/main.py' o 'DIR       0  src/'
Usa per: esplorare progetto, trovare file, vedere struttura.""",
    
    "safe_shell": """Esegue comando shell limitato da allowlist (primo comando).
Comandi permessi: ls, cat, grep, find, python3, git, curl, jq, mkdir, ecc.
Input: {"command": "ls -la", "timeout": 30, "cwd": "subdir"}
Output: stdout + stderr + exit code.
NON usa per: rm -rf, sudo, comandi distruttivi, network tools non in lista.""",
    
    "http_request": """Fa richiesta HTTP/REST a API esterne.
Input: {"url": "https://api.example.com/data", "method": "GET", "headers": {}, "json_body": {}, "params": {}}
Allowlist domini: variabile AGENT_ALLOWED_DOMAINS (comma-separated).
Output: JSON pretty-printed o testo (max 5000 char).
Usa per: chiamare API, webhook, fetch dati JSON.""",
    
    "calculator": """Calcola espressioni matematiche semplici.
Input: {"expression": "2 * (3 + 4) / 5"}
Operatori: + - * / ( ) % . numeri
Output: "Risultato: 2.8"
Usa per: calcoli veloci, conversioni unità, formule.""",
    
    "duckduckgo_search": """Cerca web su DuckDuckGo (no API key).
Input: {"query": "Ollama local LLM tutorial"}
Output: Snippet risultati con URL.
Usa per: info attuali, documentazione, troubleshooting, news.""",
    
    "python_repl": """Esegue codice Python in REPL persistente.
Mantiene stato variabili tra chiamate.
Input: {"code": "x = 42\nprint(x * 2)"}
Output: Output stdout/stderr.
Usa per: calcoli complessi, data processing, test codice, prototipi.""",
}


def enhance_tool_descriptions(tools: list) -> list:
    """Aggiunge descrizioni ottimizzate ai tool."""
    for tool in tools:
        if tool.name in TOOL_DESCRIPTIONS:
            tool.description = TOOL_DESCRIPTIONS[tool.name]
    return tools

5. Sicurezza: Sandboxing & Allowlist

# security.py
"""
Sicurezza per agenti con tool use.
Principio: Zero Trust. Niente esegue senza validazione esplicita.
"""

import os
import subprocess
from pathlib import Path
from typing import Set, List
from functools import wraps


class SecurityConfig:
    """Configurazione centralizzata sicurezza."""
    
    def __init__(self):
        # Work directory (root per tutti i file operations)
        self.work_dir = Path(os.getenv("AGENT_WORK_DIR", ".")).resolve()
        
        # Allowlist domini HTTP
        self.allowed_domains: Set[str] = set(
            d.strip() for d in os.getenv("AGENT_ALLOWED_DOMAINS", "").split(",") if d.strip()
        )
        
        # Allowlist comandi shell
        self.allowed_commands: Set[str] = set(
            c.strip() for c in os.getenv("AGENT_ALLOWED_COMMANDS", "").split(",") if c.strip()
        ) or {
            "ls", "cat", "head", "tail", "grep", "find", "wc",
            "python3", "python", "pip", "uv",
            "git", "cargo", "go", "node", "npm",
            "curl", "wget", "jq", "yq",
            "mkdir", "touch", "cp", "mv",
            "echo", "printf", "date", "whoami", "pwd",
        }
        
        # Timeout default
        self.default_timeout = int(os.getenv("AGENT_TOOL_TIMEOUT", "30"))
        
        # Max output size
        self.max_output_chars = int(os.getenv("AGENT_MAX_OUTPUT", "10000"))
        
        # Read-only mode
        self.read_only = os.getenv("AGENT_READ_ONLY", "false").lower() == "true"


security = SecurityConfig()


def validate_path(path: str, base: Path = None) -> Path:
    """Valida che path sia dentro work_dir. Ritorna Path risolto."""
    base = base or security.work_dir
    target = (base / path).resolve()
    try:
        target.relative_to(base)
    except ValueError:
        raise PermissionError(f"Path traversal attempt: {path} -> {target}")
    return target


def validate_domain(url: str) -> bool:
    """Controlla se dominio è in allowlist."""
    if not security.allowed_domains:
        return True  # Nessuna restrizione se allowlist vuota
    from urllib.parse import urlparse
    domain = urlparse(url).netloc
    return domain in security.allowed_domains


def validate_command(command: str) -> List[str]:
    """Valida comando contro allowlist. Ritorna parti parsate."""
    import shlex
    parts = shlex.split(command)
    if not parts:
        raise ValueError("Comando vuoto")
    if parts[0] not in security.allowed_commands:
        raise PermissionError(f"Comando non permesso: {parts[0]}")
    return parts


def safe_tool(func):
    """Decorator per tool sicuri: valida input, catcha errori, logga."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            # Read-only check
            if security.read_only and func.__name__ in ["write_file", "safe_shell", "http_request"]:
                if func.__name__ == "http_request" and kwargs.get("method", "GET") != "GET":
                    return "Errore: Modalità read-only - scritture disabilitate"
                elif func.__name__ in ["write_file", "safe_shell"]:
                    return "Errore: Modalità read-only - scritture disabilitate"
            
            return func(*args, **kwargs)
        except PermissionError as e:
            return f"Errore sicurezza: {e}"
        except Exception as e:
            return f"Errore tool {func.__name__}: {e}"
    return wrapper

6. Error Handling & Recovery

# error_handling.py
from langchain.agents import AgentExecutor
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.exceptions import OutputParserException
from typing import Any, Dict, List, Optional
import logging


class RobustAgentExecutor(AgentExecutor):
    """AgentExecutor con error handling migliorato."""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.error_counts: Dict[str, int] = {}
        self.max_tool_errors = 3
    
    def _take_next_step(
        self,
        name_to_tool_map: Dict[str, Any],
        color_mapping: Dict[str, str],
        inputs: Dict[str, str],
        intermediate_steps: List[tuple],
        **kwargs,
    ) -> Any:
        try:
            return super()._take_next_step(
                name_to_tool_map, color_mapping, inputs, intermediate_steps, **kwargs
            )
        except OutputParserException as e:
            # Parsing error: LLM ha generato formato non valido
            logging.warning(f"Parsing error: {e}. Tentativo recovery...")
            # Aggiungi hint al prossimo prompt
            if intermediate_steps:
                last_action, last_output = intermediate_steps[-1]
                return self.agent.plan(
                    steps=intermediate_steps,
                    **inputs,
                )
            raise
        except Exception as e:
            logging.error(f"Errore step agente: {e}")
            raise
    
    def _handle_tool_error(self, error: Exception, tool_name: str) -> str:
        """Gestisce errori tool con retry logic."""
        self.error_counts[tool_name] = self.error_counts.get(tool_name, 0) + 1
        
        if self.error_counts[tool_name] >= self.max_tool_errors:
            return f"Tool {tool_name} ha fallito troppe volte ({self.max_tool_errors}). Abbandono."
        
        # Suggerimento per LLM
        return f"Errore tool {tool_name}: {error}. Riprova con input diverso o usa tool alternativo."

7. Esempio Completo: Agente Autonomo

# main_agent.py
import os
from agent_setup import create_agent
from custom_tools import get_all_tools
from tool_descriptions import enhance_tool_descriptions
from llm_setup import get_llm


def setup_environment():
    """Configura environment per sicurezza."""
    # Work directory
    os.environ["AGENT_WORK_DIR"] = os.path.abspath("./agent_workspace")
    
    # Allowlist domini (es. API interne, documentazione)
    os.environ["AGENT_ALLOWED_DOMAINS"] = "api.github.com,raw.githubusercontent.com,httpbin.org"
    
    # Allowlist comandi (estende default)
    os.environ["AGENT_ALLOWED_COMMANDS"] = "ls,cat,head,tail,grep,find,wc,python3,git,curl,jq,mkdir,touch,echo,date,pwd"
    
    # Timeout
    os.environ["AGENT_TOOL_TIMEOUT"] = "60"
    
    # Crea workspace
    os.makedirs(os.environ["AGENT_WORK_DIR"], exist_ok=True)


def main():
    setup_environment()
    
    # LLM
    llm = get_llm()
    print(f"Modello: {llm.model}")
    
    # Tools
    tools = get_all_tools()
    tools = enhance_tool_descriptions(tools)
    print(f"Tools caricati: {[t.name for t in tools]}")
    
    # Agente
    agent = create_agent(
        llm=llm,
        tools=tools,
        prefer_function_calling=True,
        max_iterations=15,
        verbose=True,
    )
    
    # Task esempio
    task = """
    Crea un file 'report.md' con:
    1. Cerca web "Ollama local LLM best practices"
    2. Salva i primi 3 risultati nel file
    3. Aggiungi una sezione con calcolo: 15% di 2450
    4. Lista i file nella directory corrente
    """
    
    print(f"\n{'='*60}")
    print(f"TASK: {task.strip()}")
    print(f"{'='*60}\n")
    
    result = agent.invoke({"input": task})
    
    print(f"\n{'='*60}")
    print(f"RISULTATO FINALE:")
    print(result["output"])
    print(f"{'='*60}")


if __name__ == "__main__":
    main()

L'RobustAgentExecutor è un esempio didattico: sovrascrive _take_next_step, un metodo interno di LangChain la cui firma cambia tra le versioni, e _handle_tool_error non viene richiamato dal flusso standard. In un progetto reale usa i parametri ufficiali (handle_parsing_errors, max_iterations, max_execution_time).

Tabella Confronto: ReAct vs Function Calling

| Aspetto | ReAct (ZERO_SHOT_REACT) | Function calling nativo (create_tool_calling_agent) | | --- | --- | --- | | Modelli supportati | Tutti, anche piccoli | Solo modelli con supporto ai tool (per esempio Llama 3.1, Mistral, Qwen 2.5) | | Parsing | Regex su testo (fragile) | JSON nativo tool_calls (robusto) | | Affidabilità | Più sensibile al formato del testo | Di solito più stabile, da misurare sul tuo modello | | Token usati | Prompt più lungo | Schema dei tool passato in forma strutturata | | Debugging | Vedi reasoning testuale | Vedi JSON strutturato | | Tool descriptions | Critiche (prompt engineering) | Schema JSON + description | | Parallel tool calls | No (sequenziale) | Sì (batch tool_calls) | | Fallback | Sempre disponibile | Fallback a ReAct se non supportato |

Troubleshooting Tool Use

| Problema | Causa | Soluzione | | --- | --- | --- | | "Could not parse LLM output" | ReAct: formato Thought/Action sbagliato | handle_parsing_errors=True, migliora tool description, usa function calling | | Tool non chiamato mai | Description poco chiara / LLM non capisce | Riscrivi description con esempi input/output espliciti | | Tool chiamato con argomenti sbagliati | Schema args non valido / description ambigua | Usa args_schema Pydantic, valida in tool | | Timeout tool | Comando lento / rete lenta | Aumenta AGENT_TOOL_TIMEOUT, usa async per HTTP | | Permission denied | Path traversal / comando non in allowlist | Controlla AGENT_WORK_DIR, AGENT_ALLOWED_COMMANDS, AGENT_ALLOWED_DOMAINS | | Loop infinito | LLM ripete stessa azione | max_iterations, aggiungi tool finish esplicito, migliora system prompt | | Output troncato | max_output_chars troppo basso | Aumenta AGENT_MAX_OUTPUT o usa file per output grandi |

Come fare, passo per passo: dal primo tool all'agente

Questa sequenza ti porta dalla chat del Cap. 5 a un agente che usa strumenti, senza saltare le verifiche.

1. Fissa le versioni. Crea un ambiente virtuale e installa i pacchetti di requirements.txt con i limiti indicati sopra (langchain<1.0). Un pip install senza vincoli oggi installerebbe versioni in cui le API di questo capitolo sono cambiate.
2. Prova un tool alla volta. Esegui la sezione "Demo" di builtin_tools.py: se la ricerca web o il REPL non rispondono, l'errore è nel tool e non nell'agente.
3. Crea la cartella di lavoro. Imposta AGENT_WORK_DIR su una cartella vuota e dedicata (agent_workspace). Tutti i tool su file lavorano solo lì dentro.
4. Parti in sola lettura. Registra solo read_file, list_files e calculator. Aggiungi write_file, http_request e safe_shell quando il resto funziona.
5. Scrivi descrizioni chiare. Il modello sceglie il tool leggendo la descrizione: indica a cosa serve, che cosa si aspetta in input e quando non va usato.
6. Imposta i limiti. Valorizza max_iterations e max_execution_time, e lascia attivo il logging con AgentCallbackHandler, così vedi ogni azione richiesta dal modello.
7. Prova con richieste ostili. Chiedi all'agente di leggere ../../etc/passwd con il tool di lettura dei file, di cancellare file o di contattare un dominio fuori allowlist. Ogni richiesta deve fallire con un messaggio di errore. Il tool della shell, invece, passa questo test solo se togli cat e find dall'allowlist: controlla solo il primo comando, non i suoi argomenti (vedi «Sicurezza: limiti degli esempi» più sotto).
8. Allarga gradualmente. Aggiungi un tool solo dopo aver scritto il test che prova come può essere abusato.

Function calling nativo e agenti "structured chat"

Una precisazione tecnica. AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION non usa il function calling nativo del modello: è un agente basato su testo, in cui il modello risponde con un blocco JSON (action e action_input) che LangChain interpreta. Supporta tool con più argomenti, ma resta un meccanismo di parsing. Il function calling nativo, in cui il modello restituisce tool_calls strutturati, si ottiene con create_tool_calling_agent:

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "Sei un assistente che usa gli strumenti a disposizione."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent, tools=tools, max_iterations=10, handle_parsing_errors=True
)

Il modello deve supportare i tool (con Ollama, verifica nella scheda del modello). initialize_agent fa parte dell'API storica di LangChain: la documentazione LangChain e la guida di migrazione indicano per i nuovi progetti i costruttori più recenti e LangGraph. Per questo il capitolo fissa langchain<1.0.

Sicurezza: limiti degli esempi

Gli strumenti sono la parte più delicata di un agente: un modello può essere indotto, da un documento o da una pagina web letta, a chiamare un tool in modo dannoso (il rischio è descritto nella OWASP Top 10 per le applicazioni LLM come prompt injection ed excessive agency). Gli esempi di questo capitolo riducono il rischio, ma non lo eliminano:

• L'allowlist dei comandi controlla solo il primo comando. python3, node, curl, git, find ed rm possono eseguire codice arbitrario o leggere e cancellare file fuori dalla cartella di lavoro, perché il controllo sul percorso vale per i tool su file e non per gli argomenti dei comandi. Per un uso reale restringi la lista ai comandi che servono davvero oppure esegui l'agente in un container o in una macchina virtuale senza accesso alla rete interna.
• Con AGENT_ALLOWED_DOMAINS vuota, http_request raggiunge qualsiasi indirizzo, comprese le risorse della tua rete locale. Compila sempre l'allowlist.
• Il REPL Python esegue codice arbitrario con i tuoi permessi: non abilitarlo su dati o macchine che non puoi perdere.
• security.py definisce una configurazione centralizzata, ma i tool di custom_tools.py fanno i propri controlli: se la usi, collega davvero i tool a quelle funzioni di validazione.

Checklist prima di dare gli strumenti all'agente

• Cartella di lavoro dedicata, vuota e fuori dai tuoi dati personali.
• Allowlist dei comandi ridotta al minimo e senza interpreti.
• Allowlist dei domini compilata.
• Tool di scrittura e shell disattivati finché non servono.
• max_iterations e max_execution_time impostati.
• Log delle azioni attivo e letto dopo ogni prova.
• Test con richieste ostili superati.
• Agente eseguito con un utente senza privilegi.

Errori da evitare

• Copiare l'esempio con l'allowlist predefinita e considerarlo sicuro per la produzione.
• Dare all'agente il REPL Python o la shell per "comodità".
• Lasciare la descrizione di un tool vaga: il modello sceglie quello sbagliato.
• Mescolare nella stessa sessione documenti non fidati e tool con effetti sul sistema.
• Ignorare i messaggi di errore dei tool: sono ciò che il modello usa per correggersi.
• Aggiornare LangChain senza rileggere le note di rilascio, perché le API degli agenti cambiano.

Consigli pratici

• Scrivi la descrizione del tool come se spiegassi a un collega quando usarlo, con un esempio di input.
• Restituisci errori brevi e utili ("file non trovato: controlla il percorso") e non eccezioni complete.
• Tronca gli output lunghi: un risultato di 100.000 caratteri riempie il contesto e peggiora le risposte.
• Salva su file i risultati lunghi e passa al modello solo il riassunto.
• Prova prima con un modello più grande, poi riduci: sai se il problema sono i tool o il modello.
• Per capire dove un agente aziendale è utile e dove no, vedi l'articolo sugli agenti AI in azienda e la pagina su chatbot e agenti.

Domande frequenti

Posso usare un modello da 3B con i tool? Con l'approccio ReAct sì, ma l'affidabilità cala con i modelli piccoli: prova con i tuoi casi d'uso e misura quante volte il tool giusto viene scelto.

Devo usare ReAct o function calling? Se il modello supporta i tool, il function calling nativo è più semplice da gestire. ReAct resta utile con modelli che non li supportano.

Posso fare a meno della shell? Sì, ed è la scelta più sicura. Molti compiti si risolvono con i tool su file e con le richieste HTTP verso un elenco ristretto di domini.

Come evito che l'agente cancelli file? Non dargli i tool che li cancellano, usa una cartella di lavoro dedicata e fai backup. Il controllo sul percorso non basta se il modello può lanciare comandi.

Dove vedo che cosa ha fatto l'agente? Nei log di AgentCallbackHandler e, con verbose=True, nel terminale.

Prossimo capitolo

Cap. 7: costruisci un'interfaccia grafica → Gradio Blocks: chat interface, file upload per RAG, parametri modello (temperature, max_tokens), streaming, theme custom, deploy locale con app.launch(share=False), HTTPS tunnel opzionale.

File completo

Modulo agent tools: agent_tools.py — tutti i tool, security, agent factory, error handling.

Serie "Il Mio Primo Agente Digitale" — Requisiti: LangChain 0.3.x (non 1.0), Python 3.11+, DuckDuckGo-Search 5+, Ollama 0.3+ (Llama 3.1, Mistral, Qwen 2.5 per function calling).