Couchbase Partners with Arize AI to Enable Trustworthy, Production-Ready AI Agent Applications

A medida que las empresas buscan desplegar aplicaciones de agentes de IA listas para la producción, la observabilidad de Large Language Model (LLM) ha surgido como un requisito crítico para garantizar tanto el rendimiento como la confianza. Las organizaciones necesitan visibilidad sobre cómo interactúan los agentes con los datos, toman decisiones y recuperan información para mantener la fiabilidad, la seguridad y el cumplimiento. Sin una observabilidad adecuada, las empresas corren el riesgo de implantar modelos que produzcan resultados incoherentes, imprecisos o sesgados, lo que se traduce en malas experiencias de usuario e ineficiencias operativas. La nueva asociación entre Couchbase y Arize AI desempeña un papel vital a la hora de aportar sólidas capacidades de supervisión, evaluación y optimización a las aplicaciones basadas en IA.

La integración de Couchbase y Arize AI ofrece una potente solución para construir y monitorizar Retrieval Augmented Generation (RAG) y aplicaciones de agentes a escala. Al aprovechar la base de datos vectorial de alto rendimiento de Couchbase y la plataforma de observabilidad Arize AI y las capacidades de monitorización mejoradas, las empresas pueden construir, desplegar y optimizar con confianza soluciones Agentic RAG en producción.

En este blog, vamos a caminar a través de la creación de un chatbot Agentic RAG QA utilizando LangGraph y el Couchbase Catálogo de agentes de la recientemente anunciada Servicios de IA de Capella (en vista previa), y evaluando y optimizando su rendimiento con Arize AI. Este es un ejemplo tangible de cómo Couchbase y Arize AI permiten a los desarrolladores mejorar los flujos de trabajo de recuperación, mejorar la precisión de las respuestas y supervisar las interacciones impulsadas por LLM en tiempo real.

El valor de la asociación entre Couchbase y Arize AI

Al unir fuerzas, Couchbase y Arize AI están revolucionando la forma en que los desarrolladores construyen y evalúan aplicaciones de agentes de IA. Los desarrolladores pueden construir sofisticadas aplicaciones de agentes aprovechando Couchbase Capella como una única plataforma de datos para el almacenamiento en caché LLM, memoria de agentes a largo y corto plazo, casos de uso de incrustación de vectores, análisis y cargas de trabajo operativas junto con su marco de desarrollo de agentes favorito para orquestar flujos de trabajo de agentes.

Couchbase Agent Catalog mejora aún más este sistema proporcionando un almacén centralizado para flujos de trabajo multi-agente dentro de una organización que permite el almacenamiento, la gestión y el descubrimiento de varias herramientas de agente, el versionado rápido y la depuración de trazas LLM.

Para garantizar una alta fiabilidad y transparencia, Arize AI ofrece funciones de observabilidad críticas, entre las que se incluyen:

- Rastreo de llamadas a funciones del agente: Arize permite una supervisión detallada de las llamadas a funciones del agente, incluidos los pasos de recuperación y las interacciones LLM, para hacer un seguimiento de cómo se generan las respuestas.
- Comparación de conjuntos de datos: Los desarrolladores pueden crear un conjunto de datos estructurados para evaluar y comparar el rendimiento de los agentes a lo largo del tiempo.
- Evaluación del rendimiento con LLM como juez: Mediante evaluadores integrados, Arize aprovecha los LLM para evaluar la precisión de las respuestas, la pertinencia y la eficacia general del agente.
- Experimentar con estrategias de recuperación: Ajustando el tamaño de los trozos, los solapamientos y el número de documentos recuperados (valor K), los desarrolladores pueden analizar su impacto en el rendimiento del agente.
- Análisis comparativo en Arize: La plataforma permite comparar diferentes estrategias de recuperación, lo que ayuda a los equipos a determinar la configuración óptima para su agente.

La importancia de la observabilidad del LLM

Para garantizar que las aplicaciones de IA funcionen bien en producción, las empresas necesitan un marco de evaluación sólido. Las herramientas de observabilidad como Arize AI permiten a los desarrolladores:

- Evaluar los resultados del LLM en función de factores como la pertinencia, los índices de alucinación y la latencia.
- Realización de evaluaciones sistemáticas para medir el impacto de los cambios en los avisos, las modificaciones en la recuperación y los ajustes de los parámetros.
- Recopilar conjuntos de datos exhaustivos para evaluar el rendimiento en diferentes casos de uso.
- Automatizar los procesos de evaluación dentro de las canalizaciones CI/CD, garantizando una fiabilidad constante de las aplicaciones.

Utilizando un LLM como juez, Arize AI permite a los desarrolladores medir la eficacia del agente utilizando evaluadores probados previamente, técnicas de evaluación personalizadas de varios niveles y evaluaciones comparativas de rendimiento a gran escala. Mediante la ejecución de miles de evaluaciones, los equipos pueden iterar rápidamente y perfeccionar las indicaciones del LLM, los métodos de recuperación y los flujos de trabajo del agente para mejorar la calidad general de la aplicación.

Creación de un chatbot RAG Agentic QA

La RAG Agentic combina la potencia de la generación tradicional de recuperación aumentada con la toma de decisiones inteligente. En esta implementación, permitimos que un LLM decida dinámicamente si la recuperación es necesaria en función del contexto de la consulta.

Ilustración que representa el flujo de trabajo del agente de Ejemplo de RAG agéntico de Langgraph.

Aplicación paso a paso

El resto de este blog se basa en el cuaderno tutorial. Antes de crear y desplegar un agente de IA observable, tendrá que configurar su entorno de desarrollo.

Requisitos previos:

1. Para seguir este tutorial, necesitarás inscribirse en Arize y obtén tus claves Space, API y Developer. Puede ver las guía aquí. También necesitará un Clave API de OpenAI.
2. Tendrás que configurar tu clúster Couchbase haciendo lo siguiente:
  1. Cree una cuenta en Nube Couchbase
  2. Cree un clúster libre con los servicios de datos, índices y búsqueda activados*.
  3. Crear credenciales de acceso al clúster
  4. Permitir el acceso al clúster desde su máquina local
  5. Crea un cubo para almacenar tus documentos
  6. Crear un índice de búsqueda
3. Cree herramientas y avisos requeridos por los agentes usando el Catálogo de Agentes de Couchbase (para la instalación y más instrucciones, explore la documentación aquí)

*El Servicio de Búsqueda se utilizará para realizar la Búsqueda Semántica más adelante cuando utilicemos el catálogo de Agentes.

1) Crear un chatbot Agentic RAG usando LangGraph, Couchbase como almacén de vectores y Catálogo de Agentes para gestionar los agentes de IA.

Configuración de dependencias

%pip install -qU langchain-openai langchain-community langchain langgraph langgraph.prebuilt openai langchain-couchbase agentc langchain-huggingface langchain_core

%pip install -qq "arize-phoenix[evals]>=7.0.0" "arize-otel>=0.7.0" "openinference-instrumentation-openai>=0.1.18" "openinference-instrumentation-langchain>=0.1.29"

%pip instale -qU langchain-openai langchain-comunidad langchain langógrafo langógrafo.preconstruido openai langchain-couchbase agentec langchain-huggingface langchain_núcleo

%pip instale -qq "arize-phoenix[evals]>=7.0.0" "arize-otel>=0.7.0" "openinference-instrumentation-openai>=0.1.18" "openinference-instrumentation-langchain>=0.1.29"

Conexión a Couchbase

Usaremos Couchbase como nuestro almacén de vectores. He aquí cómo configurar la conexión:

from datetime import timedelta
from couchbase.auth import PasswordAuthenticator
from couchbase.cluster import Cluster
from couchbase.options import ClusterOptions
from langchain_couchbase.vectorstores import CouchbaseSearchVectorStore
from langchain_huggingface import HuggingFaceEmbeddings

# Cluster settings
CB_CONN_STRING = "your-connection-string"
CB_USERNAME = "your-username"
CB_PASSWORD = "your-password"
BUCKET_NAME = "your-bucket-name"
SCOPE_NAME = "your-scope-name"
COLLECTION_NAME = "your-collection-name"
SEARCH_INDEX_NAME = "your-search-index-name"

# Connect to couchbase cluster
auth = PasswordAuthenticator(CB_USERNAME, CB_PASSWORD)
options = ClusterOptions(auth)
options.apply_profile("wan_development")
cluster = Cluster(CB_CONN_STRING, options)
cluster.wait_until_ready(timedelta(seconds=5))

# Initialize vector store
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L12-v2")
vector_store = CouchbaseVectorStore(
    cluster=cluster,
    bucket_name=BUCKET_NAME,
    scope_name=SCOPE_NAME,
    collection_name=COLLECTION_NAME,
    embedding=embeddings,
    index_name=SEARCH_INDEX_NAME,
)

de datetime importar timedelta

de couchbase.auth importar PasswordAuthenticator

de couchbase.grupo importar Grupo

de couchbase.opciones importar ClusterOptions

de langchain_couchbase.vectorstores importar CouchbaseSearchVectorStore

de langchain_huggingface importar HuggingFaceEmbeddings

# Configuración del clúster

CB_CONN_STRING = "tu-cadena-de-conexión"

CB_USERNAME = "tu-nombre-de-usuario"

CB_PASSWORD = "tu-contraseña"

NOMBRE_CUBO = "tu-nombre-del-cubo"

SCOPE_NAME = "tu-nombre-del-ámbito"

NOMBRE_COLECCIÓN = "tu-nombre-de-colección"

NOMBRE_ÍNDICE_BÚSQUEDA = "tu-nombre-de-busqueda-index"

# Conectarse al clúster couchbase

auth = PasswordAuthenticator(CB_USERNAME, CB_PASSWORD)

opciones = ClusterOptions(auth)

opciones.aplicar_perfil("wan_development")

grupo = Grupo(CB_CONN_STRING, opciones)

grupo.wait_until_ready(timedelta(segundos=5))

# Inicializar almacén de vectores

incrustaciones = HuggingFaceEmbeddings(nombre_modelo="sentencia-transformadores/todos-MiniLM-L12-v2")

vector_store = CouchbaseVectorStore(

grupo=grupo,

nombre_cubo=NOMBRE_CUBO,

nombre_ámbito=SCOPE_NAME,

nombre_colección=NOMBRE_COLECCIÓN,

incrustación=incrustaciones,

nombre_índice=NOMBRE_ÍNDICE_BÚSQUEDA,

)

Introducción de documentos

Crearemos una función de ayuda para cargar e indexar documentos con parámetros de fragmentación configurables:

from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

def reset_vector_store(vector_store, chunk_size=1024, chunk_overlap=20):
    try:    
        # Delete existing documents
        results = vector_store.similarity_search(
            k=1000,
            query="",  
            search_options={
                "query": {"field": "metadata.source", "match": "lilian_weng_blog"}
            },
        )
        if results:
            deleted_ids = [result.id for result in results]
            vector_store.delete(ids=deleted_ids)
            
        # Load documents from URLs
        urls = [
            "https://lilianweng.github.io/posts/2024-07-07-hallucination/",
"https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
"https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
        ]
        docs = [WebBaseLoader(url).load() for url in urls]
        docs_list = [item for sublist in docs for item in sublist]

        # Use RecursiveCharacterTextSplitter with configurable parameters
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", " ", ""],  
        )
        doc_splits = text_splitter.split_documents(docs_list)
        # Add metadata to documents
        for doc in doc_splits:
            doc.metadata["source"] = "lilian_weng_blog"
            
        # Add documents to vector store
        vector_store.add_documents(doc_splits)
        return vector_store
    except ValueError as e:
        print(f"Error: {e}")

# Initialize with default settings
reset_vector_store(vector_store)

de langchain_community.cargadores_de_documentos importar WebBaseLoader

de langchain_text_splitters importar RecursiveCharacterTextSplitter

def reset_vector_store(vector_store, tamaño_trozo=1024, chunk_overlap=20):

pruebe:

# Suprimir documentos existentes

resultados = vector_store.búsqueda_similar(

k=1000,

consulta="",

opciones_de_busqueda={

"consulta": {"campo": "metadatos.fuente", "match": "lilian_weng_blog"}

)

si resultados:

Id_borrados = [resultado.id para resultado en resultados]

vector_store.borrar(ids=Id_borrados)

# Cargar documentos desde URL

urls = [

"https://lilianweng.github.io/posts/2024-07-07-hallucination/",

"https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",

"https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",

]

docs = [WebBaseLoader(url).carga() para url en urls]

docs_list = [artículo para sublista en docs para artículo en sublista]

# Utilizar RecursiveCharacterTextSplitter con parámetros configurables

divisor_texto = RecursiveCharacterTextSplitter(

tamaño_trozo=tamaño_trozo,

chunk_overlap=chunk_overlap,

separadores=["\n\n", "\n", " ", ""],

)

doc_splits = divisor_texto.dividir_documentos(docs_list)

# Añadir metadatos a los documentos

para doc en doc_splits:

doc.metadatos["fuente"] = "lilian_weng_blog"

# Añadir documentos al almacén vectorial

vector_store.añadir_documentos(doc_splits)

devolver vector_tienda

excepto ErrorValor como e:

imprimir(f"Error: {e}")

# Inicializar con ajustes por defecto

reset_vector_store(vector_store)

Configuración de la herramienta Retriever

Obtenga nuestra herramienta de recuperación del Catálogo de Agentes mediante la función agentec proveedor. En el futuro, cuando se necesiten más herramientas (y/o avisos) y la aplicación sea más compleja, el SDK y la CLI del Catálogo de Agentes podrán utilizarse para obtener automáticamente las herramientas en función del caso de uso (búsqueda semántica) o por nombre.

Para obtener instrucciones sobre cómo se creó esta herramienta y más capacidades del catálogo de agentes, consulte la documentación aquí.

import agentc.langchain
import agentc
from langchain_core.tools import tool

provider = agentc.Provider(
    decorator=lambda t: tool(t.func), 
    secrets={"CB_USERNAME": CB_USERNAME, 
            "CB_PASSWORD": CB_PASSWORD, 
            "CB_CONN_STRING": CB_CONN_STRING})

# Get the retriever tool from Agent Catalog
retriever_tool = provider.get_item(name="retriever_tool", item_type="tool")
tools = retriever_tool

importar agentec.langchain

importar agentec

de langchain_core.herramientas importar herramienta

proveedor = agentec.Proveedor(

decorador=lambda t: herramienta(t.func),

secretos={"CB_USERNAME": CB_USERNAME,

"CB_PASSWORD": CB_PASSWORD,

"CB_CONN_STRING": CB_CONN_STRING})

# Obtenga la herramienta recuperadora del Catálogo de Agentes

herramienta_recuperador = proveedor.obtener_elemento(nombre="herramienta_recuperadora", tipo_artículo="herramienta")

herramientas = herramienta_recuperador

Definición del estado del agente

Definiremos un grafo de agentes para ayudar a todos los agentes implicados a comunicarse mejor entre sí. Los agentes se comunican a través de un estado que se pasa a cada nodo y se modifica con la salida de ese nodo.

Nuestro estado será una lista de mensajes y cada nodo de nuestro grafo lo añadirá a él:

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # The add_messages function defines how an update should be processed
    # Default is to replace. add_messages says "append"
    messages: Annotated[Sequence[BaseMessage], add_messages]

de escribiendo importar Anotado, Secuencia, TypedDict

de langchain_core.mensajes importar MensajeBase

de langógrafo.gráfico.mensaje importar añadir_mensajes

clase AgenteEstado(TypedDict):

# La función add_messages define cómo debe procesarse una actualización

# Por defecto es reemplazar. add_messages dice "append"

mensajes: Anotado[Secuencia[MensajeBase], añadir_mensajes]

Creación de nodos de agente

Definiremos los componentes básicos de nuestro canal de agentes:

Nodos: Función de comprobación de relevancia, Reescritura de consultas, Agente principal, Generación de respuestas

from typing import Annotated, Literal, Sequence, TypedDict
from langchain import hub
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import tools_condition

### Relevance Checking Function
def grade_documents(state) -> Literal["generate", "rewrite"]:
    """
    Determines whether the retrieved documents are relevant to the question.
    """
    print("---CHECK RELEVANCE---")
    # Data model
    class grade(BaseModel):
        """Binary score for relevance check."""
        binary_score: str = Field(description="Relevance score 'yes' or 'no'")
    # LLM
    model = ChatOpenAI(temperature=0, model="gpt-4o", streaming=True)
    llm_with_tool = model.with_structured_output(grade)
    #fetch a prompt from Agent Catalog
    grade_documents_prompt = PromptTemplate(
       template=provider.get_item(name="grade_documents", item_type="prompt").prompt.render(),
       input_variables=["context", "question"],
    )
    # Chain
    chain = grade_documents_prompt | llm_with_tool
    messages = state["messages"]
    last_message = messages[-1]
    question = messages[0].content
    docs = last_message.content

    scored_result = chain.invoke({"question": question, "context": docs})
    score = scored_result.binary_score
    if score == "yes":
        print("---DECISION: DOCS RELEVANT---")
        return "generate"
    else:
        print("---DECISION: DOCS NOT RELEVANT---")
        print(score)
        return "rewrite"
### Main Agent Node
def agent(state):
    """
    Invokes the agent model to generate a response or use tools.
    """
    print("---CALL AGENT---")
    messages = state["messages"]
    model = ChatOpenAI(temperature=0, streaming=True, model="gpt-4-turbo")
    model = model.bind_tools(tools)
    response = model.invoke(messages)
    return {"messages": [response]}
### Query Rewriting Node
def rewrite(state):
    """
    Transform the query to produce a better question.
    """
    print("---TRANSFORM QUERY---")
    messages = state["messages"]
    question = messages[0].content
    msg = [
        HumanMessage(
            content=f""" \n 
    Look at the input and try to reason about the underlying semantic intent / meaning. \n 
    Here is the initial question:
    \n ------- \n
    {question} 
    \n ------- \n
    Formulate an improved question: """,
        )
    ]
    model = ChatOpenAI(temperature=0, model="gpt-4-0125-preview", streaming=True)
    response = model.invoke(msg)
    return {"messages": [response]}

### Response Generation Node
def generate(state):
    """
    Generate answer using retrieved documents
    """
    print("---GENERATE---")
    messages = state["messages"]
    question = messages[0].content
    last_message = messages[-1]
    docs = last_message.content

    # Prompt
    prompt = hub.pull("rlm/rag-prompt")
    llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0, streaming=True)
    rag_chain = prompt | llm | StrOutputParser()

    # Run
    response = rag_chain.invoke({"context": docs, "question": question})
    return {"messages": [response]}

de escribiendo importar Anotado, Literal, Secuencia, TypedDict

de langchain importar cubo

de langchain_core.mensajes importar MensajeBase, MensajeHumano

de langchain_core.output_parsers importar StrOutputParser

de langchain_core.solicita importar PromptTemplate

de langchain_core.pydantic_v1 importar ModeloBase, Campo

de langchain_openai importar ChatOpenAI

de langógrafo.preconstruido importar herramientas_condición

### Función de comprobación de pertinencia

def documentos_de_calificación(estado) -> Literal["generar", "reescribir"]:

"""

Determina si los documentos recuperados son pertinentes para la pregunta.

"""

imprimir("---COMPROBAR RELEVANCIA---")

# Modelo de datos

clase grado(ModeloBase):

"""Puntuación binaria para la comprobación de relevancia."""

puntuación_binaria: str = Campo(descripción="Puntuación de relevancia 'sí' o 'no'")

# LLM

modelo = ChatOpenAI(temperatura=0, modelo="gpt-4o", streaming=Verdadero)

llm_con_herramienta = modelo.con_salida_estructurada(grado)

1TP5Obtener un aviso del Catálogo de Agentes

notas_documents_prompt = PromptTemplate(

plantilla=proveedor.obtener_elemento(nombre="grado_documentos", tipo_artículo="prompt").consulte.render(),

variables_de_entrada=["contexto", "pregunta"],

)

Cadena #

cadena = notas_documents_prompt | llm_con_herramienta

mensajes = estado["mensajes"]

último_mensaje = mensajes[-1]

pregunta = mensajes[0].contenido

docs = último_mensaje.contenido

resultado_puntuado = cadena.invoque({"pregunta": pregunta, "contexto": docs})

puntuación = resultado_puntuado.binario_puntuación

si puntuación == "sí":

imprimir("---DECISIÓN: DOCS RELEVANTES---")

devolver "generar"

si no:

imprimir("---DECISIÓN: LOS DOCUMENTOS NO SON RELEVANTES---")

imprimir(puntuación)

devolver "reescribir"

### Nodo Agente Principal

def agente(estado):

"""

Invoca el modelo de agente para generar una respuesta o utilizar herramientas.

"""

imprimir("--- LLAMAR AGENTE---")

mensajes = estado["mensajes"]

modelo = ChatOpenAI(temperatura=0, streaming=Verdadero, modelo="gpt-4-turbo")

modelo = modelo.bind_tools(herramientas)

respuesta = modelo.invoque(mensajes)

devolver {"mensajes": [respuesta]}

### Nodo de reescritura de consultas

def reescribir(estado):

"""

Transformar la consulta para producir una pregunta mejor.

"""

imprimir("---TRANSFORMAR CONSULTA---")

mensajes = estado["mensajes"]

pregunta = mensajes[0].contenido

msg = [

MensajeHumano(

contenido=f""" \n

Observa la entrada e intenta razonar sobre la intención semántica / significado subyacente. \n

He aquí la pregunta inicial:

\n ------- \n

{pregunta}

\n ------- \n

Formular una pregunta mejorada: """,

)

]

modelo = ChatOpenAI(temperatura=0, modelo="gpt-4-0125-preview", streaming=Verdadero)

respuesta = modelo.invoque(msg)

devolver {"mensajes": [respuesta]}

### Nodo de generación de respuesta

def generar(estado):

"""

Generar la respuesta a partir de los documentos recuperados

"""

imprimir("---GENERAR---")

mensajes = estado["mensajes"]

pregunta = mensajes[0].contenido

último_mensaje = mensajes[-1]

docs = último_mensaje.contenido

# Prompt

consulte = cubo.tire de("rlm/rag-prompt")

llm = ChatOpenAI(nombre_modelo="gpt-4o-mini", temperatura=0, streaming=Verdadero)

cadena_de_trapo = consulte | llm | StrOutputParser()

Carrera #

respuesta = cadena_de_trapo.invoque({"contexto": docs, "pregunta": pregunta})

devolver {"mensajes": [respuesta]}

Construcción del gráfico de agentes

Ahora conectaremos los nodos en un flujo de trabajo coherente:

from langgraph.graph import END, StateGraph, START
from langgraph.prebuilt import ToolNode

# Define a new graph
workflow = StateGraph(AgentState)

# Define the nodes
workflow.add_node("agent", agent)
retrieve = ToolNode(retriever_tool)
workflow.add_node("retrieve", retrieve)
workflow.add_node("rewrite", rewrite)
workflow.add_node("generate", generate)

# Define edges
workflow.add_edge(START, "agent")

# Conditional edges based on agent's decision
workflow.add_conditional_edges(
    "agent",
    tools_condition,
    {
        "tools": "retrieve",
        END: END,
    },
)

# Conditional edges after retrieval based on document relevance
workflow.add_conditional_edges(
    "retrieve",
    grade_documents,
)
workflow.add_edge("generate", END)
workflow.add_edge("rewrite", "agent")

# Compile the graph
graph = workflow.compile()

de langógrafo.gráfico importar FIN, StateGraph, INICIO

de langógrafo.preconstruido importar ToolNode

# Definir un nuevo gráfico

flujo de trabajo = StateGraph(AgenteEstado)

# Definir los nodos

flujo de trabajo.añadir_nodo("agente", agente)

recuperar = ToolNode(herramienta_recuperador)

flujo de trabajo.añadir_nodo("recuperar", recuperar)

flujo de trabajo.añadir_nodo("reescribir", reescribir)

flujo de trabajo.añadir_nodo("generar", generar)

# Definir bordes

flujo de trabajo.añadir_borde(INICIO, "agente")

# Bordes condicionales basados en la decisión del agente

flujo de trabajo.add_conditional_edges(

"agente",

condición_herramientas,

{

"herramientas": "recuperar",

FIN: FIN,

)

# Bordes condicionales tras la recuperación basada en la relevancia del documento

flujo de trabajo.add_conditional_edges(

"recuperar",

documentos_de_calificación,

)

flujo de trabajo.añadir_borde("generar", FIN)

flujo de trabajo.añadir_borde("reescribir", "agente")

# Compilar el gráfico

gráfico = flujo de trabajo.compilar()

Visualización del gráfico de agentes

Visualicemos nuestro flujo de trabajo para entenderlo mejor:

from IPython.display import Image, display

try:
    display(Image(graph.get_graph(xray=True).draw_mermaid_png()))
except Exception:
    # This requires some extra dependencies and is optional
    pass

de IPython.mostrar importar Imagen, mostrar

pruebe:

mostrar(Imagen(gráfico.get_graph(radiografía=Verdadero).dibujar_sirena_png()))

excepto Excepción:

# Esto requiere algunas dependencias adicionales y es opcional

pase

2) Rastrear las llamadas a funciones del agente utilizando Arize, capturando las consultas de recuperación, las respuestas LLM y el uso de herramientas.

Arize proporciona una observabilidad completa para nuestro sistema de agentes. Vamos a configurar el rastreo:

from arize.otel import register
# Setup tracer provider
tracer_provider = register(
    space_id = SPACE_ID,
    api_key = API_KEY,
    project_name = "langgraph-agentic-rag",
)
# Import the automatic instrumentor from OpenInference
from openinference.instrumentation.langchain import LangChainInstrumentor
# Instrument LangChain
LangChainInstrumentor().instrument(tracer_provider=tracer_provider)

de arize.otel importar regístrese en

# Setup tracer provider

proveedor_trazador = regístrese en(

espacio_id = SPACE_ID,

clave_api = CLAVE_API,

nombre_proyecto = "langgraph-agentic-rag",

)

# Importar el instrumentador automático de OpenInference

de openinference.instrumentación.langchain importar LangChainInstrumentor

# Instrumento LangChain

LangChainInstrumentor().instrumento(proveedor_trazador=proveedor_trazador)

Ahora vamos a ejecutar el agente para ver cómo funciona:

import pprint
inputs = {
    "messages": [
        ("user", "What does Lilian Weng say about the types of adversarial attacks on LLMs?"),
    ]
}
for output in graph.stream(inputs):
    for key, value in output.items():
        pprint.pprint(f"Output from node '{key}':")
        pprint.pprint(value, indent=2, width=80, depth=None)

importar pprint

entradas = {

"mensajes": [

("usuario", "¿Qué dice Lilian Weng sobre los tipos de ataques adversarios contra los LLM?".),

]

}

para salida en gráfico.flujo(entradas):

para clave, valor en salida.artículos():

pprint.pprint(f"Salida del nodo '{clave}':")

pprint.pprint(valor, sangría=2, anchura=80, profundidad=Ninguno)

Esto ejecutará nuestro grafo de agentes y mostrará información detallada de cada nodo a medida que procesa la consulta. En Arize, podrás ver una visualización de trazas que muestra el flujo de ejecución, la latencia y los detalles de cada llamada a función.

Tracing visualization from Arize platform

Visualización del seguimiento desde Arize Platform

3) Evaluar el rendimiento generando un conjunto de datos con consultas y respuestas esperadas.

Para evaluar sistemáticamente nuestro sistema, necesitamos un conjunto de datos de referencia:

import pandas as pd
from langchain_openai import ChatOpenAI
# Define a template for generating questions
GEN_TEMPLATE = """
You are an assistant that generates Q&A questions about the content below.
The questions should involve the content, specific facts and figures,names, and elements of the story. Do not ask any questions where the answer is not in the content.
Respond with one question per line. Do not include any numbering at the beginning of each line. Do not include any category headings.
Generate 10 questions. Be sure there are no duplicate questions.
[START CONTENT]
{content}
[END CONTENT]
"""
# Load the content you want to generate questions about
content = """
Lilian Weng discusses various aspects of adversarial attacks on LLMs and prompt engineering techniques. Make sure to use Lilian Weng's name in the questions.
"""
# Format the template with the content
formatted_template = GEN_TEMPLATE.format(content=content)
# Initialize the language model
model = ChatOpenAI(model="gpt-4o", max_tokens=1300)
# Generate questions
response = model.invoke(formatted_template)
questions_content = response.content
questions = questions_content.strip().split("\n")
# Create a dataframe to store the questions
questions_df = pd.DataFrame(questions, columns=["input"])

importar pandas como pd

de langchain_openai importar ChatOpenAI

# Definir una plantilla para generar preguntas

GEN_TEMPLATE = """

Eres un asistente que genera preguntas y respuestas sobre el contenido que aparece a continuación.

Las preguntas deben referirse al contenido, hechos y cifras concretos, nombres y elementos de la historia. No hagas preguntas cuya respuesta no esté en el contenido.

Responda con una pregunta por línea. No incluya numeración al principio de cada línea. No incluya ningún encabezamiento de categoría.

Genere 10 preguntas. Asegúrese de que no hay preguntas duplicadas.

[INICIAR CONTENIDO]

{content}

[FIN DEL CONTENIDO]

"""

# Carga el contenido sobre el que quieres generar preguntas

contenido = """

Lilian Weng analiza diversos aspectos de los ataques adversarios a los LLM y las técnicas de ingeniería de avisos. Asegúrate de utilizar el nombre de Lilian Weng en las preguntas.

"""

# Formatear la plantilla con el contenido

plantilla_formateada = GEN_TEMPLATE.formato(contenido=contenido)

# Inicializar el modelo lingüístico

modelo = ChatOpenAI(modelo="gpt-4o", max_tokens=1300)

# Generar preguntas

respuesta = modelo.invoque(plantilla_formateada)

contenido_preguntas = respuesta.contenido

preguntas = contenido_preguntas.tira().dividir("\n")

# Crea un marco de datos para almacenar las preguntas

preguntas_df = pd.DataFrame(preguntas, columnas=["entrada"])

4) Evaluar el rendimiento utilizando el LLM como juez

Utilizaremos la evaluación basada en LLM para valorar la calidad de las respuestas de nuestro agente:

from phoenix.evals import (
    RAG_RELEVANCY_PROMPT_RAILS_MAP,
    RAG_RELEVANCY_PROMPT_TEMPLATE,
    QA_PROMPT_RAILS_MAP,
    QA_PROMPT_TEMPLATE,
    OpenAIModel,
    llm_classify
)

# The rails is used to hold the output to specific values based on the template
RELEVANCE_RAILS = list(RAG_RELEVANCY_PROMPT_RAILS_MAP.values())
QA_RAILS = list(QA_PROMPT_RAILS_MAP.values())

relevance_eval_df = llm_classify(
    dataframe=response_df,
    template=RAG_RELEVANCY_PROMPT_TEMPLATE,
    model=OpenAIModel(model="gpt-4o"),
    rails=RELEVANCE_RAILS,
    provide_explanation=True,
    include_prompt=True,
    concurrency=4,
)

correctness_eval_df = llm_classify(
    dataframe=response_df,
    template=QA_PROMPT_TEMPLATE,
    model=OpenAIModel(model="gpt-4o"),
    rails=QA_RAILS,
    provide_explanation=True,
    include_prompt=True,
    concurrency=4,
)

de fénix.evals importar (

RAG_RELEVANCY_PROMPT_RAILS_MAP,

RAG_RELEVANCY_PROMPT_TEMPLATE,

QA_PROMPT_RAILS_MAP,

QA_PROMPT_TEMPLATE,

OpenAIModel,

llm_clasificar

)

# Los carriles se utilizan para mantener la salida a valores específicos basados en la plantilla

RELEVANCIA_CARRILES = lista(RAG_RELEVANCY_PROMPT_RAILS_MAP.valores())

QA_RAILS = lista(QA_PROMPT_RAILS_MAP.valores())

relevancia_eval_df = llm_classify(

marco de datos=respuesta_df,

plantilla=RAG_RELEVANCY_PROMPT_TEMPLATE,

modelo=OpenAIModel(modelo="gpt-4o"),

raíles=RELEVANCIA_CARRILES,

dar_explicaciones=Verdadero,

include_prompt=Verdadero,

concurrencia=4,

)

corrección_eval_df = llm_classify(

marco de datos=respuesta_df,

plantilla=QA_PROMPT_TEMPLATE,

modelo=OpenAIModel(modelo="gpt-4o"),

raíles=QA_RAILS,

dar_explicaciones=Verdadero,

include_prompt=Verdadero,

concurrencia=4,

)

5) Experimente con los ajustes de recuperación

Ahora vamos a experimentar con diferentes configuraciones para optimizar nuestro sistema:

# Create a dataset in Arize to store our experiments
from arize.experimental.datasets import ArizeDatasetsClient
from uuid import uuid1
from arize.experimental.datasets.experiments.types import (
    ExperimentTaskResultColumnNames,
    EvaluationResultColumnNames,
)
from arize.experimental.datasets.utils.constants import GENERATIVE
# Set up the arize client
arize_client = ArizeDatasetsClient(developer_key=DEVELOPER_KEY, api_key=API_KEY)
dataset_name = "rag-experiments-" + str(uuid1())[:3]
dataset_id = arize_client.create_dataset(
    space_id=SPACE_ID,
    dataset_name=dataset_name,
    dataset_type=GENERATIVE,
    data=questions_df,
)
dataset = arize_client.get_dataset(space_id=SPACE_ID, dataset_id=dataset_id)
# Define column mappings for task and evaluation results
task_cols = ExperimentTaskResultColumnNames(
    example_id="example_id", result="output"
)
relevance_evaluator_cols = EvaluationResultColumnNames(
    label="relevance",
    explanation="relevance_explanation",
)
correctness_evaluator_cols = EvaluationResultColumnNames(
    label="correctness",
    explanation="correctness_explanation",
)

# Function to log experiments to Arize
def log_experiment_to_arize(experiment_df, experiment_name):
    experiment_df["example_id"] = dataset["id"]
    return arize_client.log_experiment(
        space_id=SPACE_ID,
        experiment_name=experiment_name + "-" + str(uuid1())[:2],
        experiment_df=experiment_df,
        task_columns=task_cols,
        evaluator_columns={
            "correctness": correctness_evaluator_cols,
            "relevance": relevance_evaluator_cols,
        },
        dataset_name=dataset_name,
    )

# Crear un conjunto de datos en Arize para almacenar nuestros experimentos

de arize.experimental.conjuntos de datos importar ArizeDatasetsClient

de uuid importar uuid1

de arize.experimental.conjuntos de datos.experimentos.tipos importar (

ExperimentTaskResultColumnNames,

EvaluationResultColumnNames,

)

de arize.experimental.conjuntos de datos.utilidades.constantes importar GENERATIVO

# Configurar el cliente arize

arize_client = ArizeDatasetsClient(clave_desarrollador=CLAVE_DESARROLLADOR, clave_api=CLAVE_API)

nombre_conjunto_datos = "trapo-experimentos-" + str(uuid1())[:3]

dataset_id = arize_client.crear_conjunto_datos(

espacio_id=SPACE_ID,

nombre_conjunto_datos=nombre_conjunto_datos,

tipo_conjunto_datos=GENERATIVO,

datos=preguntas_df,

)

conjunto de datos = arize_client.obtener_conjunto_datos(espacio_id=SPACE_ID, dataset_id=dataset_id)

# Definir asignaciones de columnas para resultados de tareas y evaluaciones

protocolos_tarea = ExperimentTaskResultColumnNames(

ejemplo_id="ejemplo_id", resultado="salida"

)

relevancia_evaluador_cols = EvaluationResultColumnNames(

etiqueta="relevancia",

explicación="pertinencia_explicación",

)

correctness_evaluator_cols = EvaluationResultColumnNames(

etiqueta="corrección",

explicación="corrección_explicación",

)

# Función para registrar experimentos en Arize

def log_experiment_to_arize(experimento_df, nombre_experimento):

experimento_df["ejemplo_id"] = conjunto de datos["id"]

devolver arize_client.log_experimento(

espacio_id=SPACE_ID,

nombre_experimento=nombre_experimento + "-" + str(uuid1())[:2],

experimento_df=experimento_df,

columnas_tarea=protocolos_tarea,

columnas_evaluador={

"corrección": correctness_evaluator_cols,

"relevancia": relevancia_evaluador_cols,

nombre_conjunto_datos=nombre_conjunto_datos,

)

Ahora haremos experimentos con distintas configuraciones:

# Experiment 1: Chunks of 1024 tokens, k=2
reset_vector_store(vector_store, chunk_size=1024, chunk_overlap=20)
k_2_chunk_1024_overlap_20 = run_rag(questions_df, k_value=2)
k_2_chunk_1024_overlap_20 = run_evaluators(k_2_chunk_1024_overlap_20)
# Experiment 2: Chunks of 1024 tokens, k=4
k_4_chunk_1024_overlap_20 = run_rag(questions_df, k_value=4)
k_4_chunk_1024_overlap_20 = run_evaluators(k_4_chunk_1024_overlap_20)
# Experiment 3: Smaller chunks (200 tokens), k=2
reset_vector_store(vector_store, chunk_size=200, chunk_overlap=20)
k_2_chunk_200_overlap_20 = run_rag(questions_df, k_value=2)
k_2_chunk_200_overlap_20 = run_evaluators(k_2_chunk_200_overlap_20)
# Experiment 4: Medium chunks (500 tokens), k=2
reset_vector_store(vector_store, chunk_size=500, chunk_overlap=20)
k_2_chunk_500_overlap_20 = run_rag(questions_df, k_value=2)
k_2_chunk_500_overlap_20 = run_evaluators(k_2_chunk_500_overlap_20)
# Log all experiments to Arize
log_experiment_to_arize(k_2_chunk_1024_overlap_20, "k_2_chunk_1024_overlap_20")
log_experiment_to_arize(k_4_chunk_1024_overlap_20, "k_4_chunk_1024_overlap_20")
log_experiment_to_arize(k_2_chunk_200_overlap_20, "k_2_chunk_200_overlap_20")
log_experiment_to_arize(k_2_chunk_500_overlap_20, "k_2_chunk_500_overlap_20")

# Experimento 1: Trozos de 1024 fichas, k=2

reset_vector_store(vector_store, tamaño_trozo=1024, chunk_overlap=20)

k_2_chunk_1024_overlap_20 = run_rag(preguntas_df, k_valor=2)

k_2_chunk_1024_overlap_20 = run_evaluators(k_2_chunk_1024_overlap_20)

# Experimento 2: Trozos de 1024 fichas, k=4

k_4_chunk_1024_overlap_20 = run_rag(preguntas_df, k_valor=4)

k_4_chunk_1024_overlap_20 = run_evaluators(k_4_chunk_1024_overlap_20)

# Experimento 3: Trozos más pequeños (200 fichas), k=2

reset_vector_store(vector_store, tamaño_trozo=200, chunk_overlap=20)

k_2_chunk_200_overlap_20 = run_rag(preguntas_df, k_valor=2)

k_2_chunk_200_overlap_20 = run_evaluators(k_2_chunk_200_overlap_20)

# Experimento 4: Trozos medianos (500 fichas), k=2

reset_vector_store(vector_store, tamaño_trozo=500, chunk_overlap=20)

k_2_chunk_500_overlap_20 = run_rag(preguntas_df, k_valor=2)

k_2_chunk_500_overlap_20 = run_evaluators(k_2_chunk_500_overlap_20)

# Registrar todos los experimentos en Arize

log_experiment_to_arize(k_2_chunk_1024_overlap_20, "k_2_chunk_1024_overlap_20")

log_experiment_to_arize(k_4_chunk_1024_overlap_20, "k_4_chunk_1024_overlap_20")

log_experiment_to_arize(k_2_chunk_200_overlap_20, "k_2_chunk_200_overlap_20")

log_experiment_to_arize(k_2_chunk_500_overlap_20, "k_2_chunk_500_overlap_20")

6) Comparar experimentos en Arize

Después de ejecutar todos los experimentos, ya puedes verlos y compararlos en la interfaz de Arize. Los experimentos deben ser visibles en su espacio de trabajo Arize bajo el nombre del conjunto de datos que hemos creado anteriormente.

Vista comparativa de experimentos desde Arize Platform

En Arize, puedes:

1. Comparar las métricas de rendimiento global entre distintas configuraciones.
2. Analizar el rendimiento por pregunta para identificar patrones
3. Examinar los detalles de la traza para comprender el flujo de ejecución
4. Ver las puntuaciones de relevancia y corrección de cada experimento
5. Ver las explicaciones de las decisiones de evaluación
6. Evalúe los resultados utilizando un LLM como juez para puntuar la pertinencia y la corrección de las respuestas.
7. Optimice la configuración de recuperación experimentando con tamaños de trozos, configuraciones de solapamiento y límites de recuperación de documentos.
8. Compare y analice experimentos en Arize para determinar las configuraciones de mejor rendimiento.

Innovar con Couchbase y Arize AI

La integración de Couchbase y Arize empodera a las empresas para construir aplicaciones GenAI robustas y listas para producción con fuertes capacidades de observabilidad y optimización. Al aprovechar Agentic RAG con decisiones de recuperación monitorizadas, las organizaciones pueden mejorar la precisión, reducir las alucinaciones y garantizar un rendimiento óptimo a lo largo del tiempo.

A medida que las empresas continúan ampliando los límites de GenAI, combinar el almacenamiento vectorial de alto rendimiento con la observabilidad de la IA será clave para desplegar aplicaciones fiables y escalables. Con Couchbase y Arize, las organizaciones tienen las herramientas necesarias para afrontar con confianza los retos del despliegue de GenAI en la empresa.

Recursos adicionales

- Empieza con Plataforma para desarrolladores Couchbase Capella gratis
- Solicite una demostración de Arize AI aquí
- Lea más entradas y tutoriales en IA Generativa (GenAI)

Platform

Self-Managed

Services

Capabilities

Why Couchbase?

Migrate to Capella

By Use Case

By Industry

By Application Need

Popular Docs

By Developer Role

Quickstart

Resource Center

About

Partnerships

Our Services

Partners: Register a Deal

Ready to register a deal with Couchbase?

Marriott

Todos los mensajes

Couchbase se asocia con Arize AI para permitir aplicaciones de agentes de IA fiables y listas para la producción

El valor de la asociación entre Couchbase y Arize AI

La importancia de la observabilidad del LLM

Creación de un chatbot RAG Agentic QA

Aplicación paso a paso

Requisitos previos:

1) Crear un chatbot Agentic RAG usando LangGraph, Couchbase como almacén de vectores y Catálogo de Agentes para gestionar los agentes de IA.

Configuración de dependencias

Conexión a Couchbase

Introducción de documentos

Configuración de la herramienta Retriever

Definición del estado del agente

Creación de nodos de agente

Construcción del gráfico de agentes

Visualización del gráfico de agentes

2) Rastrear las llamadas a funciones del agente utilizando Arize, capturando las consultas de recuperación, las respuestas LLM y el uso de herramientas.

3) Evaluar el rendimiento generando un conjunto de datos con consultas y respuestas esperadas.

4) Evaluar el rendimiento utilizando el LLM como juez

5) Experimente con los ajustes de recuperación

6) Comparar experimentos en Arize

Innovar con Couchbase y Arize AI

Recursos adicionales

Autor

Publicado por Richard Young - Dir. Arquitectura de soluciones para socios, Arize AI

Dejar una respuesta Cancelar respuesta