22Esempio 4 · Assistenza clienti

assistenza-clienti/
├── .env
├── assistenza_clienti.py
└── knowledge/
    └── faq_ortoverde.txt

MODEL=ollama/qwen2.5:7b
CREWAI_DISABLE_TELEMETRY=true

class Smistamento(BaseModel):
    categoria: Literal["spedizione", "reso", "prodotto_danneggiato", "pagamento", "altro"]
    urgenza: Literal["bassa", "media", "alta"]
    motivo: str = Field(description="Una frase in italiano che spiega la scelta")


class StatoTicket(BaseModel):
    ticket: str = ""
    smistamento: Smistamento | None = None
    esito: str = ""

smistatore = Agent(
    role="Addetto allo smistamento dei ticket di Ortoverde",
    goal="Assegnare a ogni ticket la categoria e l'urgenza giuste, senza rispondere al cliente",
    backstory=(
        "Da anni leggi per primo le richieste dei clienti di un negozio online di giardinaggio. "
        "Consideri urgente chi minaccia vie legali, chi ha pagato due volte o chi è molto arrabbiato."
    ),
    llm=MODELLO,
    verbose=True,
)

addetto = Agent(
    role="Addetto all'assistenza clienti di Ortoverde",
    goal="Scrivere risposte gentili e corrette basate solo sulle domande frequenti del negozio",
    backstory=(
        "Rispondi ai clienti da cinque anni. Non prometti mai cose che non sono scritte nelle FAQ: "
        "se non trovi la risposta, dici che un collega ricontatterà il cliente."
    ),
    llm=MODELLO,
    knowledge_sources=[TextFileKnowledgeSource(file_paths=["faq_ortoverde.txt"])],
    embedder=EMBEDDER,
    verbose=True,
)

class AssistenzaFlow(Flow[StatoTicket]):

    @start()
    def smista(self):
        classificazione = Task(
            description="Leggi questo ticket di un cliente e classificalo.\n\nTicket: {ticket}",
            expected_output=(
                "La categoria (spedizione, reso, prodotto_danneggiato, pagamento o altro), "
                "l'urgenza (bassa, media o alta) e una frase in italiano che spiega la scelta."
            ),
            agent=smistatore,
            output_pydantic=Smistamento,
        )
        crew = Crew(agents=[smistatore], tasks=[classificazione], process=Process.sequential, verbose=True)
        risultato = crew.kickoff(inputs={"ticket": self.state.ticket})
        self.state.smistamento = risultato.pydantic

    @router(smista)
    def decidi(self):
        s = self.state.smistamento
        if s is None or s.urgenza == "alta" or s.categoria == "altro":
            return "a_una_persona"
        return "risposta_automatica"

    @listen("risposta_automatica")
    def rispondi(self):
        bozza = Task(
            description=(
                "Scrivi la risposta a questo ticket (categoria: {categoria}).\n\nTicket: {ticket}\n\n"
                "Usa solo le informazioni delle FAQ di Ortoverde e indica tra parentesi il numero della FAQ usata."
            ),
            expected_output="Una email in italiano di massimo 120 parole, con saluto iniziale e firma \"Il servizio clienti Ortoverde\".",
            agent=addetto,
            output_file="bozza_risposta.md",
        )
        crew = Crew(agents=[addetto], tasks=[bozza], process=Process.sequential, verbose=True)
        risultato = crew.kickoff(inputs={"ticket": self.state.ticket, "categoria": self.state.smistamento.categoria})
        self.state.esito = "Bozza di risposta salvata in bozza_risposta.md:\n\n" + risultato.raw
        return self.state.esito

    @listen("a_una_persona")
    def passa_a_una_persona(self):
        s = self.state.smistamento
        scheda = (
            f"TICKET DA GESTIRE A MANO\n"
            f"Categoria: {s.categoria if s else 'non classificato'}\n"
            f"Urgenza: {s.urgenza if s else 'sconosciuta'}\n"
            f"Motivo: {s.motivo if s else 'la classificazione non è riuscita'}\n\n"
            f"Testo del cliente:\n{self.state.ticket}\n"
        )
        with open("da_gestire_a_mano.txt", "w", encoding="utf-8") as f:
            f.write(scheda)
        self.state.esito = "Ticket passato a una persona: scheda salvata in da_gestire_a_mano.txt\n\n" + scheda
        return self.state.esito

if __name__ == "__main__":
    ticket = (
        "Buongiorno, ho ricevuto ieri il mio ordine ma il vaso in ceramica è arrivato con una crepa. "
        "Cosa devo fare? Grazie, Marta"
    )
    flow = AssistenzaFlow()
    esito = flow.kickoff(inputs={"ticket": ticket})
    print("\n=== ESITO ===")
    print(esito)

# /// script
# requires-python = ">=3.10,<3.14"
# dependencies = ["crewai[tools]==1.15.21", "ollama"]
# ///
"""Smista un ticket del negozio online (inventato) Ortoverde: risponde con le FAQ o lo passa a una persona."""
import os
import sys
from typing import Literal

from pydantic import BaseModel, Field

from crewai import Agent, Crew, Process, Task
from crewai.flow.flow import Flow, listen, router, start
from crewai.knowledge.source.text_file_knowledge_source import TextFileKnowledgeSource

# Il modello si sceglie nel file .env (capitolo 7). Se MODEL manca, si usa quello scritto qui.
MODELLO = os.getenv("MODEL", "openai/gpt-4.1-mini")

# L'embedder serve alla knowledge: con Ollama uno locale, altrimenti quello di OpenAI.
if MODELLO.startswith("ollama/"):
    EMBEDDER = {"provider": "ollama", "config": {"model_name": "nomic-embed-text"}}
else:
    EMBEDDER = {"provider": "openai", "config": {"model_name": "text-embedding-3-small"}}


# CrewAI cerca i file nella cartella knowledge/: il percorso si scrive a partire da lì.
if not os.path.exists("knowledge/faq_ortoverde.txt"):
    sys.exit("Manca il file knowledge/faq_ortoverde.txt: scaricalo come spiegato in Preparazione.")


# La "scheda" che lo smistatore deve compilare: solo valori ammessi.
class Smistamento(BaseModel):
    categoria: Literal["spedizione", "reso", "prodotto_danneggiato", "pagamento", "altro"]
    urgenza: Literal["bassa", "media", "alta"]
    motivo: str = Field(description="Una frase in italiano che spiega la scelta")


# Lo stato del flow: il ticket e quello che scopriamo strada facendo.
class StatoTicket(BaseModel):
    ticket: str = ""
    smistamento: Smistamento | None = None
    esito: str = ""


smistatore = Agent(
    role="Addetto allo smistamento dei ticket di Ortoverde",
    goal="Assegnare a ogni ticket la categoria e l'urgenza giuste, senza rispondere al cliente",
    backstory=(
        "Da anni leggi per primo le richieste dei clienti di un negozio online di giardinaggio. "
        "Consideri urgente chi minaccia vie legali, chi ha pagato due volte o chi è molto arrabbiato."
    ),
    llm=MODELLO,
    verbose=True,
)

addetto = Agent(
    role="Addetto all'assistenza clienti di Ortoverde",
    goal="Scrivere risposte gentili e corrette basate solo sulle domande frequenti del negozio",
    backstory=(
        "Rispondi ai clienti da cinque anni. Non prometti mai cose che non sono scritte nelle FAQ: "
        "se non trovi la risposta, dici che un collega ricontatterà il cliente."
    ),
    llm=MODELLO,
    knowledge_sources=[TextFileKnowledgeSource(file_paths=["faq_ortoverde.txt"])],
    embedder=EMBEDDER,
    verbose=True,
)


class AssistenzaFlow(Flow[StatoTicket]):

    @start()
    def smista(self):
        # Primo passo: classificare il ticket con un output strutturato.
        classificazione = Task(
            description="Leggi questo ticket di un cliente e classificalo.\n\nTicket: {ticket}",
            expected_output=(
                "La categoria (spedizione, reso, prodotto_danneggiato, pagamento o altro), "
                "l'urgenza (bassa, media o alta) e una frase in italiano che spiega la scelta."
            ),
            agent=smistatore,
            output_pydantic=Smistamento,
        )
        crew = Crew(agents=[smistatore], tasks=[classificazione], process=Process.sequential, verbose=True)
        risultato = crew.kickoff(inputs={"ticket": self.state.ticket})
        self.state.smistamento = risultato.pydantic

    @router(smista)
    def decidi(self):
        # Regola scritta da noi, non dal modello: i casi delicati vanno a una persona.
        s = self.state.smistamento
        if s is None or s.urgenza == "alta" or s.categoria == "altro":
            return "a_una_persona"
        return "risposta_automatica"

    @listen("risposta_automatica")
    def rispondi(self):
        bozza = Task(
            description=(
                "Scrivi la risposta a questo ticket (categoria: {categoria}).\n\nTicket: {ticket}\n\n"
                "Usa solo le informazioni delle FAQ di Ortoverde e indica tra parentesi il numero della FAQ usata."
            ),
            expected_output="Una email in italiano di massimo 120 parole, con saluto iniziale e firma \"Il servizio clienti Ortoverde\".",
            agent=addetto,
            output_file="bozza_risposta.md",
        )
        crew = Crew(agents=[addetto], tasks=[bozza], process=Process.sequential, verbose=True)
        risultato = crew.kickoff(inputs={"ticket": self.state.ticket, "categoria": self.state.smistamento.categoria})
        self.state.esito = "Bozza di risposta salvata in bozza_risposta.md:\n\n" + risultato.raw
        return self.state.esito

    @listen("a_una_persona")
    def passa_a_una_persona(self):
        # Nessun agente qui: prepariamo una scheda per l'operatore e ci fermiamo.
        s = self.state.smistamento
        scheda = (
            f"TICKET DA GESTIRE A MANO\n"
            f"Categoria: {s.categoria if s else 'non classificato'}\n"
            f"Urgenza: {s.urgenza if s else 'sconosciuta'}\n"
            f"Motivo: {s.motivo if s else 'la classificazione non è riuscita'}\n\n"
            f"Testo del cliente:\n{self.state.ticket}\n"
        )
        with open("da_gestire_a_mano.txt", "w", encoding="utf-8") as f:
            f.write(scheda)
        self.state.esito = "Ticket passato a una persona: scheda salvata in da_gestire_a_mano.txt\n\n" + scheda
        return self.state.esito


if __name__ == "__main__":
    ticket = (
        "Buongiorno, ho ricevuto ieri il mio ordine ma il vaso in ceramica è arrivato con una crepa. "
        "Cosa devo fare? Grazie, Marta"
    )
    flow = AssistenzaFlow()
    esito = flow.kickoff(inputs={"ticket": ticket})
    print("\n=== ESITO ===")
    print(esito)