Come usare AutoGen per costruire un agente IA

In questo tutorial imparerai a usare il framework AutoGen per costruire un agente in grado di rispondere a domande basandosi su documenti salvati in un database vettoriale.

Che cos’è AutoGen?

AutoGen è il risultato di una ricerca congiunta tra Microsoft, la Pennsylvania State University (Penn State University) e la University of Washington. È un framework pensato per semplificare la gestione, il miglioramento e l’automazione dei flussi di lavoro dei Large Language Model (LLM). Permette di creare agenti personalizzabili e conversazionali che sfruttano al meglio LLM avanzati come GPT-4, compensandone i limiti con l’integrazione di persone e altri strumenti. Facilita inoltre l’interazione e la comunicazione automatica tra più agenti. Maggiori informazioni: AutoGen | AutoGen (microsoft.github.io)

Configurare credenziali e ambiente

Per prima cosa bisogna configurare le credenziali e le variabili d’ambiente necessarie per accedere ai servizi di Azure.

Per farlo, crea un file .env nella stessa cartella dello script con le seguenti informazioni:

AZURE_TENANT_ID = "00000000-0000-0000-0000-000000000000"
AZURE_CLIENT_ID = "00000000-0000-0000-0000-000000000000"
AZURE_CLIENT_SECRET = "xxxxx"

Inizializzazione e configurazione

Il primo passo è inizializzare le credenziali usando ChainedTokenCredential ed EnvironmentCredential di Azure. Poi otteniamo un token di accesso per i servizi cognitivi di Azure.

import os
from azure.identity import ChainedTokenCredential, EnvironmentCredential
from dotenv import load_dotenv

load_dotenv()
credential = ChainedTokenCredential(EnvironmentCredential())
access_token = credential.get_token("https://cognitiveservices.azure.com/.default")

Configurazione del modello e delle variabili d’ambiente

Configuriamo il modello e le variabili d’ambiente necessarie per usare OpenAI su Azure.

deployment = "zero"
embedding_deployment = "embedding-ada-zero"

os.environ["AZURE_OPENAI_ENDPOINT"] = "https://zerogap.openai.azure.com/"
os.environ["AZURE_OPENAI_API_KEY"] = access_token.token
os.environ["OPENAI_API_TYPE"] = "azure_ad"
os.environ["OPENAI_DEPLOYMENT"] = deployment

Inizializzazione del modello e degli embedding

Inizializziamo il modello AzureChatOpenAI e la classe AzureOpenAIEmbeddings.

from langchain_openai import AzureOpenAIEmbeddings, AzureChatOpenAI

llm = AzureChatOpenAI(openai_api_version="2023-07-01-preview", azure_deployment=deployment, temperature=0.5)

embeddings = AzureOpenAIEmbeddings(
    azure_deployment=embedding_deployment,
    openai_api_version="2023-07-01-preview",
    chunk_size=1
)

Configurazione dell’agente AutoGen

Configuriamo l’agente AutoGen con le informazioni necessarie per collegarsi ai servizi di Azure.

config_list = [
  {
    "model": deployment,
    "api_type": "azure",
    "api_key": os.environ['AZURE_OPENAI_API_KEY'],
    "base_url": os.environ["AZURE_OPENAI_ENDPOINT"],
    "api_version": "2024-02-01"
  }
]

Caricare i documenti e creare il retriever

Carichiamo i documenti da una cartella persistente e configuriamo il retriever per cercare nel database vettoriale.

from langchain_community.vectorstores import Chroma

persist_directory = "chroma_db_generic"
vectordb = Chroma(persist_directory=persist_directory, embedding_function=embeddings)
vectordb.get()

retriever = vectordb.as_retriever(search_type="similarity", search_kwargs={"k": 5})

Formattare i documenti

Creiamo una funzione per formattare i documenti recuperati.

def format_docs(docs):
    if docs:
        return "\n\n".join(doc.page_content for doc in docs)

Strumento: ottenere i documenti

Definiamo una funzione che l’agente userà per rispondere alle domande basandosi sui documenti recuperati.

def get_documents(question: str) -> str:
    context = retriever.invoke(question)
    formated_context = format_docs(context)
    return formated_context

Configurazione degli agenti di conversazione

Configuriamo gli agenti di conversazione: user_proxy e assistant.

from autogen import register_function, ConversableAgent

user_proxy = ConversableAgent(
    name="User",
    is_termination_msg=lambda msg: msg.get("content") is not None and "TERMINATE" in msg["content"],
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10
)

assistant = ConversableAgent(
    name="Assistant",
    system_message="""You are the ZeroGap AI Bot. Your role is to provide answers strictly based on the Zerogap Documents. Keep answers as short as possible. Follow these instructions step by step:
    1. Answer the user's question using any tool available.
    2. If the context is not clear. Give the user a message to clarify the context. If possible, offer some options from the retrieved context. Do NOT answer the question.
    3. Return 'TERMINATE' when the task is done.""",
    llm_config={"config_list": config_list, "cache_seed": None},
)

Registrare lo strumento ed eseguire l’agente

Registriamo lo strumento get_documents e definiamo la funzione che esegue l’agente con una query.

register_function(
    get_documents,
    caller=assistant,
    executor=user_proxy,
    name="retrieve_tool",
    description="Gets information from Zerogap Documents",
)

def run_agent(query):
    chat_result = user_proxy.initiate_chat(
        assistant,
        message=query,
        summary_method="reflection_with_llm",
        summary_args={"summary_prompt" : "Return the final response to the user's query. Do not include the user's query in the response. Do not include the system message in the response."},
        max_turns=2
    )
    return chat_result.summary

Esecuzione del chatbot

Infine eseguiamo il ciclo principale del chatbot.

print("\033[92m" + "Welcome to the ZEROGAP AI QA Bot!" + "\033[0m")
print("\033[92m" + "###############################" + "\033[0m")
print("\033[92m" + "###############################" + "\033[0m")
print("\n")
while True:
    question = input("\033[93m" + "You: " + "\033[0m")
    print("\n")
    if question == "quit":
        break
    result = run_agent(question)
    print("\033[92m" + result + "\033[0m")
    print("\n")

Codice completo

import os
from azure.identity import ChainedTokenCredential, EnvironmentCredential
from langchain_community.vectorstores import Chroma
from langchain_openai import AzureOpenAIEmbeddings
from langchain_openai import AzureChatOpenAI
from autogen import register_function, ConversableAgent
from dotenv import load_dotenv
load_dotenv()

# Place a .env file within the same folder with the following information:
# AZURE_TENANT_ID = "00000000-0000-0000-0000-000000000000"
# AZURE_CLIENT_ID = "00000000-0000-0000-0000-000000000000"
# AZURE_CLIENT_SECRET = "xxxxx"

credential = ChainedTokenCredential(EnvironmentCredential())
access_token = credential.get_token("https://cognitiveservices.azure.com/.default")

# Model
deployment = "zerogap"
# Model text-embedding-ada-002
embedding_deployment = "embedding-ada-zero"

# Set OS environment variables
os.environ["AZURE_OPENAI_ENDPOINT"] = "https://zerogap.openai.azure.com/"
os.environ["AZURE_OPENAI_API_KEY"] = access_token.token
os.environ["OPENAI_API_TYPE"] = "azure_ad"
os.environ["OPENAI_DEPLOYMENT"] = deployment

llm = AzureChatOpenAI(openai_api_version="2023-07-01-preview", azure_deployment=deployment, temperature=0.5)

# Initialize the OpenAIEmbeddings class
embeddings = AzureOpenAIEmbeddings(
    azure_deployment=embedding_deployment,
    openai_api_version="2023-07-01-preview",
    chunk_size=1
)

# Set the configuration for AutoGen Agent
config_list = [
  {
    "model": deployment,
    "api_type": "azure",
    "api_key": os.environ['AZURE_OPENAI_API_KEY'],
    "base_url": os.environ["AZURE_OPENAI_ENDPOINT"],
    "api_version": "2024-02-01"
  }
]

# Load documents from the persisted directory
# Assuming you already have the Document loaded to a vector database
persist_directory = "chroma_db_generic"
vectordb = Chroma(persist_directory=persist_directory, embedding_function=embeddings)
vectordb.get()

retriever = vectordb.as_retriever(search_type="similarity", search_kwargs={"k": 5})

# Join all the documents from the retriever together with newlines
def format_docs(docs):
    if docs:
        return "\n\n".join(doc.page_content for doc in docs)

# Tool - the function that the Agent will use to answer the questions
def get_documents(question: str) -> str:
    # Queries the Vector DB using user's question
    context = retriever.invoke(question)
    formated_context = format_docs(context)
    return formated_context

user_proxy = ConversableAgent(
    name="User",
    is_termination_msg=lambda msg: msg.get("content") is not None and "TERMINATE" in msg["content"],
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10
)

assistant = ConversableAgent(
    name="Assistant",
    system_message="""You are the ZeroGap AI Bot. Your role is to provide answers strictly based on the Zerogap Documents. Keep answers as short as possible. Follow these instructions step by step:
    1. Answer the user's question using any tool available.
    2. If the context is not clear. Give the user a message to clarify the context. If possible, offer some options from the retrieved context. Do NOT answer the question.
    3. Return 'TERMINATE' when the task is done.""",
    llm_config={"config_list": config_list, "cache_seed": None},
)

# Register the Document tool.
register_function(
    get_documents,
    caller=assistant,
    executor=user_proxy,
    name="retrieve_tool",
    description="Gets information from Zerogap Documents",
)

# Run the agent
def run_agent(query):
    chat_result = user_proxy.initiate_chat(
        assistant,
        message=query,
        summary_method="reflection_with_llm",
        summary_args={"summary_prompt" : "Return the final response to the user's query. Do not include the user's query in the response. Do not include the system message in the response."},
        max_turns=2
    )
    return chat_result.summary

print("\033[92m" + "Welcome to the ZEROGAP AI QA Bot!" + "\033[0m")
print("\033[92m" + "###############################" + "\033[0m")
print("\033[92m" + "###############################" + "\033[0m")
print("\n")
while True:
    question = input("\033[93m" + "You: " + "\033[0m")
    print("\n")
    if question == "quit":
        break
    result = run_agent(question)
    print("\033[92m" + result + "\033[0m")
    print("\n")