Architettura degli AI Agent: Agenti Autonomi e Framework LangChain

Gli AI Agents sono sistemi di intelligenza artificiale capaci di prendere decisioni autonomamente, utilizzare strumenti e risolvere compiti complessi passo dopo passo. In questa guida esaminiamo le architetture degli agenti e le loro implementazioni.

Che cos’è un Agente?

Un semplice confronto tra un chatbot e un agente:

1Chatbot:
2User: "How is the weather today?"
3Bot: "I don't know the weather." (static response)
4
5Agent:
6User: "How is the weather today?"
7Agent: 
8  1. "I need to find out the weather."
9  2. Call weather_api.get_weather()
10  3. Interpret the result
11  4. "It's 15°C in Istanbul, partly cloudy."

Componenti di un Agente

1. LLM (Cervello)

Presa di decisioni e ragionamento.

2. Tools

Interazione con il mondo esterno:

• Web search
• API calls
• Database queries
• Code execution

3. Memoria

Cronologia della conversazione e archiviazione delle informazioni.

4. Pianificazione

Scomposizione dei compiti e strategia.

Pattern ReAct

Ciclo Ragionamento + Azione:

1Thought: What do I need to do?
2Action: Which tool should I use?
3Action Input: Tool parameters
4Observation: Tool result
5... (repeat)
6Final Answer: Final response

Esempio ReAct

1Question: "What is Tesla's current stock price and how much has it changed in the last month?"
2
3Thought: I need to find Tesla's stock price.
4Action: stock_price
5Action Input: {"symbol": "TSLA"}
6Observation: Current price: $248.50
7
8Thought: Now I need to calculate the 1-month change.
9Action: stock_history  
10Action Input: {"symbol": "TSLA", "period": "1mo"}
11Observation: 1 month ago: $235.20, Change: +5.65%
12
13Thought: I can now answer the question.
14Final Answer: Tesla's current stock price is $248.50. 
15It has shown a 5.65% increase in the last month.

Sviluppo di Agenti con LangChain

Agente Base

1from langchain.agents import create_react_agent, AgentExecutor
2from langchain.tools import Tool
3from langchain_openai import ChatOpenAI
4from langchain import hub
5
6# LLM
7llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
8
9# Tools
10def search_web(query: str) -> str:
11    """Searches the web"""
12    # Web search implementation
13    return f"Search results for: {query}"
14
15def calculate(expression: str) -> str:
16    """Performs mathematical calculation"""
17    return str(eval(expression))
18
19tools = [
20    Tool(name="search", func=search_web, description="Searches the web"),
21    Tool(name="calculator", func=calculate, description="Performs math calculations"),
22]
23
24# Prompt
25prompt = hub.pull("hwchase17/react")
26
27# Agent
28agent = create_react_agent(llm, tools, prompt)
29agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
30
31# Run
32result = agent_executor.invoke({"input": "What is the world population in 2024?"})

Tools Personalizzati

1from langchain.tools import BaseTool
2from pydantic import BaseModel, Field
3
4class WeatherInput(BaseModel):
5    city: str = Field(description="City name")
6    unit: str = Field(default="celsius", description="Temperature unit")
7
8class WeatherTool(BaseTool):
9    name = "weather"
10    description = "Fetches weather information"
11    args_schema = WeatherInput
12    
13    def _run(self, city: str, unit: str = "celsius") -> str:
14        # API call
15        weather_data = fetch_weather(city)
16        return f"{city}: {weather_data['temp']}°{'C' if unit == 'celsius' else 'F'}"
17    
18    async def _arun(self, city: str, unit: str = "celsius") -> str:
19        return await async_fetch_weather(city)
20## Sistemi di Memoria
21
22### Conversation Buffer Memory
23
24```python
25from langchain.memory import ConversationBufferMemory
26
27memory = ConversationBufferMemory(
28    memory_key="chat_history",
29    return_messages=True
30)
31
32# Add to Agent
33agent_executor = AgentExecutor(
34    agent=agent,
35    tools=tools,
36    memory=memory,
37    verbose=True
38)

Summary Memory

Riepilogo per conversazioni lunghe:

1from langchain.memory import ConversationSummaryMemory
2
3memory = ConversationSummaryMemory(
4    llm=llm,
5    memory_key="chat_history"
6)

Vector Store Memory

Memoria a lungo termine con ricerca semantica:

1from langchain.memory import VectorStoreRetrieverMemory
2from langchain.vectorstores import Chroma
3
4vectorstore = Chroma(embedding_function=embeddings)
5retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
6
7memory = VectorStoreRetrieverMemory(retriever=retriever)

Sistemi Multi-Agent

Supervisor Pattern

1from langgraph.graph import StateGraph, END
2
3def supervisor_agent(state):
4    """Supervisor distributing tasks"""
5    task = state["task"]
6    
7    if "research" in task:
8        return "researcher"
9    elif "write" in task:
10        return "writer"
11    else:
12        return "general"
13
14def researcher_agent(state):
15    """Agent conducting research"""
16    # Research logic
17    return {"research_results": results}
18
19def writer_agent(state):
20    """Agent writing content"""
21    # Writing logic
22    return {"content": content}
23
24# Create Graph
25workflow = StateGraph(AgentState)
26workflow.add_node("supervisor", supervisor_agent)
27workflow.add_node("researcher", researcher_agent)
28workflow.add_node("writer", writer_agent)
29
30workflow.add_conditional_edges("supervisor", route_to_agent)
31workflow.add_edge("researcher", "supervisor")
32workflow.add_edge("writer", END)
33
34app = workflow.compile()

Agenti Collaborativi

1┌─────────────────────────────────────────────────────┐
2│                    Orchestrator                      │
3└───────────────────────┬─────────────────────────────┘
4                        │
5        ┌───────────────┼───────────────┐
6        │               │               │
7┌───────▼───────┐ ┌─────▼─────┐ ┌──────▼──────┐
8│   Researcher  │ │  Analyst  │ │   Writer    │
9│    Agent      │ │   Agent   │ │   Agent     │
10└───────────────┘ └───────────┘ └─────────────┘

Strategie di Pianificazione

Plan-and-Execute

1from langchain_experimental.plan_and_execute import (
2    PlanAndExecute,
3    load_agent_executor,
4    load_chat_planner
5)
6
7planner = load_chat_planner(llm)
8executor = load_agent_executor(llm, tools, verbose=True)
9
10agent = PlanAndExecute(planner=planner, executor=executor)
11
12result = agent.run("Research and create a draft for a blog post")

Decomposizione dei Task

1Main Task: "Analyze products for an e-commerce site"
2
3Sub-Tasks:
41. Research competitor products
52. Perform price comparison
63. Analyze customer reviews
74. Generate report
8
9Each sub-task is handled by an agent.
10## Gestione degli Errori e Retry
11
12```python
13from langchain.callbacks import get_openai_callback
14
15class RobustAgentExecutor:
16    def __init__(self, agent_executor, max_retries=3):
17        self.executor = agent_executor
18        self.max_retries = max_retries
19    
20    def invoke(self, input_dict):
21        for attempt in range(self.max_retries):
22            try:
23                with get_openai_callback() as cb:
24                    result = self.executor.invoke(input_dict)
25                    
26                    # Token limit check
27                    if cb.total_tokens > 10000:
28                        self.summarize_memory()
29                    
30                    return result
31                    
32            except Exception as e:
33                if attempt == self.max_retries - 1:
34                    raise
35                
36                # Strategy based on Error
37                if "rate_limit" in str(e):
38                    time.sleep(60)
39                elif "context_length" in str(e):
40                    self.truncate_context()
41                else:
42                    time.sleep(2 ** attempt)

Streaming e Tempo Reale

1from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
2
3llm = ChatOpenAI(
4    streaming=True,
5    callbacks=[StreamingStdOutCallbackHandler()]
6)
7
8async def stream_agent_response(query):
9    async for event in agent_executor.astream_events(
10        {"input": query},
11        version="v1"
12    ):
13        kind = event["event"]
14        if kind == "on_chat_model_stream":
15            print(event["data"]["chunk"].content, end="")
16        elif kind == "on_tool_start":
17            print(f"\n🔧 Using tool: {event['name']}")

Considerazioni per la Produzione

Logging e Monitoraggio

1import logging
2
3class AgentLogger:
4    def __init__(self):
5        self.logger = logging.getLogger("agent")
6    
7    def log_action(self, action, input_data, output):
8        self.logger.info({
9            "action": action,
10            "input": input_data,
11            "output_length": len(str(output)),
12            "timestamp": datetime.now().isoformat()
13        })
14    
15    def log_error(self, error, context):
16        self.logger.error({
17            "error": str(error),
18            "context": context
19        })

Gestione dei Costi

1def estimate_cost(agent_run):
2    # Cost per token (GPT)
3    input_cost = agent_run.input_tokens * 0.00003
4    output_cost = agent_run.output_tokens * 0.00006
5    
6    return input_cost + output_cost
7
8# Budget limit
9MAX_BUDGET_PER_REQUEST = 0.50  # $0.50
10
11if estimate_cost(run) > MAX_BUDGET_PER_REQUEST:
12    raise BudgetExceededError()

Conclusione

Gli AI Agents consentono applicazioni autonome e proattive dell’intelligenza artificiale. Framework come LangChain e LangGraph rendono semplice sviluppare sistemi di agenti pronti per la produzione.

In Veni AI, sviluppiamo soluzioni enterprise basate su agenti.