2025년 March 18일

개발 히스토리
[Backend] LangChain 구조

[Backend] LangChain 구조

기본 설정 및 디렉터리 생성

LangChain 라이브러리 설치

pip install langchain langchain-openai langchain-anthropic langchain-community faiss-cpu pgvector python-dotenv tiktoken chromadb

파이썬 가상환경에 langchain 라이브러리와, 함께 쓰일 패키지들을 전부 설치합니다.

디렉터리 구조

project/
├── api/                  # FastAPI 라우터/엔드포인트
├── core/                 # 설정 및 핵심 기능
├── crud/                 # DB 작업
├── database/             # DB 연결 및 마이그레이션
├── models/               # DB 모델
├── schemas/              # Pydantic 스키마
└── langchain_service/    # 새로 추가할 LangChain 관련 모듈
    ├── __init__.py
    ├── agents/           # 에이전트 정의
    ├── chains/           # 체인 정의
    ├── document_loaders/ # 문서 로더
    ├── embeddings/       # 임베딩 모델 연결
    ├── llms/             # LLM 모델 연결
    ├── memory/           # 대화 메모리 관리
    ├── tools/            # 에이전트 도구
    └── vector_stores/    # 벡터 저장소

CORE

core/config.py

CLAUDE_API = "YOUR API KEY"
GPT_API = "YOUR API KEY"

VECTOR_DB_CONNECTION = f'{DB}://{DB_USER}:{DB_PASSWORD}@{DB_SERVER}:{DB_PORT}/{DB_NAME}'
EMBEDDING_MODEL = 'test-embedding-ada-002'
DEFAULT_CHAT_MODEL = 'gpt-3.5-turbo'
CHROMA_PERSIST_DIRECTORY = './chroma_db'

기존 config.py에서 해당 내용을 추가합니다.

사용할 모델의 API Key나, 사용할 모델명을 정의합니다.

LLM

LLM 모델과 연결되는 역할을 담당합니다.

langchain_service/llms/setup.py

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
import core.config as config

def get_llm(provider="openai", model=None):
    if provider == "openai":
        model_name = model or config.DEFAULT_CHAT_MODEL
        return ChatOpenAI(
            api_key=config.GPT_API,
            model_name=model_name,
            temperature=0.7
        )
    elif provider == "anthropic":
        model_name = model or "claude-3-sonnet-20240229"
        return ChatAnthropic(
            api_key=config.CLAUDE_API,
            model=model_name,
            temperature=0.7
        )
    else:
        raise ValueError(f"지원되지 않는 제공자: {provider}")

get_llm 함수 정의

provider가 OpenAI일 경우 ChatGPT, anthropic일 경우 Claude를 사용합니다.

model 값으로 특정 모델의 이름이 입력되면 해당 모델을 사용하고,

그렇지 않을 경우 기본적으로 OpenAI 기준 DEFAULT_CHAT_MODEL로 입력된 GPT 3.5 Turbo를 사용합니다.

( DEFAULT_CHAT_MODEL은 CORE의 config.py에서 설정됨 )

Anthropic으로 설정하고, 특정 모델을 선택하지 않을 경우 기본 모델은 Claude3 Sonnect입니다.

테스트 코드

Anthropic과 OpenAI 둘 다 테스트하는 코드

결과 화면

임베딩 모델

langchain_service/embeddings/setup.py

from langchain_openai import OpenAIEmbeddings
import core.config as config

def get_embeddings():
    return OpenAIEmbeddings(
        api_key=config.GPT_API,
        model=config.EMBEDDING_MODEL
    )

CORE 디렉터리에서 설정한 임베딩 모델을 그대로 반환합니다.

테스트

embeddings/setup.py에서 해당 코드 실행

실행 결과 화면

벡터 DB

langchain_service/vector_stores/setup.py

from langchain_community.vectorstores import Chroma, PGVector
from langchain_service.embeddings.setup import get_embeddings
import core.config as config

def get_chroma_db(collection_name="documents"):
    embedding_function = get_embeddings()
    return Chroma(
        collection_name=collection_name,
        embedding_function=embedding_function,
        persist_directory=config.CHROMA_PERSIST_DIRECTORY
    )

def get_pgvector_db(collection_name="documents"):
    embedding_function = get_embeddings()
    connection_string = config.VECTOR_DB_CONNECTION
    return PGVector(
        collection_name=collection_name,
        embedding_function=embedding_function,
        connection_string=connection_string
    )

텍스트 임베딩을 위한 벡터 DB를 설정하는 코드입니다.

두 가지 벡터 DB를 설정합니다. ( Chroma, PG Vector )

Chroma : 임베딩 벡터를 파일 시스템에 영구적으로 저장하고, 나중에 검색할 수 있도록 해줍니다.

데이터는 CHROMA_PERSIST_DIRECTORY에 저장됩니다.

PG Vector : 임베딩 벡터를 PostgreSQL에 저장하여 DB 기반 검색을 지원합니다.

VECTOR_DB_CONNECTION을 통해 데이터베이스와 연동됩니다.

테스트

해당 코드를 실행하여 Chroma DB와 PG Vector를 테스트합니다.

PG Admin에서는 다음과 같은 데이터베이스가 생성되어야 합니다.

Chroma DB의 경우 테스트를 진행한 디렉터리에 chroma_db 디렉터리가 생성되면 정상적으로 실행이 된 것입니다.

문서 로더

langchain_service/document_loaders/file_loader.py

from langchain_community.document_loaders import (
    TextLoader, 
    PyPDFLoader, 
    Docx2txtLoader, 
    CSVLoader
)
import os

def load_document(file_path):
    ext = os.path.splitext(file_path)[1].lower()
    
    if ext == '.txt':
        loader = TextLoader(file_path)
    elif ext == '.pdf':
        loader = PyPDFLoader(file_path)
    elif ext in ['.docx', '.doc']:
        loader = Docx2txtLoader(file_path)
    elif ext == '.csv':
        loader = CSVLoader(file_path)
    else:
        raise ValueError(f"지원되지 않는 파일 형식: {ext}")
    
    return loader.load()

입력된 파일에 맞는 문서 로더를 설정합니다.

txt, pdf , docx, doc, csv 등의 확장자를 지원합니다.

테스트

테스트 코드

import os
from langchain_service.document_loaders.file_loader import load_document 
from pprint import pprint

if __name__ == "__main__":
    test_files = [
        "./test_documents/File_Loader_Test.txt",
        "./test_documents/File_Loader_Test.pdf",
        "./test_documents/File_Loader_Test.docx",
        "./test_documents/File_Loader_Test.csv"
    ]

    for file_path in test_files:
        if os.path.exists(file_path):
            try:
                print(f"\n=== {file_path} 로드 테스트 ===")
                documents = load_document(file_path)
                print(f"✅ {file_path} 로드 성공! 문서 개수: {len(documents)}")
                pprint(documents[:2])  # 첫 2개 문서 미리보기
            except Exception as e:
                print(f"❌ {file_path} 로드 실패: {e}")
        else:
            print(f"⚠️ {file_path} 파일이 존재하지 않습니다. 테스트를 위해 준비해주세요.")

테스트 결과 화면

langchain_service/document_loaders/indexer.py

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_service.vector_stores.setup import get_pgvector_db
from typing import List
from langchain.schema import Document

def split_documents(documents: List[Document], chunk_size=1000, chunk_overlap=200):
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap
    )
    return text_splitter.split_documents(documents)

def index_documents(documents: List[Document], collection_name="documents", use_pgvector=True):
    chunks = split_documents(documents)

    if use_pgvector:
        vector_db = get_pgvector_db(collection_name)
    else:
        from langchain_service.vector_stores.setup import get_chroma_db
        vector_db = get_chroma_db(collection_name)

    vector_db.add_documents(chunks)
    return vector_db

split_documents 함수 : Document 객체들의 리스트를 받아서, 각 문서를 청크로 나누어 반환합니다.

청크 최대 크기 ( 기본값 1000 ), 청크 간 겹치는 부분의 크기 ( 기본값 200 )을 설정합니다.

index_documents 함수 : 분할된 청크들을 벡터 DB에 추가하는 역할, 기본적으로 PG Vector에 저장하지만 use_pgvector 값이 False일 경우에는 Chroma DB를 사용합니다.

테스트

테스트 코드

실행 결과 화면

에이전트 툴

langchain/service/tools.code_tools.py

from langchain.agents import tool
import subprocess
import tempfile
import os

@tool
def execute_python_code(code: str) -> str:
    try:
        with tempfile.NamedTemporaryFile(suffix='.py', delete=False) as temp:
            temp.write(code.encode('utf-8'))
            temp_name = temp.name

        result = subprocess.run(
            ['python3', temp_name],
            capture_output=True,
            text=True,
            timeout=10
        )
        os.unlink(temp_name)

        if result.returncode != 0:
            return f"Error: {result.stderr}"
        return result.stdout
    except Exception as e:
        return f"Error executing code: {str(e)}"
@tool
def get_database_schema(project_id: int) -> str:
    return f"데이터베이스 스키마 정보 (프로젝트 ID: {project_id})..."

excude_python_code : 주어진 파이썬 코드를 임시 파일로 저장한 후, subprocess.run( )을 사용하여 인터프리터 실행한 결과를 반환합니다.

get_database_schema : 프로젝트 ID를 받아서, 프로젝트에 대한 데이터베이스 스키마 정보를 반환합니다.

대화 메모리

langchain_service/memory/conversation.py

from langchain.memory import ConversationBufferMemory, ConversationBufferWindowMemory, PostgresChatMessageHistory
import core.config as config

def get_memory(session_id, memory_type="buffer", window_size=5):
    if memory_type == "postgres":
        history = PostgresChatMessageHistory(
            connection_string=config.VECTOR_DB_CONNECTION,
            session_id=session_id,
            table_name="chat_history"
        )
        return ConversationBufferMemory(memory_key="chat_history", chat_memory=history)

    elif memory_type == "window":
        return ConversationBufferWindowMemory(
            memory_key="chat_history",
            k=window_size
        )

    else:
        return ConversationBufferMemory(memory_key="chat_history")

LangChain 라이브러리의 메모리 관리 기능을 설정하는 get_memory 함수를 정의합니다.

세션 ID와 메모리 타입에 따라 적합한 메모리 객체를 반환합니다.

session_id : 현재 대화 세션의 ID, 메모리 내에서 대화 히스토리를 고유하게 식별하는 데에 사용합니다.

memory_type : 메모리 타입을 기본값인 buffer, PostgreSQL을 사용하는 postgres, 창 크기가 제한된 버퍼인 window 중 하나로 선택할 수 있습니다.

window_size : memory_type이 window일 경우, 대화 히스토리에서 유지할 최대 문서 수를 설정합니다.

체인

langchain_service/chains/qa_chain.py

from langchain.chains import ConversationalRetrievalChain
from langchain_service.llms.setup import get_llm
from langchain_service.memory.conversation import get_memory
from langchain_service.vector_stores.setup import get_pgvector_db

def get_qa_chain(session_id, collection_name="documents", provider="openai", model=None):
    llm = get_llm(provider, model)

    vector_db = get_pgvector_db(collection_name)
    retriever = vector_db.as_retriever(search_kwargs={"k": 5})

    memory = get_memory(session_id, memory_type="postgres")

    qa_chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=retriever,
        memory=memory,
        return_source_documents=True
    )

    return qa_chain

질문 – 답변 체인을 생성하는 get_qa_chain 함수를 정의합니다.

사용자의 대화 및 문서 검색 기능을 결합하여 대화형 질문-답변 시스템을 구축합니다.

get_llm 함수를 호출하여 LLM을 가져옵니다.
벡터 DB에 연결하여 검색 기능을 설정하고, 최대 5개의 문서를 반환하도록 설정합니다.
get_memory를 호출하여 PostgreSQL을 사용하는 대화 메모리를 설정합니다. 이 메모리에서 사용자의 대화 기록을 유지하고, 이전 대화 내용을 기반으로 한 답변을 제공합니다.
위의 기능들을 결합한 Q&A 체인을 생성합니다. 이 체인을 통해 사용자가 질문을 할 때마다 관련 문서를 검색하고, 해당 문서를 바탕으로 답변을 생성합니다.
생성된 QA 체인을 반환합니다.

에이전트 설정

langchain_service/agents/project_agent.py

from langchain.agents import AgentExecutor, create_react_agent
from langchain.prompts import ChatPromptTemplate
from langchain_service.llms.setup import get_llm
from langchain_service.tools.code_tools import execute_python_code, get_database_schema
from typing import List
from langchain.tools import Tool


def get_project_agent(project_id: int, provider="openai", model=None):
    llm = get_llm(provider, model)

    tools = [
        execute_python_code,
        get_database_schema
    ]

    prompt = ChatPromptTemplate.from_template(
        """너는 META LLM MSP 시스템의 AI 에이전트입니다. 프로젝트 ID {project_id}에 대한 작업을 수행합니다.

        너의 목표는 소프트웨어 개발 프로젝트의 분석, 설계, 구현을 지원하는 것입니다.

        다음 도구를 사용할 수 있습니다:
        {tools}

        질문: {input}

        {agent_scratchpad}
        """
    )

    agent = create_react_agent(llm, tools, prompt)

    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True
    )

    return agent_executor