다음을 통해 공유

Azure Database for PostgreSQL에서 LangChain 사용

Azure Database for PostgreSQL은 LangChain과 같은 주요 LLM(대규모 언어 모델) 오케스트레이션 패키지와 원활하게 통합되어 개발자가 애플리케이션 내에서 고급 AI 기능의 기능을 활용할 수 있도록 합니다. LangChain은 LLM, 포함 모델 및 데이터베이스의 관리 및 사용을 간소화하여 생성 AI 애플리케이션을 더욱 쉽게 개발할 수 있습니다.

이 자습서에서는 Azure Database for PostgreSQL 통합 벡터 데이터베이스 를 사용하여 LangChain을 사용하여 컬렉션에 문서를 저장하고 관리하는 방법을 보여 줍니다. 또한 Cosine Distance, L2(유클리드 거리) 및 IP(내부 제품)와 같은 근사한 인접 알고리즘을 사용하여 인덱스를 만들고 벡터 검색 쿼리를 수행하여 쿼리 벡터와 가까운 문서를 찾는 방법을 보여 줍니다.

벡터 지원

Azure Database for PostgreSQL - 유연한 서버를 사용하면 수백만 개의 벡터 포함을 PostgreSQL에 효율적으로 저장 및 쿼리하고 POC(개념 증명)에서 프로덕션으로 AI 사용 사례를 확장할 수 있습니다.

  • 벡터 포함 및 관계형 데이터를 쿼리하기 위한 친숙한 SQL 인터페이스를 제공합니다.
  • pgvectorDiskANN 인덱싱 알고리즘을 사용하여 100M+ 벡터에서 더 빠르고 정확하게 유사성 검색을 향상시킵니다.
  • 관계형 메타데이터, 벡터 포함 및 시계열 데이터를 단일 데이터베이스에 통합하여 작업을 간소화합니다.
  • 복제 및 고가용성을 비롯한 엔터프라이즈급 기능을 위해 강력한 PostgreSQL 에코시스템 및 Azure Cloud의 기능을 사용합니다.

인증

Azure Database for PostgreSQL - 유연한 서버는 암호 기반 및 Microsoft Entra (이전의 Azure Active Directory) 인증을 지원합니다. Entra 인증을 사용하면 Entra ID를 사용하여 PostgreSQL 서버에 인증할 수 있습니다. Entra ID를 사용하면 데이터베이스 사용자에 대해 별도의 사용자 이름과 암호를 관리할 필요가 없으며 다른 Azure 서비스에 사용하는 것과 동일한 보안 메커니즘을 사용할 수 있습니다.

이 Notebook은 인증 방법 중 하나를 사용하도록 설정됩니다. 나중에 Notebook에서 Entra 인증을 사용할지 여부를 구성할 수 있습니다.

설치 프로그램

Azure Database for PostgreSQL은 오픈 소스 LangChain의 Postgres 지원을 사용하여 Azure Database for PostgreSQL에 연결합니다. 먼저 파트너 패키지를 다운로드합니다.

%pip install -qU langchain_postgres
%pip install -qU langchain-openai
%pip install -qU azure-identity

Azure Database for PostgreSQL - 유연한 서버에서 pgvector 활성화

Azure Database for PostgreSQL에 대한 사용 지침을 참조하세요.

자격 증명

PostgreSQL용 Azure Database의 연결 세부 정보가 필요하며, 이를 환경 변수로 추가해야 이 노트북을 실행할 수 있습니다.

USE_ENTRA_AUTH 플래그를 True로 설정하여 Microsoft Entra 인증을 사용합니다. Entra 인증을 사용하는 경우 호스트 및 데이터베이스 이름만 제공해야 합니다. 암호 인증을 사용하는 경우 사용자 이름과 암호도 설정해야 합니다.

import getpass
import os

USE_ENTRA_AUTH = True

# Supply the connection details for the database
os.environ["DBHOST"] = "<server-name>"
os.environ["DBNAME"] = "<database-name>"
os.environ["SSLMODE"] = "require"

if not USE_ENTRA_AUTH:
    # If using a username and password, supply them here
    os.environ["DBUSER"] = "<username>"
    os.environ["DBPASSWORD"] = getpass.getpass("Database Password:")

Azure OpenAI 포함 설정

os.environ["AZURE_OPENAI_ENDPOINT"] = "<azure-openai-endpoint>"
os.environ["AZURE_OPENAI_API_KEY"] = getpass.getpass("Azure OpenAI API Key:")

AZURE_OPENAI_ENDPOINT = os.environ["AZURE_OPENAI_ENDPOINT"]
AZURE_OPENAI_API_KEY = os.environ["AZURE_OPENAI_API_KEY"]

from langchain_openai import AzureOpenAIEmbeddings

embeddings = AzureOpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key=AZURE_OPENAI_API_KEY,
    azure_endpoint=AZURE_OPENAI_ENDPOINT,
    azure_deployment="text-embedding-3-small",
)

초기화

Microsoft Entra 인증

아래 셀에는 Entra 인증을 사용하도록 LangChain을 설정하는 함수가 포함되어 있습니다. 함수 get_token_and_username는 라이브러리 azure.identity를 사용하여 Azure Databases for PostgreSQL 서비스에 대한 토큰을 검색하는 기능을 제공합니다. sqlalchemy 엔진에 새 연결을 만들 수 있는 유효한 토큰이 있는지 확인합니다. 또한 JWT(Java Web Token)인 토큰을 구문 분석하여 데이터베이스에 연결하는 데 사용되는 사용자 이름을 추출합니다.

create_postgres_engine 함수는 TokenManager에서 가져온 토큰에 따라 사용자 이름과 암호를 동적으로 설정하는 sqlalchemy Engine 를 만듭니다. 이 Engine 매개 변수는 LangChain VectorStore의 PGVector 매개 변수로 connection 전달될 수 있습니다.

Azure에 로그인

Azure에 로그인하려면 Azure CLI 가 설치되어 있는지 확인합니다. 터미널에서 다음 명령을 실행해야 합니다.

az login

로그인하면 다음 코드가 토큰을 가져옵니다.

import base64
import json
from functools import lru_cache

from azure.identity import DefaultAzureCredential
from sqlalchemy import create_engine, event
from sqlalchemy.engine.url import URL


@lru_cache(maxsize=1)
def get_credential():
    """Memoized function to create the Azure credential, which caches tokens."""
    return DefaultAzureCredential()


def decode_jwt(token):
    """Decode the JWT payload to extract claims."""
    payload = token.split(".")[1]
    padding = "=" * (4 - len(payload) % 4)
    decoded_payload = base64.urlsafe_b64decode(payload + padding)
    return json.loads(decoded_payload)


def get_token_and_username():
    """Fetches a token returns the username and token."""
    # Fetch a new token and extract the username
    token = get_credential().get_token(
        "https://ossrdbms-aad.database.windows.net/.default"
    )
    claims = decode_jwt(token.token)
    username = claims.get("upn")
    if not username:
        raise ValueError("Could not extract username from token. Have you logged in?")

    return username, token.token


def create_postgres_engine():
    db_url = URL.create(
        drivername="postgresql+psycopg",
        username="",  # This will be replaced dynamically
        password="",  # This will be replaced dynamically
        host=os.environ["DBHOST"],
        port=os.environ.get("DBPORT", 5432),
        database=os.environ["DBNAME"],
    )

    # Create a sqlalchemy engine
    engine = create_engine(db_url, echo=True)

    # Listen for the connection event to inject dynamic credentials
    @event.listens_for(engine, "do_connect")
    def provide_dynamic_credentials(dialect, conn_rec, cargs, cparams):
        # Fetch the dynamic username and token
        username, token = get_token_and_username()

        # Override the connection parameters
        cparams["user"] = username
        cparams["password"] = token

    return engine

암호 인증

Entra 인증 get_connection_uri 을 사용하지 않는 경우 환경 변수에서 사용자 이름과 암호를 가져오는 연결 URI를 제공합니다.

import urllib.parse


def get_connection_uri():
    # Read URI parameters from the environment
    dbhost = os.environ["DBHOST"]
    dbname = os.environ["DBNAME"]
    dbuser = urllib.parse.quote(os.environ["DBUSER"])
    password = os.environ["DBPASSWORD"]
    sslmode = os.environ["SSLMODE"]

    # Construct connection URI
    # Use psycopg 3!
    db_uri = (
        f"postgresql+psycopg://{dbuser}:{password}@{dbhost}/{dbname}?sslmode={sslmode}"
    )
    return db_uri

벡터 저장소 만들기

from langchain_core.documents import Document
from langchain_postgres import PGVector
from langchain_postgres.vectorstores import PGVector

collection_name = "my_docs"

# The connection is either a sqlalchemy engine or a connection URI
connection = create_postgres_engine() if USE_ENTRA_AUTH else get_connection_uri()

vector_store = PGVector(
    embeddings=embeddings,
    collection_name=collection_name,
    connection=connection,
    use_jsonb=True,
)

벡터 저장소 관리

벡터 저장소에 항목 추가

ID로 문서를 추가하면 해당 ID와 일치하는 기존 문서가 과도하게 작성됩니다.

docs = [
    Document(
        page_content="there are cats in the pond",
        metadata={"id": 1, "location": "pond", "topic": "animals"},
    ),
    Document(
        page_content="ducks are also found in the pond",
        metadata={"id": 2, "location": "pond", "topic": "animals"},
    ),
    Document(
        page_content="fresh apples are available at the market",
        metadata={"id": 3, "location": "market", "topic": "food"},
    ),
    Document(
        page_content="the market also sells fresh oranges",
        metadata={"id": 4, "location": "market", "topic": "food"},
    ),
    Document(
        page_content="the new art exhibit is fascinating",
        metadata={"id": 5, "location": "museum", "topic": "art"},
    ),
    Document(
        page_content="a sculpture exhibit is also at the museum",
        metadata={"id": 6, "location": "museum", "topic": "art"},
    ),
    Document(
        page_content="a new coffee shop opened on Main Street",
        metadata={"id": 7, "location": "Main Street", "topic": "food"},
    ),
    Document(
        page_content="the book club meets at the library",
        metadata={"id": 8, "location": "library", "topic": "reading"},
    ),
    Document(
        page_content="the library hosts a weekly story time for kids",
        metadata={"id": 9, "location": "library", "topic": "reading"},
    ),
    Document(
        page_content="a cooking class for beginners is offered at the community center",
        metadata={"id": 10, "location": "community center", "topic": "classes"},
    ),
]

vector_store.add_documents(docs, ids=[doc.metadata["id"] for doc in docs])

벡터 저장소의 항목 업데이트

docs = [
    Document(
        page_content="Updated - cooking class for beginners is offered at the community center",
        metadata={"id": 10, "location": "community center", "topic": "classes"},
    )
]
vector_store.add_documents(docs, ids=[doc.metadata["id"] for doc in docs])

벡터 저장소에서 항목 삭제

vector_store.delete(ids=["3"])

쿼리 벡터 저장소

벡터 저장소가 만들어지고 관련 문서가 추가된 경우 체인 또는 에이전트의 벡터 저장소를 쿼리할 수 있습니다.

필터링 지원

벡터 저장소는 문서의 메타데이터 필드에 적용할 수 있는 필터 집합을 지원합니다.

오퍼레이터 의미/범주
$eq 같음(==)
$ne Inequality(!=)
$lt 보다 작음(<)
$lte 작거나 같음(<=)
$gt 보다 큼(>)
$gte 크거나 같음(>=)
$in 특수 사례(in)
$nin 특수 사례(not in)
$between 특수 사례(between)
$like 텍스트(like)
$ilike 텍스트(대/소문자 구분 없음)
$and 논리적(and)
$or 논리적(또는)

직접 쿼리

간단한 유사성 검색 수행은 다음과 같이 수행할 수 있습니다.

results = vector_store.similarity_search(
    "kitty", k=10, filter={"id": {"$in": [1, 5, 2, 9]}}
)
for doc in results:
    print(f"* {doc.page_content} [{doc.metadata}]")

    * there are cats in the pond [{'id': 1, 'topic': 'animals', 'location': 'pond'}]
    * ducks are also found in the pond [{'id': 2, 'topic': 'animals', 'location': 'pond'}]
    * the new art exhibit is fascinating [{'id': 5, 'topic': 'art', 'location': 'museum'}]
    * the library hosts a weekly story time for kids [{'id': 9, 'topic': 'reading', 'location': 'library'}]

여러 필드가 있는 받아쓰기를 제공하지만 연산자가 없는 경우 최상위 수준은 논리적 AND 필터로 해석됩니다.

vector_store.similarity_search(
    "ducks",
    k=10,
    filter={"id": {"$in": [1, 5, 2, 9]}, "location": {"$in": ["pond", "market"]}},
)

[Document(id='2', metadata={'id': 2, 'topic': 'animals', 'location': 'pond'}, page_content='ducks are also found in the pond'),
 Document(id='1', metadata={'id': 1, 'topic': 'animals', 'location': 'pond'}, page_content='there are cats in the pond')]

vector_store.similarity_search(
    "ducks",
    k=10,
    filter={
        "$and": [
            {"id": {"$in": [1, 5, 2, 9]}},
            {"location": {"$in": ["pond", "market"]}},
        ]
    },
)

[Document(id='2', metadata={'id': 2, 'topic': 'animals', 'location': 'pond'}, page_content='ducks are also found in the pond'),
 Document(id='1', metadata={'id': 1, 'topic': 'animals', 'location': 'pond'}, page_content='there are cats in the pond')]

유사성 검색을 실행하고 해당 점수를 받으려면 다음을 실행할 수 있습니다.

results = vector_store.similarity_search_with_score(query="cats", k=1)
for doc, score in results:
    print(f"* [SIM={score:3f}] {doc.page_content} [{doc.metadata}]")

* [SIM=0.528338] there are cats in the pond [{'id': 1, 'topic': 'animals', 'location': 'pond'}]

벡터 저장소에서 PGVector 실행할 수 있는 다양한 검색의 전체 목록은 API 참조를 참조하세요.

검색기로 전환하여 쿼리

체인에서 더 쉽게 사용할 수 있는 리트리버로 벡터 저장소를 변환할 수도 있습니다.

retriever = vector_store.as_retriever(search_type="mmr", search_kwargs={"k": 1})
retriever.invoke("kitty")

[Document(id='1', metadata={'id': 1, 'topic': 'animals', 'location': 'pond'}, page_content='there are cats in the pond')]

현재 제한 사항

  • langchain_postgres psycopg3에서만 작동합니다. 연결 문자열을 postgresql+psycopg2://...에서 postgresql+psycopg://langchain:langchain@...로 업데이트하세요.
  • 포함 저장소 및 컬렉션의 스키마가 변경되어 사용자가 지정한 ID에서 add_documents 제대로 작동합니다.
  • 이제 명시적 연결 개체를 전달해야 합니다.

현재 스키마 변경에 대한 쉬운 데이터 마이그레이션을 지원하는 메커니즘은 없습니다 . 따라서 벡터 저장소의 스키마를 변경하려면 사용자가 테이블을 다시 만들고 문서를 읽어야 합니다.

관련 콘텐츠