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

À medida que as empresas buscam implementar aplicativos de agentes de IA prontos para produção, a observabilidade do modelo de linguagem grande (LLM) surgiu como um requisito essencial para garantir o desempenho e a confiança. As organizações precisam ter visibilidade de como os agentes interagem com os dados, tomam decisões e recuperam informações para manter a confiabilidade, a segurança e a conformidade. Sem a observabilidade adequada, as empresas correm o risco de implantar modelos que produzam resultados inconsistentes, imprecisos ou tendenciosos, levando a experiências de usuário ruins e ineficiências operacionais. A nova parceria entre Couchbase e Arize AI desempenha um papel fundamental ao trazer recursos robustos de monitoramento, avaliação e otimização para aplicativos orientados por IA.

A integração do Couchbase e do Arize AI oferece uma solução avançada para a criação e o monitoramento de aplicativos RAG (Retrieval Augmented Generation) e de agentes em escala. Ao aproveitar o banco de dados vetorial de alto desempenho do Couchbase, a plataforma de observabilidade da Arize AI e os recursos de monitoramento aprimorados, as empresas podem criar, implantar e otimizar com confiança as soluções Agentic RAG na produção.

Neste blog, vamos criar um chatbot de controle de qualidade Agentic RAG usando o LangGraph e o Couchbase Catálogo de agentes componente do recém-anunciado Serviços de IA da Capella (em visualização), além de avaliar e otimizar seu desempenho com o Arize AI. Esse é um exemplo tangível de como o Couchbase e o Arize AI permitem que os desenvolvedores aprimorem os fluxos de trabalho de recuperação, melhorem a precisão da resposta e monitorem as interações alimentadas por LLM em tempo real.

O valor da parceria entre a Couchbase e a Arize AI

Ao unir forças, a Couchbase e a Arize AI estão revolucionando a forma como os desenvolvedores criam e avaliam aplicativos de agentes de IA. Os desenvolvedores podem criar aplicativos de agente sofisticados aproveitando o Couchbase Capella como uma plataforma de dados única para cache LLM, memória de agente de longo e curto prazo, casos de uso de incorporação de vetor, análises e cargas de trabalho operacionais, juntamente com sua estrutura de desenvolvimento de agente favorita para orquestrar fluxos de trabalho de agente.

O Couchbase Agent Catalog aprimora ainda mais esse sistema, fornecendo um armazenamento centralizado para fluxos de trabalho de vários agentes em uma organização que permite o armazenamento, o gerenciamento e a descoberta de várias ferramentas de agentes, o controle de versão imediato e a depuração de rastreamento LLM.

Para garantir alta confiabilidade e transparência, o Arize AI oferece recursos essenciais de observabilidade, incluindo:

- Rastreamento de chamadas de função do agente: O Arize permite o monitoramento detalhado das chamadas de função do agente, incluindo etapas de recuperação e interações LLM, para rastrear como as respostas são geradas.
- Benchmarking de conjunto de dados: Os desenvolvedores podem criar um conjunto de dados estruturado para avaliar e comparar o desempenho do agente ao longo do tempo.
- Avaliação de desempenho com o LLM como juiz: Usando avaliadores integrados, o Arize aproveita os LLMs para avaliar a precisão da resposta, a relevância e a eficácia geral do agente.
- Experimentando estratégias de recuperação: Ao ajustar os tamanhos dos blocos, as sobreposições e o número de documentos recuperados (valor K), os desenvolvedores podem analisar seu impacto no desempenho do agente.
- Análise comparativa no Arize: A plataforma permite comparações lado a lado de diferentes estratégias de recuperação, ajudando as equipes a determinar a configuração ideal para seu agente.

A importância da observabilidade do LLM

Para garantir que os aplicativos de IA tenham um bom desempenho na produção, as empresas precisam de uma estrutura de avaliação robusta. Ferramentas de observabilidade como o Arize AI permitem que os desenvolvedores:

- Avaliar os resultados do LLM com base em fatores como relevância, taxas de alucinação e latência
- Conduzir avaliações sistemáticas para medir o impacto de alterações imediatas, modificações de recuperação e ajustes de parâmetros
- Organizar conjuntos de dados abrangentes para avaliar o desempenho em diferentes casos de uso
- Automatize os processos de avaliação nos pipelines de CI/CD, garantindo a confiabilidade consistente dos aplicativos

Usando um LLM como juiz, o Arize AI permite que os desenvolvedores meçam a eficácia do agente usando avaliadores pré-testados, técnicas de avaliação personalizadas em vários níveis e benchmarking de desempenho em grande escala. Ao executar milhares de avaliações, as equipes podem iterar rapidamente e refinar as solicitações do LLM, os métodos de recuperação e os fluxos de trabalho do agente para melhorar a qualidade geral do aplicativo.

Criação de um chatbot de controle de qualidade RAG autêntico

O Agentic RAG combina o poder da geração tradicional aumentada por recuperação com a tomada de decisão inteligente. Nessa implementação, permitimos que um LLM decida dinamicamente se a recuperação é necessária com base no contexto da consulta.

Ilustração que descreve o fluxo de trabalho do agente de Exemplo de RAG autêntico da Langgraph.

Implementação passo a passo

O restante deste blog se baseia no texto que o acompanha caderno de tutoriais. Antes de criar e implantar um agente de IA observável, você precisará configurar seu ambiente de desenvolvimento.

Pré-requisitos:

1. Para acompanhar este tutorial, você precisará inscrever-se no Arize e obtenha suas chaves de espaço, API e desenvolvedor. Você pode ver as guia aqui. Você também precisará de um Chave da API da OpenAI.
2. Você precisará configurar seu cluster do Couchbase fazendo o seguinte:
  1. Crie uma conta em Nuvem do Couchbase
  2. Crie um cluster gratuito com os serviços de dados, índice e pesquisa ativados*
  3. Criar credenciais de acesso ao cluster
  4. Permitir acesso ao cluster a partir de seu computador local
  5. Crie um bucket para armazenar seus documentos
  6. Criar um índice de pesquisa
3. Crie ferramentas e prompts exigidos pelos agentes usando o Couchbase Agent Catalog (para obter instruções de instalação e outras, explore a documentação aqui)

*O serviço de pesquisa será usado para realizar a pesquisa semântica posteriormente, quando usarmos o catálogo de agentes.

1) Criar um chatbot Agentic RAG usando LangGraph, Couchbase como armazenamento de vetores e Agent Catalog para gerenciar agentes de IA

Configuração de dependências

%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"

%tubulação instalar -qU langchain-openai langchain-comunidade langchain langógrafo langógrafo.pré-construído openai langchain-couchbase agente langchain-cara de abraço langchain_núcleo

%tubulação instalar -qq "arize-phoenix[evals]>=7.0.0" "arize-otel>=0.7.0" "openinference-instrumentation-openai>=0.1.18" "openinference-instrumentation-langchain>=0.1.29"

Conexão com o Couchbase

Usaremos o Couchbase como nosso armazenamento de vetores. Veja a seguir como configurar a conexão:

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 data e hora importação timedelta

de couchbase.autenticação importação PasswordAuthenticator

de couchbase.agrupamento importação Aglomerado

de couchbase.opções importação ClusterOptions

de langchain_couchbase.lojas de vetores importação CouchbaseSearchVectorStore

de langchain_huggingface importação HuggingFaceEmbeddings

Configurações do cluster #

CB_CONN_STRING = "your-connection-string" (sua cadeia de conexão)

CB_USERNAME = "seu nome de usuário"

CB_PASSWORD = "your-password" (sua senha)

BUCKET_NAME = "seu-nome-do-balde"

SCOPE_NAME = "your-scope-name" (seu nome de escopo)

NOME_DA_COLEÇÃO = "your-collection-name" (nome de sua coleção)

NOME_DO_ÍNDICE_DE_PESQUISA = "your-search-index-name" (seu nome de índice de pesquisa)

# Conectar-se ao cluster do couchbase

autenticação = PasswordAuthenticator(CB_USERNAME, CB_PASSWORD)

opções = ClusterOptions(autenticação)

opções.apply_profile("wan_development")

agrupamento = Aglomerado(CB_CONN_STRING, opções)

agrupamento.wait_until_ready(timedelta(segundos=5))

# Inicializar o armazenamento de vetores

incorporações = HuggingFaceEmbeddings(nome_do_modelo="sentence-transformers/all-MiniLM-L12-v2")

vector_store = CouchbaseVectorStore(

agrupamento=agrupamento,

nome_do_balde=BUCKET_NAME,

nome_do_escopo=SCOPE_NAME,

nome_da_coleção=NOME_DA_COLEÇÃO,

incorporação=incorporações,

nome_do_índice=NOME_DO_ÍNDICE_DE_PESQUISA,

)

Ingestão de documentos

Criaremos uma função auxiliar para carregar e indexar documentos com parâmetros de fragmentação configuráveis:

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.carregadores de documentos importação Carregador WebBase

de langchain_text_splitters importação RecursiveCharacterTextSplitter

def reset_vector_store(vector_store, tamanho do bloco=1024, chunk_overlap=20):

tentar:

# Excluir documentos existentes

resultados = vector_store.similarity_search(

k=1000,

consulta="",

opções_de_pesquisa={

"query" (consulta): {"campo": "metadata.source", "match": "lilian_weng_blog"}

)

se resultados:

deleted_ids = [resultado.id para resultado em resultados]

vector_store.excluir(IDs=deleted_ids)

# Carregar documentos de 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/",

]

documentos = [Carregador WebBase(url).carregar() para url em urls]

lista_de_documentos = [item para Sublista em documentos para item em Sublista]

# Use RecursiveCharacterTextSplitter com parâmetros configuráveis

divisor de texto = RecursiveCharacterTextSplitter(

tamanho do bloco=tamanho do bloco,

chunk_overlap=chunk_overlap,

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

)

doc_splits = divisor de texto.split_documents(lista_de_documentos)

# Adicionar metadados aos documentos

para doc em doc_splits:

doc.metadados["fonte"] = "lilian_weng_blog"

# Adicionar documentos à loja de vetores

vector_store.add_documents(doc_splits)

retorno vetor_loja

exceto ValueError como e:

impressão(f"Error: {e}")

# Inicializar com as configurações padrão

reset_vector_store(vector_store)

Configuração da ferramenta Retriever

Busque nossa ferramenta retriever no Catálogo de Agentes usando o comando agente provedor. No futuro, quando mais ferramentas (e/ou prompts) forem necessárias e o aplicativo se tornar mais complexo, o SDK e a CLI do Agent Catalog poderão ser usados para buscar automaticamente as ferramentas com base no caso de uso (pesquisa semântica) ou pelo nome.

Para obter instruções sobre como essa ferramenta foi criada e mais recursos do catálogo de agentes, consulte a documentação aqui.

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

importação agente.langchain

importação agente

de langchain_core.ferramentas importação ferramenta

provedor = agente.Provedor(

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

segredos={"CB_USERNAME": CB_USERNAME,

"CB_PASSWORD": CB_PASSWORD,

"CB_CONN_STRING": CB_CONN_STRING})

# Obtenha a ferramenta de recuperação no Agent Catalog

retriever_tool = provedor.get_item(nome="retriever_tool", tipo_de_item="ferramenta")

ferramentas = retriever_tool

Definição do estado do agente

Definiremos um gráfico de agentes para ajudar todos os agentes envolvidos a se comunicarem melhor uns com os outros. Os agentes se comunicam por meio de um estado que é passado para cada nó e modificado com a saída desse nó.

Nosso estado será uma lista de mensagens e cada nó em nosso gráfico será anexado a ele:

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 digitação importação Anotado, Sequência, TypedDict

de langchain_core.mensagens importação BaseMessage

de langógrafo.gráfico.mensagem importação add_messages

classe Estado do agente(TypedDict):

# A função add_messages define como uma atualização deve ser processada

# O padrão é substituir. add_messages diz "append"

mensagens: Anotado[Sequência[BaseMessage], add_messages]

Criação de nós de agente

Definiremos os principais componentes de nosso pipeline de agentes:

Nós: Função de verificação de relevância, reescritor de consultas, agente principal, geração de respostas

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 digitação importação Anotado, Literal, Sequência, TypedDict

de langchain importação centro

de langchain_core.mensagens importação BaseMessage, HumanMessage

de langchain_core.analisadores de saída importação StrOutputParser

de langchain_core.avisos importação PromptTemplate

de langchain_core.pydantic_v1 importação BaseModel, Campo

de langchain_openai importação ChatOpenAI

de langógrafo.pré-construído importação ferramentas_condição

### Função de verificação de relevância

def documentos de notas(estado) -> Literal["gerar", "rewrite" (reescrever)]:

"""

Determina se os documentos recuperados são relevantes para a pergunta.

"""

impressão("---VERIFICAR RELEVÂNCIA---")

# Modelo de dados

classe grau(BaseModel):

"""Pontuação binária para verificação de relevância."""

binary_score: str = Campo(descrição="Pontuação de relevância 'sim' ou 'não'")

# LLM

modelo = ChatOpenAI(temperatura=0, modelo="gpt-4o", transmissão=Verdadeiro)

llm_with_tool = modelo.with_structured_output(grau)

1TP5Busca um prompt do Agent Catalog

documentos_de_grau_prompt = PromptTemplate(

modelo=provedor.get_item(nome="grade_documents", tipo_de_item="prompt").imediato.renderizar(),

variáveis_de_entrada=["contexto", "pergunta"],

)

Corrente #

cadeia = documentos_de_grau_prompt | llm_with_ferramenta

mensagens = estado["mensagens"]

last_message = mensagens[-1]

pergunta = mensagens[0].conteúdo

documentos = last_message.conteúdo

resultado_marcado = cadeia.invocar({"pergunta": pergunta, "contexto": documentos})

pontuação = resultado_marcado.binário_pontuação

se pontuação == "sim":

impressão("---DECISÃO: DOCUMENTOS RELEVANTES---")

retorno "gerar"

mais:

impressão("---DECISÃO: OS DOCUMENTOS NÃO SÃO RELEVANTES---")

impressão(pontuação)

retorno "rewrite" (reescrever)

### Nó do agente principal

def agente(estado):

"""

Invoca o modelo de agente para gerar uma resposta ou usar ferramentas.

"""

impressão("---CHAME O AGENTE---")

mensagens = estado["mensagens"]

modelo = ChatOpenAI(temperatura=0, transmissão=Verdadeiro, modelo="gpt-4-turbo")

modelo = modelo.bind_tools(ferramentas)

resposta = modelo.invocar(mensagens)

retorno {"mensagens": [resposta]}

### Nó de reescrita de consulta

def reescrever(estado):

"""

Transforme a consulta para produzir uma pergunta melhor.

"""

impressão("---CONSULTA DE TRANSFORMAÇÃO---")

mensagens = estado["mensagens"]

pergunta = mensagens[0].conteúdo

mensagem = [

HumanMessage(

conteúdo=f""" \n

Observe a entrada e tente raciocinar sobre a intenção/significado semântico subjacente. \n

Aqui está a pergunta inicial:

\n ------- \n

{pergunta}

\n ------- \n

Formule uma pergunta aprimorada: """,

)

]

modelo = ChatOpenAI(temperatura=0, modelo="gpt-4-0125-preview", transmissão=Verdadeiro)

resposta = modelo.invocar(mensagem)

retorno {"mensagens": [resposta]}

#### Nó de geração de respostas

def gerar(estado):

"""

Gerar respostas usando os documentos recuperados

"""

impressão("---GERAR---")

mensagens = estado["mensagens"]

pergunta = mensagens[0].conteúdo

last_message = mensagens[-1]

documentos = last_message.conteúdo

# Prompt

imediato = centro.puxar("rlm/rag-prompt")

lm = ChatOpenAI(nome_do_modelo="gpt-4o-mini", temperatura=0, transmissão=Verdadeiro)

corrente de trapos = imediato | lm | StrOutputParser()

# Executar

resposta = corrente de trapos.invocar({"contexto": documentos, "pergunta": pergunta})

retorno {"mensagens": [resposta]}

Criação do gráfico do agente

Agora, conectaremos os nós em um fluxo de trabalho coerente:

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 importação FIM, StateGraph, INICIAR

de langógrafo.pré-construído importação ToolNode

# Definir um novo gráfico

fluxo de trabalho = StateGraph(Estado do agente)

# Definir os nós

fluxo de trabalho.add_node("agente", agente)

recuperar = ToolNode(retriever_tool)

fluxo de trabalho.add_node("recuperar", recuperar)

fluxo de trabalho.add_node("rewrite" (reescrever), reescrever)

fluxo de trabalho.add_node("gerar", gerar)

# Definir bordas

fluxo de trabalho.add_edge(INICIAR, "agente")

# Bordas condicionais baseadas na decisão do agente

fluxo de trabalho.add_conditional_edges(

"agente",

condição_das_ferramentas,

{

"ferramentas": "recuperar",

FIM: FIM,

)

# Bordas condicionais após a recuperação com base na relevância do documento

fluxo de trabalho.add_conditional_edges(

"recuperar",

documentos de notas,

)

fluxo de trabalho.add_edge("gerar", FIM)

fluxo de trabalho.add_edge("rewrite" (reescrever), "agente")

# Compilar o gráfico

gráfico = fluxo de trabalho.compilar()

Visualização do gráfico do agente

Vamos visualizar nosso fluxo de trabalho para entendê-lo melhor:

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.exibição importação Imagem, exibição

tentar:

exibição(Imagem(gráfico.get_graph(raio X=Verdadeiro).draw_mermaid_png()))

exceto Exceção:

# Isso requer algumas dependências extras e é opcional

passe

2) Rastrear as chamadas de função do agente usando o Arize, capturando consultas de recuperação, respostas LLM e uso de ferramentas

O Arize oferece observabilidade abrangente para o nosso sistema de agentes. Vamos configurar o rastreamento:

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 importação registro

# Provedor de rastreamento de configuração

provedor de rastreamento = registro(

space_id = ID DO ESPAÇO,

chave api = API_KEY,

nome_do_projeto = "langgraph-agentic-rag",

)

# Importar o instrumentador automático do OpenInference

de openinference.instrumentação.langchain importação LangChainInstrumentor

# Instrumento LangChain

LangChainInstrumentor().instrumento(provedor de rastreamento=provedor de rastreamento)

Agora vamos executar o agente para ver como ele 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)

importação impressão

insumos = {

"mensagens": [

("usuário", "O que Lilian Weng diz sobre os tipos de ataques adversários aos LLMs?"),

]

}

para saída em gráfico.fluxo(insumos):

para chave, valor em saída.itens():

impressão.impressão(f"Saída do nó '{key}':")

impressão.impressão(valor, recuo=2, largura=80, profundidade=Nenhum)

Isso executará nosso gráfico de agentes e produzirá informações detalhadas de cada nó à medida que ele processa a consulta. No Arize, você poderá ver uma visualização de rastreamento que mostra o fluxo de execução, a latência e os detalhes de cada chamada de função.

Tracing visualization from Arize platform

Visualização de rastreamento da plataforma Arize

3) Avaliar o desempenho gerando um conjunto de dados com consultas e respostas esperadas

Para avaliar sistematicamente nosso sistema, precisamos de um conjunto de dados de referência:

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"])

importação pandas como pd

de langchain_openai importação ChatOpenAI

# Definir um modelo para gerar perguntas

GEN_TEMPLATE = """

Você é um assistente que gera perguntas e respostas sobre o conteúdo abaixo.

As perguntas devem envolver o conteúdo, fatos e números específicos, nomes e elementos da história. Não faça perguntas cuja resposta não esteja no conteúdo.

Responda com uma pergunta por linha. Não inclua nenhuma numeração no início de cada linha. Não inclua nenhum título de categoria.

Gere 10 perguntas. Certifique-se de que não haja perguntas duplicadas.

[INICIAR CONTEÚDO]

{conteúdo}

[END CONTENT]

"""

# Carregue o conteúdo sobre o qual você deseja gerar perguntas

conteúdo = """

Lilian Weng discute vários aspectos de ataques adversários a LLMs e técnicas de engenharia imediata. Certifique-se de usar o nome de Lilian Weng nas perguntas.

"""

# Formate o modelo com o conteúdo

formatted_template = GEN_TEMPLATE.formato(conteúdo=conteúdo)

# Inicializar o modelo de linguagem

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

# Gerar perguntas

resposta = modelo.invocar(formatted_template)

conteúdo_das_perguntas = resposta.conteúdo

perguntas = conteúdo_das_perguntas.faixa().dividir("\n")

# Crie um quadro de dados para armazenar as perguntas

perguntas_df = pd.DataFrame(perguntas, colunas=["input" (entrada)])

4) Avalie o desempenho usando o LLM como juiz

Usaremos a avaliação baseada em LLM para avaliar a qualidade das respostas do nosso 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.provas importação (

RAG_RELEVANCY_PROMPT_RAILS_MAP,

RAG_RELEVANCY_PROMPT_TEMPLATE,

QA_PROMPT_RAILS_MAP,

QA_PROMPT_TEMPLATE,

OpenAIModel,

lm_classificar

)

# Os trilhos são usados para manter a saída em valores específicos com base no modelo

TRILHOS_RELEVANTES = lista(RAG_RELEVANCY_PROMPT_RAILS_MAP.valores())

QA_RAILS = lista(QA_PROMPT_RAILS_MAP.valores())

relevance_eval_df = llm_classificar(

quadro de dados=resposta_df,

modelo=RAG_RELEVANCY_PROMPT_TEMPLATE,

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

trilhos=TRILHOS_RELEVANTES,

fornecer_explicação=Verdadeiro,

include_prompt=Verdadeiro,

concorrência=4,

)

avaliação_de_correção_df = llm_classificar(

quadro de dados=resposta_df,

modelo=QA_PROMPT_TEMPLATE,

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

trilhos=QA_RAILS,

fornecer_explicação=Verdadeiro,

include_prompt=Verdadeiro,

concorrência=4,

)

5) Experimente as configurações de recuperação

Agora vamos experimentar diferentes configurações para otimizar nosso 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,
    )

# Crie um conjunto de dados no Arize para armazenar nossos experimentos

de arize.experimental.conjuntos de dados importação Cliente ArizeDatasets

de uuid importação uuid1

de arize.experimental.conjuntos de dados.experimentos.tipos importação (

ExperimentTaskResultColumnNames,

EvaluationResultColumnNames,

)

de arize.experimental.conjuntos de dados.utilitários.constantes importação GENERATIVO

# Configurar o cliente arize

cliente arize = Cliente ArizeDatasets(developer_key=CHAVE DO DESENVOLVEDOR, chave api=API_KEY)

nome do conjunto de dados = "rag-experiments-" + str(uuid1())[:3]

dataset_id = cliente arize.create_dataset(

space_id=ID DO ESPAÇO,

nome do conjunto de dados=nome do conjunto de dados,

tipo de conjunto de dados=GENERATIVO,

dados=perguntas_df,

)

conjunto de dados = cliente arize.get_dataset(space_id=ID DO ESPAÇO, dataset_id=dataset_id)

# Definir mapeamentos de colunas para resultados de tarefas e avaliações

task_cols = ExperimentTaskResultColumnNames(

exemplo_id="example_id", resultado="output"

)

colunas_avaliadoras_de_relevância = EvaluationResultColumnNames(

rótulo="relevância",

explicação="relevance_explanation" (explicação da relevância),

)

colunas_avaliadoras_de_correção = EvaluationResultColumnNames(

rótulo="correção",

explicação="correctness_explanation" (explicação da correção),

)

# Função para registrar experimentos no Arize

def registro de experimentos a serem realizados(experimento_df, nome_do_experimento):

experimento_df["example_id"] = conjunto de dados["id"]

retorno cliente arize.experimento de registro(

space_id=ID DO ESPAÇO,

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

experimento_df=experimento_df,

task_columns=task_cols,

colunas_avaliador={

"correção": colunas_avaliadoras_de_correção,

"relevância": colunas_avaliadoras_de_relevância,

nome do conjunto de dados=nome do conjunto de dados,

)

Agora, faremos experimentos com diferentes configurações:

# 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: blocos de 1024 tokens, k=2

reset_vector_store(vector_store, tamanho do bloco=1024, chunk_overlap=20)

k_2_chunk_1024_overlap_20 = run_rag(perguntas_df, Valor k=2)

k_2_chunk_1024_overlap_20 = run_evaluators(k_2_chunk_1024_overlap_20)

# Experimento 2: blocos de 1024 tokens, k=4

k_4_chunk_1024_overlap_20 = run_rag(perguntas_df, Valor k=4)

k_4_chunk_1024_overlap_20 = run_evaluators(k_4_chunk_1024_overlap_20)

# Experimento 3: blocos menores (200 tokens), k=2

reset_vector_store(vector_store, tamanho do bloco=200, chunk_overlap=20)

k_2_chunk_200_overlap_20 = run_rag(perguntas_df, Valor k=2)

k_2_chunk_200_overlap_20 = run_evaluators(k_2_chunk_200_overlap_20)

# Experimento 4: blocos médios (500 tokens), k=2

reset_vector_store(vector_store, tamanho do bloco=500, chunk_overlap=20)

k_2_chunk_500_overlap_20 = run_rag(perguntas_df, Valor k=2)

k_2_chunk_500_overlap_20 = run_evaluators(k_2_chunk_500_overlap_20)

# Registre todos os experimentos no Arize

registro de experimentos a serem realizados(k_2_chunk_1024_overlap_20, "k_2_chunk_1024_overlap_20")

registro de experimentos a serem realizados(k_4_chunk_1024_overlap_20, "k_4_chunk_1024_overlap_20")

registro de experimentos a serem realizados(k_2_chunk_200_overlap_20, "k_2_chunk_200_overlap_20")

registro de experimentos a serem realizados(k_2_chunk_500_overlap_20, "k_2_chunk_500_overlap_20")

6) Comparar experimentos no Arize

Depois de executar todos os experimentos, agora você pode visualizá-los e compará-los na interface do usuário do Arize. Os experimentos devem estar visíveis no espaço de trabalho do Arize sob o nome do conjunto de dados que criamos anteriormente.

Visualização de comparação de experimentos da plataforma Arize

No Arize, você pode:

1. Comparar as métricas de desempenho geral entre diferentes configurações
2. Analisar o desempenho por pergunta para identificar padrões
3. Examinar os detalhes do rastreamento para entender o fluxo de execução
4. Veja as pontuações de relevância e correção de cada experimento
5. Veja as explicações para as decisões de avaliação
6. Avalie os resultados usando um LLM como juiz para pontuar a relevância e a correção das respostas.
7. Otimize as configurações de recuperação experimentando tamanhos de blocos, configurações de sobreposição e limites de recuperação de documentos.
8. Compare e analise experimentos no Arize para determinar as configurações de melhor desempenho.

Inovar com o Couchbase e a Arize AI

A integração do Couchbase e do Arize permite que as empresas criem aplicativos de GenAI robustos e prontos para a produção, com fortes recursos de observabilidade e otimização. Ao aproveitar o Agentic RAG com decisões de recuperação monitoradas, as organizações podem aumentar a precisão, reduzir as alucinações e garantir o desempenho ideal ao longo do tempo.

À medida que as empresas continuam a ultrapassar os limites da GenAI, a combinação do armazenamento vetorial de alto desempenho com a observabilidade da IA será fundamental para a implementação de aplicativos confiáveis e dimensionáveis. Com o Couchbase e o Arize, as organizações têm as ferramentas para enfrentar com confiança os desafios da implementação da GenAI corporativa.

Recursos adicionais

- Comece a usar Plataforma de desenvolvedor do Couchbase Capella gratuitamente
- Entre em contato com a Arize AI para uma demonstração aqui
- Leia mais postagens e tutoriais em IA generativa (GenAI)

Compartilhe este artigo

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

Couchbase faz parceria com Arize AI para permitir aplicativos de agentes de IA confiáveis e prontos para produção

O valor da parceria entre a Couchbase e a Arize AI

A importância da observabilidade do LLM

Criação de um chatbot de controle de qualidade RAG autêntico

Implementação passo a passo

Pré-requisitos:

1) Criar um chatbot Agentic RAG usando LangGraph, Couchbase como armazenamento de vetores e Agent Catalog para gerenciar agentes de IA

Configuração de dependências

Conexão com o Couchbase

Ingestão de documentos

Configuração da ferramenta Retriever

Definição do estado do agente

Criação de nós de agente

Criação do gráfico do agente

Visualização do gráfico do agente

2) Rastrear as chamadas de função do agente usando o Arize, capturando consultas de recuperação, respostas LLM e uso de ferramentas

3) Avaliar o desempenho gerando um conjunto de dados com consultas e respostas esperadas

4) Avalie o desempenho usando o LLM como juiz

5) Experimente as configurações de recuperação

6) Comparar experimentos no Arize

Inovar com o Couchbase e a Arize AI

Recursos adicionais

Receba atualizações do blog do Couchbase em sua caixa de entrada

Autor

Postado por Richard Young - Dir. Arquitetura de soluções para parceiros, Arize AI

Deixe um comentário Cancelar resposta

Pronto para começar a usar o Couchbase Capella?

Iniciar a construção

Use o Capella gratuitamente

Entre em contato