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

기업이 프로덕션 지원 AI 에이전트 애플리케이션을 배포하고자 할 때, 대규모 언어 모델(LLM) 통합 가시성은 성능과 신뢰를 모두 보장하기 위한 중요한 요구 사항으로 부상했습니다. 조직은 신뢰성, 보안 및 규정 준수를 유지하기 위해 에이전트가 데이터와 상호 작용하고, 의사 결정을 내리고, 정보를 검색하는 방식에 대한 가시성이 필요합니다. 적절한 가시성이 없으면 기업은 일관되지 않거나 부정확하거나 편향된 결과를 생성하는 모델을 배포하여 사용자 경험이 저하되고 운영 비효율을 초래할 위험이 있습니다. 새로운 파트너십 카우치베이스 그리고 아리제 AI 는 AI 기반 애플리케이션에 강력한 모니터링, 평가, 최적화 기능을 제공하는 데 중요한 역할을 합니다.

Couchbase와 Arize AI의 통합은 검색 증강 세대(RAG) 및 에이전트 애플리케이션을 대규모로 구축 및 모니터링할 수 있는 강력한 솔루션을 제공합니다. Couchbase의 고성능 벡터 데이터베이스와 Arize AI 통합 가시성 플랫폼 및 향상된 모니터링 기능을 활용하여 기업은 프로덕션 환경에서 에이전트 RAG 솔루션을 자신 있게 구축, 배포 및 최적화할 수 있습니다.

이 블로그에서는 LangGraph와 Couchbase를 사용하여 에이전틱 RAG QA 챗봇을 만드는 과정을 안내합니다. 상담원 카탈로그 최근 발표된 카펠라 AI 서비스 (미리 보기), 그리고 Arize AI로 성능을 평가하고 최적화하고 있습니다. 이는 개발자가 Couchbase와 Arize AI를 통해 검색 워크플로우를 개선하고, 응답 정확도를 높이고, LLM 기반 상호 작용을 실시간으로 모니터링하는 방법을 보여주는 실질적인 예시입니다.

카우치베이스와 아리제 AI 파트너십의 가치

카우치베이스와 아리제 AI는 힘을 합쳐 개발자가 AI 에이전트 애플리케이션을 구축하고 평가하는 방식을 혁신하고 있습니다. 개발자는 에이전트 워크플로우 조정을 위해 선호하는 에이전트 개발 프레임워크와 함께 LLM 캐싱, 장기 및 단기 에이전트 메모리, 벡터 임베딩 사용 사례, 분석 및 운영 워크로드를 위한 단일 데이터 플랫폼으로 Couchbase Capella를 활용하여 정교한 에이전트 애플리케이션을 구축할 수 있습니다.

카우치베이스 에이전트 카탈로그는 조직 내에서 다양한 에이전트 도구의 저장, 관리 및 검색, 신속한 버전 관리, LLM 추적 디버깅이 가능한 멀티 에이전트 워크플로우를 위한 중앙 집중식 저장소를 제공하여 이 시스템을 더욱 향상시킵니다.

높은 신뢰성과 투명성을 보장하기 위해 Arize AI는 다음과 같은 중요한 가시성 기능을 제공합니다:

- 에이전트 함수 호출 추적: Arize를 사용하면 검색 단계 및 LLM 상호 작용을 포함한 상담원의 함수 호출을 자세히 모니터링하여 응답이 어떻게 생성되는지 추적할 수 있습니다.
- 데이터 세트 벤치마킹: 개발자는 구조화된 데이터 세트를 만들어 시간 경과에 따른 상담원 성과를 평가하고 비교할 수 있습니다.
- LLM을 심사위원으로 하는 성과 평가: 아리제는 기본 제공 평가기를 사용하여 LLM을 활용하여 응답 정확도, 관련성 및 전반적인 상담원 효과를 평가합니다.
- 검색 전략 실험하기: 청크 크기, 중복 및 검색된 문서 수(K-값)를 조정하여 개발자가 상담원 성과에 미치는 영향을 분석할 수 있습니다.
- 아리제 비교 분석: 이 플랫폼에서는 다양한 검색 전략을 나란히 비교할 수 있으므로 팀이 상담원을 위한 최적의 구성을 결정할 수 있습니다.

LLM 관찰 가능성의 중요성

AI 애플리케이션이 프로덕션 환경에서 제대로 작동하도록 하려면 기업에는 강력한 평가 프레임워크가 필요합니다. 개발자는 Arize AI와 같은 통합 가시성 도구를 통해 이를 수행할 수 있습니다:

- 관련성, 환각률, 지연 시간 등의 요소를 기준으로 LLM 결과물을 평가합니다.
- 신속한 변경, 검색 수정 및 매개변수 조정의 영향을 측정하기 위해 체계적인 평가를 수행합니다.
- 포괄적인 데이터 집합을 큐레이션하여 다양한 사용 사례에서 성능을 벤치마킹하세요.
- CI/CD 파이프라인 내 평가 프로세스를 자동화하여 일관된 애플리케이션 안정성 보장

개발자는 LLM을 판단 기준으로 삼아 사전 테스트를 거친 평가자, 다단계 사용자 지정 평가 기법, 대규모 성능 벤치마킹을 사용하여 에이전트의 효율성을 측정할 수 있습니다. 수천 번의 평가를 실행함으로써 팀은 빠르게 반복하고 LLM 프롬프트, 검색 방법 및 상담원 워크플로를 개선하여 전반적인 애플리케이션 품질을 개선할 수 있습니다.

에이전트 RAG QA 챗봇 구축하기

에이전틱 RAG는 기존의 검색 증강 생성 기능과 지능형 의사 결정을 결합합니다. 이 구현에서는 LLM이 쿼리 컨텍스트에 따라 검색이 필요한지 여부를 동적으로 결정할 수 있도록 합니다.

의 상담원 워크플로우를 설명하는 그림 Langgraph의 에이전트 RAG 예시.

단계별 구현

이 블로그의 나머지 내용은 다음과 같은 내용을 기반으로 합니다. 튜토리얼 노트. 관찰 가능한 AI 에이전트를 빌드하고 배포하기 전에 개발 환경을 구성해야 합니다.

전제 조건:

1. 이 튜토리얼을 따라 하려면 다음이 필요합니다. 아리제 가입하기 를 클릭하고 Space, API 및 개발자 키를 받으세요. 스페이스에서 여기 가이드. 또한 OpenAI API 키.
2. 다음을 수행하여 Couchbase 클러스터를 설정해야 합니다:
  1. 다음에서 계정을 만드세요. 카우치베이스 클라우드
  2. 데이터, 인덱스, 검색 서비스가 활성화된 무료 클러스터를 생성하세요*.
  3. 클러스터 액세스 자격 증명 만들기
  4. 로컬 머신에서 클러스터에 대한 액세스 허용
  5. 문서를 저장할 버킷 만들기
  6. 만들기 검색 색인
3. Couchbase 상담원 카탈로그를 사용하여 상담원에게 필요한 도구 및 프롬프트를 만듭니다(설치 및 자세한 지침은 다음 문서를 참조하세요. 여기)

*검색 서비스는 나중에 상담원 카탈로그를 사용할 때 시맨틱 검색을 수행하는 데 사용됩니다.

1) AI 에이전트를 관리하기 위한 벡터 저장소 및 에이전트 카탈로그로 LangGraph, Couchbase를 사용하여 에이전트 RAG 챗봇을 생성합니다.

종속성 설정

%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 설치 -qU 랭체인-openai 랭체인-커뮤니티 랭체인 랭그래프 랭그래프.사전 구축 openai 랭체인-카우치베이스 agentc 랭체인-포옹하는 얼굴 랭체인_핵심

%pip 설치 -qq "arize-phoenix[evals]>=7.0.0" "arize-otel>=0.7.0" "openinference-instrumentation-openai>=0.1.18" "오픈인퍼런스-인스트루먼트-랑체인>=0.1.29"

Couchbase에 연결

벡터 저장소로 Couchbase를 사용하겠습니다. 연결을 설정하는 방법은 다음과 같습니다:

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,
)

에서 날짜 시간 가져오기 timedelta

에서 카우치베이스.auth 가져오기 비밀번호 인증기

에서 카우치베이스.클러스터 가져오기 클러스터

에서 카우치베이스.옵션 가져오기 클러스터 옵션

에서 랭체인_카우치베이스.벡터 스토어 가져오기 카우치베이스검색벡터스토어

에서 랭체인_허깅페이스 가져오기 포옹 얼굴 임베딩

# 클러스터 설정

CB_CONN_STRING = "your-connection-string"

CB_USERNAME = "your-username"

CB_PASSWORD = "your-password"

BUCKET_NAME = "귀하의 버킷 이름"

SCOPE_NAME = "사용자 범위 이름"

컬렉션_이름 = "귀하의 컬렉션 이름"

검색_인덱스_이름 = "사용자 검색 색인 이름"

# 카우치베이스 클러스터에 연결

auth = 비밀번호 인증기(CB_USERNAME, CB_PASSWORD)

옵션 = 클러스터 옵션(auth)

옵션.적용_프로필("wan_development")

클러스터 = 클러스터(CB_CONN_STRING, 옵션)

클러스터.wait_until_ready(timedelta(초=5))

# 벡터 저장소 초기화

임베딩 = 포옹 얼굴 임베딩(모델_이름="문장-변압기/모든-MiniLM-L12-v2")

벡터_스토어 = 카우치베이스 벡터 스토어(

클러스터=클러스터,

버킷_이름=BUCKET_NAME,

범위_이름=SCOPE_NAME,

컬렉션_이름=컬렉션_이름,

임베딩=임베딩,

index_name=검색_인덱스_이름,

)

문서 수집

구성 가능한 청킹 매개변수를 사용하여 문서를 로드하고 색인하는 도우미 함수를 만들겠습니다:

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)

에서 랭체인_커뮤니티.문서 로더 가져오기 웹베이스 로더

에서 랭체인_텍스트_스플리터 가져오기 리커시브 문자 텍스트 스플리터

def RESET_VECTOR_STORE(벡터_스토어, 청크 크기=1024, 청크_오버랩=20):

시도:

# 기존 문서 삭제

결과 = 벡터_스토어.유사성 검색(

k=1000,

쿼리="",

검색_옵션={

"query": {"field": "메타데이터 소스", "match": "lilian_weng_blog"}

)

만약 결과:

DELETED_IDS = [결과.id 에 대한 결과 in 결과]

벡터_스토어.삭제(ids=DELETED_IDS)

# URL에서 문서 로드

URL = [

"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/",

]

문서 = [웹베이스 로더(URL).load() 에 대한 URL in URL]

문서 목록 = [항목 에 대한 하위 목록 in 문서 에 대한 항목 in 하위 목록]

# 구성 가능한 매개변수와 함께 RecursiveCharacterTextSplitter 사용

text_splitter = 리커시브 문자 텍스트 스플리터(

청크 크기=청크 크기,

청크_오버랩=청크_오버랩,

구분 기호=["\n\n", "\n", " ", ""],

)

doc_splits = text_splitter.분할_문서(문서 목록)

# 문서에 메타데이터 추가하기

에 대한 doc in doc_splits:

doc.메타데이터["source"] = "lilian_weng_blog"

# 벡터 스토어에 문서 추가

벡터_스토어.추가_문서(doc_splits)

반환 벡터_store

예외 ValueError as e:

인쇄(f"오류: {e}")

# 기본 설정으로 초기화하기

RESET_VECTOR_STORE(벡터_스토어)

리트리버 도구 설정

에이전트 카탈로그에서 리트리버 도구를 가져옵니다. agentc 제공자. 향후 더 많은 도구(및/또는 프롬프트)가 필요하고 애플리케이션이 더 복잡해지면 상담원 카탈로그 SDK 및 CLI를 사용하여 사용 사례(시맨틱 검색) 또는 이름에 따라 도구를 자동으로 가져올 수 있습니다.

이 도구가 만들어진 방법과 상담원 카탈로그의 더 많은 기능에 대한 안내는 다음 문서를 참조하세요. 여기.

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

가져오기 agentc.랭체인

가져오기 agentc

에서 랭체인_코어.도구 가져오기 도구

공급자 = agentc.공급자(

데코레이터=람다 t: 도구(t.func),

비밀={"CB_USERNAME": CB_USERNAME,

"CB_PASSWORD": CB_PASSWORD,

"cb_conn_string": CB_CONN_STRING})

# 상담원 카탈로그에서 리트리버 도구 받기

retriever_tool = 공급자.get_item(이름="retriever_tool", item_type="tool")

도구 = retriever_tool

에이전트 상태 정의하기

관련된 모든 상담원이 서로 더 잘 소통할 수 있도록 상담원 그래프를 정의합니다. 상담원은 상태 객체를 각 노드에 전달하고 해당 노드의 출력으로 수정합니다.

우리 주는 다음과 같은 목록이 될 것입니다. 메시지 를 클릭하면 그래프의 각 노드가 추가됩니다:

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]

에서 입력 가져오기 주석, 시퀀스, TypedDict

에서 랭체인_코어.메시지 가져오기 BaseMessage

에서 랭그래프.그래프.메시지 가져오기 add_messages

클래스 에이전트 상태(TypedDict):

# add_messages 함수는 업데이트 처리 방법을 정의합니다.

# 기본값은 바꾸기입니다. add_messages는 "추가"라고 합니다.

메시지: 주석[시퀀스[BaseMessage], add_messages]

에이전트 노드 만들기

에이전트 파이프라인의 핵심 구성 요소를 정의하겠습니다:

노드: 관련성 검사 기능, 쿼리 재작성기, 메인 에이전트, 응답 생성

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]}

에서 입력 가져오기 주석, 리터럴, 시퀀스, TypedDict

에서 랭체인 가져오기 허브

에서 랭체인_코어.메시지 가져오기 BaseMessage, 휴먼 메시지

에서 랭체인_코어.출력_파서 가져오기 StrOutputParser

에서 랭체인_코어.프롬프트 가져오기 프롬프트 템플릿

에서 랭체인_코어.pydantic_v1 가져오기 기본 모델, 필드

에서 랭체인_오픈아이 가져오기 ChatOpenAI

에서 랭그래프.사전 구축 가져오기 도구_조건

### 관련성 확인 기능

def grade_documents(상태) -> 리터럴["생성", "재작성"]:

"""

검색된 문서가 질문과 관련이 있는지 여부를 결정합니다.

"""

인쇄("---관련성 확인---")

# 데이터 모델

클래스 학년(기본 모델):

"""관련성 확인을 위한 이진 점수"""

binary_score: str = 필드(설명="관련성 점수 '예' 또는 '아니오'")

# LLM

모델 = ChatOpenAI(온도=0, 모델="gpt-4o", 스트리밍=True)

llm_with_tool = 모델.WITH_STRUCTURED_OUTPUT(학년)

1TP5상담원 카탈로그에서 프롬프트 가져오기

학년_문서_프롬프트 = 프롬프트 템플릿(

템플릿=공급자.get_item(이름="grade_documents", item_type="prompt").프롬프트.렌더링(),

입력_변수=["context", "question"],

)

# 체인

체인 = 학년_문서_프롬프트 | llm_with_도구

메시지 = 상태["메시지"]

last_message = 메시지[-1]

질문 = 메시지[0].콘텐츠

문서 = last_message.콘텐츠

점수_결과 = 체인.호출({"question": 질문, "context": 문서})

점수 = 점수_결과.바이너리_점수

만약 점수 == "yes":

인쇄("---결정: 관련 문서---")

반환 "생성"

else:

인쇄("---결정: 관련 없는 문서---")

인쇄(점수)

반환 "재작성"

### 메인 에이전트 노드

def 에이전트(상태):

"""

응답을 생성하거나 툴을 사용하기 위해 상담원 모델을 호출합니다.

"""

인쇄("---전화 상담원---")

메시지 = 상태["메시지"]

모델 = ChatOpenAI(온도=0, 스트리밍=True, 모델="GPT-4-TURBO")

모델 = 모델.bind_tools(도구)

응답 = 모델.호출(메시지)

반환 {"메시지": [응답]}

### 쿼리 재작성 노드

def 다시 쓰기(상태):

"""

쿼리를 변형하여 더 나은 질문을 만들어 보세요.

"""

인쇄("---변환 쿼리---")

메시지 = 상태["메시지"]

질문 = 메시지[0].콘텐츠

msg = [

휴먼 메시지(

콘텐츠=f""" \n

입력을 보고 기본 의미론적 의도/의미에 대해 추론해 보세요. \n

첫 번째 질문은 다음과 같습니다:

------- \n

{질문}

------- \n

개선된 질문을 만들어 보세요: """,

)

]

모델 = ChatOpenAI(온도=0, 모델="GPT-4-0125-프리뷰", 스트리밍=True)

응답 = 모델.호출(msg)

반환 {"메시지": [응답]}

### 응답 생성 노드

def 생성(상태):

"""

검색된 문서를 사용하여 답변 생성

"""

인쇄("---생성---")

메시지 = 상태["메시지"]

질문 = 메시지[0].콘텐츠

last_message = 메시지[-1]

문서 = last_message.콘텐츠

# 프롬프트

프롬프트 = 허브.pull("rlm/rag-prompt")

llm = ChatOpenAI(모델_이름="GPT-4O-MINI", 온도=0, 스트리밍=True)

rag_chain = 프롬프트 | llm | StrOutputParser()

# 실행

응답 = rag_chain.호출({"context": 문서, "question": 질문})

반환 {"메시지": [응답]}

상담원 그래프 구축하기

이제 노드를 일관된 워크플로에 연결해 보겠습니다:

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()

에서 랭그래프.그래프 가져오기 END, StateGraph, 시작

에서 랭그래프.사전 구축 가져오기 툴노드

# 새 그래프 정의하기

워크플로 = StateGraph(에이전트 상태)

# 노드 정의

워크플로.add_node("에이전트", 에이전트)

검색 = 툴노드(retriever_tool)

워크플로.add_node("retrieve", 검색)

워크플로.add_node("재작성", 다시 쓰기)

워크플로.add_node("생성", 생성)

# 가장자리 정의

워크플로.add_edge(시작, "에이전트")

# 상담원의 결정에 따른 조건부 에지

워크플로.추가_조건부_엣지(

"에이전트",

도구_조건,

{

"도구": "retrieve",

END: END,

)

문서 관련성에 기반한 검색 후 조건부 에지 #

워크플로.추가_조건부_엣지(

"retrieve",

grade_documents,

)

워크플로.add_edge("생성", END)

워크플로.add_edge("재작성", "에이전트")

# 그래프 컴파일

그래프 = 워크플로.컴파일()

에이전트 그래프 시각화하기

워크플로우를 더 잘 이해하기 위해 시각화해 보겠습니다:

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

에서 IPython.디스플레이 가져오기 이미지, 디스플레이

시도:

디스플레이(이미지(그래프.get_graph(xray=True).draw_mermaid_png()))

예외 예외:

# 이것은 몇 가지 추가 종속성이 필요하며 선택 사항입니다.

통과

2) Arize를 사용하여 에이전트의 함수 호출을 추적하여 검색 쿼리, LLM 응답 및 도구 사용을 캡처합니다.

Arize는 에이전트 시스템을 위한 포괄적인 관찰 기능을 제공합니다. 추적을 설정해 보겠습니다:

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)

에서 아리제.otel 가져오기 등록

# 트레이서 공급자 설정

추적자 공급자 = 등록(

space_id = SPACE_ID,

api_key = API_KEY,

프로젝트_이름 = "랑그라프-에이전트-래그",

)

# OpenInference에서 자동 계측기 가져오기

에서 오픈 추론.계측.랭체인 가져오기 랭체인 인스트루멘터

# 인스트루먼트 랭체인

랭체인 인스트루멘터().악기(추적자 공급자=추적자 공급자)

이제 에이전트를 실행하여 어떻게 작동하는지 확인해 보겠습니다:

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)

가져오기 pprint

입력 = {

"메시지": [

("user", "릴리안 웡은 LLM에 대한 적대적 공격의 유형에 대해 어떻게 말하나요?"),

]

}

에 대한 출력 in 그래프.스트림(입력):

에 대한 키, 값 in 출력.항목():

pprint.pprint(f"노드 '{키}'의 출력:")

pprint.pprint(값, 들여쓰기=2, 너비=80, 깊이=없음)

그러면 에이전트 그래프가 실행되고 쿼리를 처리하면서 각 노드에 대한 자세한 정보가 출력됩니다. 아리제에서는 각 함수 호출의 실행 흐름, 지연 시간 및 세부 정보를 보여주는 추적 시각화를 볼 수 있습니다.

Tracing visualization from Arize platform

아리제 플랫폼의 추적 시각화

3) 쿼리와 예상 응답이 포함된 데이터 세트를 생성하여 성능 벤치마크하기

시스템을 체계적으로 평가하려면 벤치마크 데이터 세트가 필요합니다:

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

가져오기 판다 as pd

에서 랭체인_오픈아이 가져오기 ChatOpenAI

# 질문 생성을 위한 템플릿 정의하기

GEN_TEMPLATE = """

귀하는 아래 콘텐츠에 대한 Q&A 질문을 생성하는 어시스턴트입니다.

질문에는 스토리의 내용, 구체적인 사실과 수치, 이름 및 요소가 포함되어야 합니다. 콘텐츠에 답이 없는 질문은 하지 마세요.

한 줄에 한 개의 질문으로 응답합니다. 각 줄의 시작 부분에 번호 매기기를 포함하지 마세요. 카테고리 제목을 포함하지 마세요.

10개의 질문을 생성합니다. 중복되는 질문이 없는지 확인하세요.

[콘텐츠 시작]

{콘텐츠}

[콘텐츠 끝]

"""

# 질문을 생성하려는 콘텐츠를 로드합니다.

콘텐츠 = """

릴리안 웡이 LLM에 대한 적대적 공격의 다양한 측면과 프롬프트 엔지니어링 기법에 대해 설명합니다. 질문에 릴리안 웡의 이름을 반드시 사용하세요.

"""

# 콘텐츠로 템플릿 서식 지정하기

형식화된_템플릿 = GEN_TEMPLATE.형식(콘텐츠=콘텐츠)

# 언어 모델 초기화하기

모델 = ChatOpenAI(모델="gpt-4o", 최대_토큰=1300)

# 질문 생성

응답 = 모델.호출(형식화된_템플릿)

질문_내용 = 응답.콘텐츠

질문 = 질문_내용.strip().분할("\n")

# 질문을 저장할 데이터 프레임 만들기

questions_df = pd.데이터프레임(질문, 열=["입력"])

4) LLM을 심사위원으로 사용하여 성과 평가하기

LLM 기반 평가를 사용하여 상담원의 응답 품질을 평가합니다:

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,
)

에서 phoenix.평가 가져오기 (

래그 관련성_프롬프트_레일 맵,

래그 관련성_프롬프트 템플릿,

QA_PROMPT_RAILS_MAP,

QA_PROMPT_템플릿,

OpenAIModel,

llm_분류

)

# 레일은 템플릿에 따라 출력을 특정 값으로 고정하는 데 사용됩니다.

관련성_레일 = 목록(래그 관련성_프롬프트_레일 맵.값())

QA_RAILS = 목록(QA_PROMPT_RAILS_MAP.값())

5) 검색 설정으로 실험하기

이제 시스템을 최적화하기 위해 다양한 구성을 실험해 보겠습니다:

# 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,
    )

# 아리제에서 데이터 세트를 생성하여 실험을 저장합니다.

에서 아리제.실험적.데이터 세트 가져오기 아리제데이터세트클라이언트

에서 uuid 가져오기 uuid1

에서 아리제.실험적.데이터 세트.실험.유형 가져오기 (

실험 작업 결과 열 이름,

평가 결과 열 이름,

)

에서 아리제.실험적.데이터 세트.유틸리티.상수 가져오기 제너레이티브

# arize 클라이언트 설정하기

arize_client = 아리제데이터세트클라이언트(개발자 키=개발자 키, api_key=API_KEY)

데이터세트_이름 = "헝겊-실험-" + str(uuid1())[:3]

데이터세트_ID = arize_client.create_dataset(

space_id=SPACE_ID,

데이터세트_이름=데이터세트_이름,

데이터세트_유형=제너레이티브,

데이터=questions_df,

)

데이터 세트 = arize_client.get_dataset(space_id=SPACE_ID, 데이터세트_ID=데이터세트_ID)

# 작업 및 평가 결과에 대한 열 매핑 정의하기

task_cols = 실험 작업 결과 열 이름(

example_id="example_id", 결과="output"

)

관련성_평가자_콜스 = 평가 결과 열 이름(

레이블="관련성",

설명="관련성_설명",

)

정확성_평가자_콜 = 평가 결과 열 이름(

레이블="정확성",

설명="정확성_설명",

)

# 아리즈에 실험을 기록하는 기능

def 로그_실험_투_아라이즈(실험_df, 실험_이름):

실험_df["example_id"] = 데이터 세트["id"]

반환 arize_client.로그_실험(

space_id=SPACE_ID,

실험_이름=실험_이름 + "-" + str(uuid1())[:2],

실험_df=실험_df,

작업_열=task_cols,

평가자_컬럼={

"정확성": 정확성_평가자_콜,

"관련성": 관련성_평가자_콜스,

데이터세트_이름=데이터세트_이름,

)

이제 다양한 구성으로 실험을 실행해 보겠습니다:

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

# 실험 1: 1024개 토큰 청크, k=2

RESET_VECTOR_STORE(벡터_스토어, 청크 크기=1024, 청크_오버랩=20)

k_2_chunk_1024_overlap_20 = run_rag(questions_df, k_value=2)

k_2_chunk_1024_overlap_20 = 실행_평가자(k_2_chunk_1024_overlap_20)

# 실험 2: 1024개 토큰 청크, k=4

k_4_chunk_1024_overlap_20 = run_rag(questions_df, k_value=4)

k_4_chunk_1024_overlap_20 = 실행_평가자(k_4_chunk_1024_overlap_20)

# 실험 3: 더 작은 청크(200토큰), k=2

RESET_VECTOR_STORE(벡터_스토어, 청크 크기=200, 청크_오버랩=20)

k_2_chunk_200_overlap_20 = run_rag(questions_df, k_value=2)

k_2_chunk_200_overlap_20 = 실행_평가자(k_2_chunk_200_overlap_20)

# 실험 4: 중간 청크(500토큰), k=2

RESET_VECTOR_STORE(벡터_스토어, 청크 크기=500, 청크_오버랩=20)

k_2_chunk_500_overlap_20 = run_rag(questions_df, k_value=2)

k_2_chunk_500_overlap_20 = 실행_평가자(k_2_chunk_500_overlap_20)

# 모든 실험을 Arize에 기록하세요.

로그_실험_투_아라이즈(k_2_chunk_1024_overlap_20, "k_2_chunk_1024_overlap_20")

로그_실험_투_아라이즈(k_4_chunk_1024_overlap_20, "k_4_chunk_1024_overlap_20")

로그_실험_투_아라이즈(k_2_chunk_200_overlap_20, "k_2_chunk_200_overlap_20")

로그_실험_투_아라이즈(k_2_chunk_500_overlap_20, "k_2_chunk_500_overlap_20")

6) 아리제에서 실험 비교

모든 실험을 실행한 후에는 이제 Arize UI에서 실험을 보고 비교할 수 있습니다. 실험은 앞서 만든 데이터 세트 이름으로 Arize 작업 영역에서 볼 수 있습니다.

아리제 플랫폼의 실험 비교 보기

아리제에서는 가능합니다:

1. 다양한 구성 간의 전체 성능 메트릭 비교
2. 질문별 성과를 분석하여 패턴 파악하기
3. 추적 세부 정보를 검토하여 실행 흐름 이해
4. 각 실험의 관련성 및 정확성 점수 보기
5. 평가 결정에 대한 설명 참조
6. 응답 관련성 및 정확성에 점수를 매기기 위해 LLM을 판정자로 사용하여 출력을 평가합니다.
7. 청크 크기, 중첩 구성, 문서 검색 제한을 실험하여 검색 설정을 최적화하세요.
8. 아리제에서 실험을 비교하고 분석하여 가장 성능이 좋은 구성을 결정하세요.

카우치베이스와 아리제 AI로 혁신하기

Couchbase와 Arize의 통합으로 기업은 강력한 통합 가시성 및 최적화 기능을 갖춘 강력한 프로덕션 지원 GenAI 애플리케이션을 구축할 수 있습니다. 조직은 검색 결정 모니터링과 함께 에이전틱 RAG를 활용함으로써 정확도를 개선하고, 착각을 줄이며, 시간이 지남에 따라 최적의 성능을 보장할 수 있습니다.

기업이 계속해서 GenAI의 한계를 확장함에 따라, 고성능 벡터 스토리지와 AI 통합 가시성을 결합하는 것이 안정적이고 확장 가능한 애플리케이션을 배포하는 데 핵심이 될 것입니다. Couchbase와 Arize를 통해 조직은 엔터프라이즈 GenAI 배포의 과제를 자신 있게 해결할 수 있는 도구를 갖추게 됩니다.

추가 리소스

- 시작하기 카우치베이스 카펠라 개발자 플랫폼 무료
- 아리제 AI에 데모 요청하기 여기
- 다음에 대한 더 많은 게시물과 자습서를 읽어보세요. 생성 AI(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

모든 게시물

카우치베이스, 아리제 AI와 제휴하여 신뢰할 수 있고 생산 준비가 완료된 AI 에이전트 애플리케이션 지원

카우치베이스와 아리제 AI 파트너십의 가치

LLM 관찰 가능성의 중요성

에이전트 RAG QA 챗봇 구축하기

단계별 구현

전제 조건:

1) AI 에이전트를 관리하기 위한 벡터 저장소 및 에이전트 카탈로그로 LangGraph, Couchbase를 사용하여 에이전트 RAG 챗봇을 생성합니다.

종속성 설정

Couchbase에 연결

문서 수집

리트리버 도구 설정

에이전트 상태 정의하기

에이전트 노드 만들기

상담원 그래프 구축하기

에이전트 그래프 시각화하기

2) Arize를 사용하여 에이전트의 함수 호출을 추적하여 검색 쿼리, LLM 응답 및 도구 사용을 캡처합니다.

3) 쿼리와 예상 응답이 포함된 데이터 세트를 생성하여 성능 벤치마크하기

4) LLM을 심사위원으로 사용하여 성과 평가하기

5) 검색 설정으로 실험하기

6) 아리제에서 실험 비교

카우치베이스와 아리제 AI로 혁신하기

추가 리소스

작성자

게시자 Richard Young - Dir. 파트너 솔루션 아키텍처, 아리제 AI

댓글 남기기 응답 취소