Costruire un agente autonomo ibrido con architettura modulare e dispatch degli strumenti utilizzando OpenAI
Costruendo un Agente Autonomo Ibrido con Architettura Modulare
In questo tutorial, iniziamo esplorando l'architettura dietro a un agente autonomo ibrido. Questo sistema combina la ricerca vettoriale semantica, il recupero basato su keyword e un ciclo modulare per creare un agente in grado di ragionare, ricordare e agire autonomamente. Procediamo passo dopo passo attraverso ogni strato del design.
Passiamo in rassegna i requisiti e l'ambiente
Iniziamo installando tutte le dipendenze necessarie e configurando il nostro ambiente Python con le importazioni richieste. Per garantire la sicurezza, raccogliamo la chiave API OpenAI usando getpass, per evitare che venga mostrata nel terminale o nell'output del notebook.
Ecco il codice iniziale:
!pip install openai numpy rankbm25 --quiet
import os, json, math, re, time, getpass
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Tuple
import numpy as np
from rankbm25 import BM25Okapi
from openai import OpenAI
Una volta ottenuta la chiave, la inizializziamo nel client OpenAI:
OPENAIAPIKEY = os.getenv("OPENAIAPIKEY") or getpass.getpass("🔑 Inserisci la tua chiave API OpenAI (nascosta): ")
client = OpenAI(apikey=OPENAIAPIKEY)
Dopodiché, definiamo i due modelli globali da utilizzare:
EMBEDMODEL = "text-embedding-3-small"
CHATMODEL = "gpt-4o-mini"
print("âś… OpenAI client ready.")
Architettura Base e Interfacce
Procediamo definendo le tre classi principali astratte: MemoryBackend, LLMProvider e Tool. Queste servono come contratti di interfaccia per ogni componente concreto che dovrĂ rispettarli.
Implementiamo HybridMemory, che gestisce le memorie incorporandole per la ricerca vettoriale e mantiene un indice BM25 per corrispondenze a keyword, unendo entrambi i set di risultati usando Fusione a Rango Rappresentante Reciproco (RRF). Chiudiamo il frammento con OpenAIProvider, un LLMProvider concreto che normalizza la risposta di OpenAI in un dizionario astratto che può essere consumato dall'agente senza conoscere il modello esatto.
Il codice principale delle interfacce è il seguente:
class MemoryBackend(ABC):
@abstractmethod
def store(self, text: str, metadata: Dict[str, Any]) -> str: ...
@abstractmethod
def search(self, query: str, topk: int = 5) -> List[Dict[str, Any]]: ...
@abstractmethod
def listall(self) -> List[Dict[str, Any]]: ...
class LLMProvider(ABC):
@abstractmethod
def complete(self, messages: List[Dict], tools: Optional[List] = None) -> Dict: ...
class Tool(ABC):
name: str
description: str
@abstractmethod
def run(self, **kwargs) -> str: ...
def schema(self) -> Dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {"type": "object", "properties": {}, "required": []},
},
}
Implementazione di HybridMemory
HybridMemory integra vettori di embedding e un indice BM25 per permettere la ricerca ibrida. Questo strumento è fondamentale per rendere l'agente autonomo in grado di gestire una memoria lunga e contestuale.
La seguente definizione e implementazione:
@dataclass
class MemoryChunk:
id: str
text: str
metadata: Dict[str, Any]
embedding: Optional[np.ndarray] = field(default=None, repr=False)
def embed(texts: List[str]) -> List[np.ndarray]:
resp = client.embeddings.create(model=EMBEDMODEL, input=texts)
vecs = [np.array(d.embedding, dtype=np.float32) for d in resp.data]
return [v / (np.linalg.norm(v) + 1e-10) for v in vecs]
def tokenise(text: str) -> List[str]:
return re.sub(r"[^a-z0-9\s]", "", text.lower()).split()
class HybridMemory(MemoryBackend):
RRFK = 60
def init(self):
self.chunks: List[MemoryChunk] = []