실전 프로젝트로 익히는 LangChain 활용법

이제 조립할 시간입니다

지금까지 우리는 LangChain의 부품을 하나씩 익혔습니다. 모델을 호출했고, 메시지를 다뤘고, 프롬프트와 출력 파서를 설계했습니다. LCEL로 체인을 연결했고, 메모리로 대화의 흐름을 보존했습니다. 문서를 로딩하고 청킹했으며, 임베딩과 벡터스토어로 검색 기반을 만들었습니다. RAG, 도구 호출, 에이전트, LangGraph, LangSmith, 배포와 최적화까지 훑었습니다.

이제 질문은 하나입니다.

"그래서 실제 업무에서는 무엇을 만들 수 있을까?"

LangChain 생태계의 샘플과 쿠크북을 보면 반복적으로 등장하는 패턴이 있습니다. 문서를 읽고, 나누고, 임베딩하고, 검색하고, 프롬프트에 넣고, 모델이 답하고, 출처를 남기는 흐름입니다. 여기에 대화 이력, 도구 호출, 구조화 출력, LangSmith 관측, 보안 정책을 붙이면 단순한 데모가 아니라 업무에 투입할 수 있는 애플리케이션이 됩니다. LangChain Samples 저장소 역시 고객-facing 팀들이 만든 예제, 쿠크북, 레퍼런스 구현을 교육용으로 모아 두고 있어 이런 반복 패턴을 살펴보기에 좋습니다.¹

이 장에서는 네 가지 실전 프로젝트를 만들며 전체 그림을 정리합니다. 사내 문서 Q&A 챗봇, 긴 문서 분석 자동화 파이프라인, 웹과 사내 문서를 함께 쓰는 리서치 에이전트, CSV·SQL 데이터 분석 보조 도구입니다. 마지막에는 이 네 프로젝트에서 공통으로 재사용할 수 있는 아키텍처를 분리합니다.

프로젝트를 시작하기 전: 같은 재료, 다른 요리

LangChain 애플리케이션은 겉모습이 달라도 내부 재료는 비슷합니다. 챗봇은 대화형 UI를 갖고, 문서 분석기는 배치 작업처럼 보이고, 리서치 에이전트는 도구를 돌아다니며, 데이터 분석 도구는 SQL을 생성합니다. 하지만 핵심은 모두 같습니다. 입력을 정리하고, 필요한 컨텍스트를 가져오고, 모델에게 안전하게 지시하고, 결과를 검증 가능한 형태로 내보내는 것입니다.

실전 프로젝트에서는 다음 네 가지 질문을 먼저 적어야 합니다.

• 입력은 무엇인가: 질문, PDF, 회의록, 웹 검색어, CSV, SQL 테이블 중 무엇을 받을 것인가.

• 근거는 어디에 있는가: 사내 문서, 벡터스토어, 웹 검색 결과, 데이터베이스, 사용자 대화 이력 중 무엇을 참조할 것인가.

• 출력은 어떤 형태여야 하는가: 자연어 답변, JSON, 표, 보고서, 액션 아이템, SQL 쿼리 중 무엇을 반환할 것인가.

• 실패하면 어떻게 할 것인가: 모르면 모른다고 말할 것인가, 재검색할 것인가, 사람에게 넘길 것인가, 실행을 차단할 것인가.

이 네 질문에 답하면 구현이 훨씬 쉬워집니다. 모델은 똑똑하지만, 업무 시스템은 똑똑함보다 일관성을 더 좋아합니다. 매번 그럴듯한 답을 다르게 내는 시스템보다, 근거와 한계를 분명히 밝히는 시스템이 운영에서 오래 살아남습니다.

프로젝트 1: 사내 문서 Q&A 챗봇

가장 먼저 만들 프로젝트는 사내 문서 Q&A 챗봇입니다. 직원이 "육아휴직 신청 절차가 어떻게 되나요?"라고 물으면 인사 규정 문서를 검색하고, 관련 조항을 바탕으로 답변하며, 어떤 문서를 참고했는지 출처를 남기는 시스템입니다. 여기에 대화 이력을 붙이면 "그럼 신청서는 어디에 제출해?" 같은 후속 질문도 처리할 수 있습니다.

이 프로젝트의 전체 흐름은 다음과 같습니다.

문서 로딩
→ 텍스트 정제
→ 청킹
→ 임베딩
→ 벡터스토어 저장
→ 검색기 구성
→ 질문 + 검색 결과 + 대화 이력으로 프롬프트 생성
→ LLM 응답
→ 출처 표시
→ 대화 이력 저장

LangChain은 모델, 도구, 데이터베이스를 교체 가능한 방식으로 연결할 수 있도록 설계되어 있으며, 공식 소개에서도 다양한 모델·도구·데이터베이스 통합과 사전 구축된 에이전트 패턴을 강조합니다.² 이 장의 코드는 특정 벡터DB나 모델에 갇히지 않도록 최대한 얇은 경계로 나누겠습니다.

문서를 넣는 파이프라인

먼저 문서를 벡터스토어에 넣는 인덱싱 코드를 만듭니다. 운영에서는 이 부분이 챗봇 응답 코드와 분리되어야 합니다. 문서가 바뀔 때마다 전체 서버를 다시 띄우는 구조는 곧 골칫거리가 됩니다.

from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

DATA_DIR = Path("./company_docs")
PERSIST_DIR = "./vectorstore/company_docs"


def load_company_documents():
    docs = []

    for path in DATA_DIR.rglob("*"):
        if path.suffix.lower() == ".pdf":
            loader = PyPDFLoader(str(path))
            loaded = loader.load()
        elif path.suffix.lower() in [".txt", ".md"]:
            loader = TextLoader(str(path), encoding="utf-8")
            loaded = loader.load()
        else:
            continue

        for doc in loaded:
            doc.metadata["source"] = str(path)
            doc.metadata["doc_type"] = path.suffix.lower().replace(".", "")
        docs.extend(loaded)

    return docs


def build_index():
    docs = load_company_documents()

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=900,
        chunk_overlap=150,
        separators=["nn", "n", ". ", " ", ""]
    )
    chunks = splitter.split_documents(docs)

    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        collection_name="company_docs",
        persist_directory=PERSIST_DIR,
    )

    return len(chunks)


if __name__ == "__main__":
    count = build_index()
    print(f"인덱싱 완료: {count}개 청크")

이 코드는 화려하지 않지만 실전에서는 매우 중요합니다. 문서 로딩과 검색 응답을 분리했기 때문에 나중에 인덱싱을 배치 작업, 관리자 버튼, CI/CD 파이프라인, 스케줄러로 옮기기 쉽습니다.

청크 크기는 정답이 없습니다. 사내 규정처럼 문단 단위 의미가 중요한 문서는 700~1,200자 정도에서 시작하고, 법률 문서처럼 조항 간 연결이 중요한 문서는 overlap을 조금 더 크게 둡니다. 중요한 것은 "검색 결과 하나만 읽어도 답변에 필요한 최소 단위가 들어 있는가"입니다.

검색 결과와 출처를 함께 반환하기

이제 질문을 받았을 때 관련 청크를 가져오고, LLM이 그 근거 안에서만 답하도록 만듭니다.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

PERSIST_DIR = "./vectorstore/company_docs"

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
    collection_name="company_docs",
    embedding_function=embeddings,
    persist_directory=PERSIST_DIR,
)

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 12},
)

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)


def format_docs(docs):
    blocks = []
    for i, doc in enumerate(docs, start=1):
        source = doc.metadata.get("source", "unknown")
        page = doc.metadata.get("page", "")
        blocks.append(
            f"[문서 {i}] source={source}, page={page}n{doc.page_content}"
        )
    return "nn".join(blocks)


prompt = ChatPromptTemplate.from_messages([
    ("system", """
당신은 사내 문서 Q&A 도우미입니다.
반드시 제공된 문서 내용에 근거해서만 답하세요.
문서에 없는 내용은 추측하지 말고 "제공된 문서만으로는 확인할 수 없습니다"라고 답하세요.
답변 끝에는 참고한 문서 번호를 적으세요.
"""),
    ("human", """
질문:
{question}

검색된 문서:
{context}
"""),
])

rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | prompt
    | model
    | StrOutputParser()
)


def answer(question: str):
    return rag_chain.invoke(question)

여기서 가장 중요한 문장은 "문서에 없는 내용은 추측하지 말라"입니다. RAG의 목적은 모델을 더 수다스럽게 만드는 것이 아니라, 답변을 근거에 묶어 두는 것입니다. 사내 규정, 계약서, 기술 문서처럼 틀리면 비용이 생기는 영역에서는 멋진 말보다 정확한 거절이 더 가치 있습니다.

후속 질문을 처리하는 대화 이력

직원은 항상 완전한 질문을 하지 않습니다. 처음에는 "복지포인트 기준이 뭐야?"라고 묻고, 다음에는 "그럼 신입도 받을 수 있어?"라고 묻습니다. 두 번째 질문만 보면 무엇을 말하는지 모호합니다. 따라서 최근 대화 이력을 질문 재작성 단계에 넣는 것이 좋습니다.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

rewrite_prompt = ChatPromptTemplate.from_messages([
    ("system", """
당신은 후속 질문을 독립적인 검색 질문으로 바꾸는 도우미입니다.
대화 이력을 참고하되, 사용자의 의도를 바꾸지 마세요.
"""),
    ("human", """
대화 이력:
{history}

현재 질문:
{question}

검색에 적합한 독립 질문으로 다시 쓰세요.
"""),
])

rewrite_chain = rewrite_prompt | model | StrOutputParser()


class SimpleChatStore:
    def __init__(self):
        self.store = {}

    def get_history(self, session_id: str):
        return self.store.get(session_id, [])

    def append(self, session_id: str, role: str, content: str):
        self.store.setdefault(session_id, []).append({"role": role, "content": content})

    def render(self, session_id: str, limit: int = 8):
        messages = self.get_history(session_id)[-limit:]
        return "n".join(f"{m['role']}: {m['content']}" for m in messages)


chat_store = SimpleChatStore()


def chat(session_id: str, question: str):
    history_text = chat_store.render(session_id)

    standalone_question = rewrite_chain.invoke({
        "history": history_text,
        "question": question,
    })

    response = rag_chain.invoke(standalone_question)

    chat_store.append(session_id, "user", question)
    chat_store.append(session_id, "assistant", response)

    return {
        "rewritten_question": standalone_question,
        "answer": response,
    }

운영에서는 이 SimpleChatStore를 Redis, PostgreSQL, DynamoDB 같은 저장소로 바꾸면 됩니다. 중요한 것은 대화 이력을 무작정 프롬프트에 모두 넣지 않는 것입니다. 긴 대화는 비용을 올리고, 오래된 맥락은 오히려 답변을 흐립니다. 최근 몇 턴만 넣거나, 요약 메모리를 따로 두는 전략이 필요합니다.

챗봇의 완성도는 검색 결과에서 갈립니다

RAG 챗봇이 틀릴 때 많은 사람이 프롬프트부터 고치려 합니다. 하지만 실제 원인은 검색 결과에 있는 경우가 많습니다. 질문과 무관한 청크가 들어오면 모델은 그 안에서 억지로 답을 찾으려 합니다. 반대로 정답 청크가 검색되지 않으면 아무리 좋은 모델을 써도 답할 수 없습니다.

따라서 챗봇을 평가할 때는 답변만 보지 말고 다음을 함께 기록해야 합니다.

• 사용자의 원 질문과 재작성된 검색 질문

• 검색된 문서의 source, page, chunk_id

• 검색 점수 또는 순위

• 최종 답변과 출처

• 사용자의 피드백

이 로그가 쌓이면 "모델이 문제인지, 검색기가 문제인지, 문서 자체가 문제인지"를 구분할 수 있습니다.

프로젝트 2: 긴 비정형 텍스트 분석 자동화

두 번째 프로젝트는 회의록, 고객 상담 transcript, 계약서 같은 긴 비정형 텍스트를 구조화된 JSON으로 바꾸는 파이프라인입니다. 이 작업은 RAG 챗봇과 다릅니다. 챗봇은 사용자의 질문에 답하지만, 문서 분석기는 문서 전체를 읽고 정해진 필드를 채워야 합니다.

예를 들어 고객 상담 transcript에서 다음 정보를 뽑는다고 해봅시다.

• 핵심 요약

• 고객 감정

• 이슈 유형

• 액션 아이템

• 담당자

• 기한

• 위험 신호

• 후속 조치 필요 여부

긴 문서를 한 번에 모델에 넣으면 토큰 제한에 걸리거나, 앞부분은 잘 보고 뒷부분은 대충 보는 문제가 생깁니다. 그래서 "분할 분석 → 부분 결과 병합 → 최종 구조화" 흐름을 씁니다.

출력 스키마부터 정하기

문서 분석 자동화에서 가장 먼저 할 일은 프롬프트 작성이 아니라 출력 스키마 정의입니다. 어떤 JSON을 원하는지 정하지 않으면 모델은 매번 조금씩 다른 형식으로 답합니다.

from typing import List, Optional, Literal
from pydantic import BaseModel, Field


class ActionItem(BaseModel):
    task: str = Field(description="해야 할 일")
    owner: Optional[str] = Field(default=None, description="담당자")
    due_date: Optional[str] = Field(default=None, description="기한. 알 수 없으면 null")
    evidence: str = Field(description="이 항목을 추출한 근거 문장")


class RiskSignal(BaseModel):
    level: Literal["low", "medium", "high"]
    description: str
    evidence: str


class TranscriptAnalysis(BaseModel):
    summary: str
    sentiment: Literal["positive", "neutral", "negative", "mixed"]
    issue_types: List[str]
    action_items: List[ActionItem]
    risk_signals: List[RiskSignal]
    follow_up_required: bool

이렇게 스키마를 먼저 정하면 프롬프트도 단단해집니다. 모델에게 "잘 정리해줘"라고 말하는 대신 "이 필드들을 채워라. 알 수 없으면 null을 써라. 근거 문장을 남겨라"라고 지시할 수 있기 때문입니다.

청크별 분석 후 병합하기

긴 문서를 청크로 나눈 뒤 각 청크에서 정보를 추출합니다. 이후 부분 결과를 다시 모델에 넣고 중복을 제거하며 최종 JSON을 만듭니다.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_text_splitters import RecursiveCharacterTextSplitter

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_model = model.with_structured_output(TranscriptAnalysis)

chunk_prompt = ChatPromptTemplate.from_messages([
    ("system", """
당신은 고객 상담 transcript 분석가입니다.
주어진 일부 텍스트에서 확인 가능한 내용만 추출하세요.
추측하지 말고, 근거 문장을 evidence에 남기세요.
"""),
    ("human", "분석할 텍스트:n{chunk}")
])

chunk_chain = chunk_prompt | structured_model


def analyze_long_text(text: str):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=2500,
        chunk_overlap=250,
    )
    chunks = splitter.split_text(text)

    partial_results = []
    for chunk in chunks:
        partial_results.append(chunk_chain.invoke({"chunk": chunk}))

    return partial_results

이 상태의 결과는 아직 최종본이 아닙니다. 같은 액션 아이템이 여러 청크에서 반복될 수 있고, 감정 판단이 앞부분과 뒷부분에서 달라질 수 있습니다. 그래서 병합 단계가 필요합니다.

class FinalAnalysis(BaseModel):
    summary: str
    sentiment: Literal["positive", "neutral", "negative", "mixed"]
    issue_types: List[str]
    action_items: List[ActionItem]
    risk_signals: List[RiskSignal]
    follow_up_required: bool
    confidence: Literal["low", "medium", "high"]


merge_model = model.with_structured_output(FinalAnalysis)

merge_prompt = ChatPromptTemplate.from_messages([
    ("system", """
당신은 여러 개의 부분 분석 결과를 하나의 최종 분석으로 병합하는 검수자입니다.
중복 액션 아이템은 합치고, 서로 충돌하는 판단은 근거가 더 명확한 쪽을 선택하세요.
확신이 낮으면 confidence를 low로 표시하세요.
"""),
    ("human", "부분 분석 결과:n{partials}")
])

merge_chain = merge_prompt | merge_model


def merge_results(partial_results):
    serialized = "nn".join(r.model_dump_json(indent=2) for r in partial_results)
    return merge_chain.invoke({"partials": serialized})

이 구조는 회의록에도, 계약서에도, 고객 상담에도 재사용할 수 있습니다. 바뀌는 것은 스키마와 프롬프트입니다. 회의록이라면 결정사항, 참석자, 논의 안건, 미해결 쟁점이 필요할 것입니다. 계약서라면 해지 조항, 손해배상, 자동 갱신, 관할 법원, 개인정보 처리, 비표준 조항이 중요할 것입니다.

문서 분석기의 품질 기준

문서 분석 자동화는 "요약이 자연스러운가"만 보면 안 됩니다. 업무 자동화에서 중요한 것은 누락과 오탐입니다. 중요한 위험 신호를 놓치면 누락이고, 위험하지 않은 문장을 위험하다고 표시하면 오탐입니다.

초기 평가셋은 거창할 필요가 없습니다. 실제 회의록 20개, 상담 transcript 30개, 계약서 10개 정도를 골라 사람이 정답 JSON을 만들어 둡니다. 그리고 다음 항목을 비교합니다.

• 필수 필드가 모두 채워졌는가.

• 액션 아이템의 담당자와 기한이 맞는가.

• 위험 신호의 근거 문장이 실제 원문에 있는가.

• 요약이 원문에 없는 결론을 만들지 않았는가.

• JSON 스키마 검증을 통과하는가.

이 과정을 거치면 모델 출력이 "그럴듯한 글"에서 "업무 시스템이 받을 수 있는 데이터"로 바뀝니다.

프로젝트 3: 웹 검색과 사내 문서를 함께 쓰는 리서치 에이전트

세 번째 프로젝트는 리서치 에이전트입니다. 예를 들어 사용자가 "경쟁사 A의 최근 가격 정책 변화와 우리 제품 문서 기준 대응 포인트를 정리해줘"라고 요청합니다. 이 질문에 답하려면 웹 검색과 사내 문서 검색이 모두 필요합니다. 웹에서는 최신 정보를 찾고, 사내 문서에서는 우리 제품의 기능, 가격, 정책을 찾아야 합니다.

여기서 단순 RAG 체인보다 에이전트가 어울리는 이유는 작업 순서가 고정되어 있지 않기 때문입니다. 어떤 질문은 웹 검색이 먼저 필요하고, 어떤 질문은 사내 문서부터 확인해야 합니다. 어떤 경우에는 검색 결과가 부족해서 다시 검색해야 합니다.

LangGraph 문서에서는 고정된 코드 경로를 따르는 워크플로와, 도구 사용과 과정을 동적으로 결정하는 에이전트를 구분합니다. 또한 프롬프트 체이닝, 병렬화, 라우팅, 오케스트레이터-워커, 평가자-최적화 같은 패턴을 소개합니다.³ 리서치 에이전트는 이 중 라우팅, 도구 사용, 오케스트레이터-워커 패턴이 잘 맞습니다.

리서치 에이전트의 도구 정의

도구는 에이전트에게 "손과 발"입니다. 도구 설명이 애매하면 에이전트가 잘못된 도구를 고릅니다. 따라서 각 도구는 언제 써야 하는지 분명해야 합니다.

from langchain_core.tools import tool


@tool
def search_internal_docs(query: str) -> str:
    """사내 제품 문서, 가격 정책, 운영 가이드에서 정보를 검색한다.
    회사 내부 정책이나 제품 기능에 관한 질문에 사용한다.
    """
    docs = retriever.invoke(query)
    return format_docs(docs)


@tool
def web_search(query: str) -> str:
    """공개 웹에서 최신 시장 정보, 경쟁사 발표, 뉴스, 블로그 자료를 검색한다.
    사내 문서에 없는 외부 최신 정보가 필요할 때 사용한다.
    """
    # 실제 구현에서는 Tavily, SerpAPI, 자체 검색 API 등을 연결한다.
    return f"웹 검색 결과 예시: {query}에 대한 검색 결과"


tools = [search_internal_docs, web_search]

실제 운영에서는 웹 검색 결과를 그대로 모델에 넘기지 않는 것이 좋습니다. 검색 결과에는 광고, 중복 문서, 품질 낮은 블로그, 오래된 페이지가 섞여 있습니다. 따라서 URL, 제목, 발행일, 요약, 신뢰도 힌트를 함께 저장하고, 필요하면 재랭킹 단계를 둡니다.

보고서 초안 생성 흐름

리서치 에이전트는 보통 한 번의 답변보다 "자료 수집 → 중복 제거 → 근거 기반 요약 → 보고서 초안" 흐름이 적합합니다. 에이전트에게 무작정 "조사해줘"라고 하면 도구를 오래 돌리거나 결론을 과장할 수 있습니다. 따라서 단계별 산출물을 나눕니다.

1. 사용자의 리서치 목표를 명확한 질문 목록으로 분해한다.
2. 질문별로 웹 검색 또는 사내 문서 검색을 수행한다.
3. 검색 결과를 출처 단위로 정리한다.
4. 중복되거나 신뢰도가 낮은 결과를 제거한다.
5. 근거가 있는 주장만 보고서 초안에 넣는다.
6. 마지막에 사용한 출처 목록을 남긴다.

보고서 생성 프롬프트는 다음처럼 만들 수 있습니다.

from langchain_core.prompts import ChatPromptTemplate

report_prompt = ChatPromptTemplate.from_messages([
    ("system", """
당신은 기업 리서치 애널리스트입니다.
반드시 제공된 자료에 근거해서만 보고서 초안을 작성하세요.
근거가 약한 내용은 '확인 필요'로 표시하세요.
각 핵심 주장 뒤에는 출처 ID를 표시하세요.
"""),
    ("human", """
리서치 주제:
{topic}

수집 자료:
{evidence}

보고서 형식:
- 요약
- 주요 발견
- 우리 회사에 대한 시사점
- 리스크와 불확실성
- 다음 조사 과제
- 참고 출처
""")
])

리서치 에이전트의 핵심은 "검색했다"가 아니라 "무엇을 근거로 말했는지 남겼다"입니다. 보고서가 틀렸을 때 출처가 없으면 원인을 찾을 수 없습니다. 출처가 있으면 검색 결과가 틀렸는지, 요약이 틀렸는지, 모델이 과장했는지 추적할 수 있습니다.

사람의 승인이 필요한 지점

리서치 에이전트는 읽기 중심 도구만 쓸 때는 비교적 안전합니다. 하지만 이메일 발송, CRM 업데이트, 보고서 공유, 티켓 생성처럼 외부 시스템을 바꾸는 도구가 붙으면 상황이 달라집니다. 이때는 반드시 사람의 승인을 넣어야 합니다.

예를 들어 "이 보고서를 팀 채널에 공유해줘"라는 요청이 들어오면 에이전트가 바로 공유하지 않고 다음처럼 멈춰야 합니다.

공유 전 확인이 필요합니다.
대상 채널: #sales-strategy
문서 제목: 경쟁사 A 가격 정책 변화 분석
포함 출처: 8개
민감 정보 포함 가능성: 있음

승인하시겠습니까?

도구 호출 장에서 다룬 것처럼 쓰기 도구는 읽기 도구와 다르게 취급해야 합니다. 실전에서는 "조회 가능", "초안 생성 가능", "외부 전송 가능", "삭제 가능"처럼 도구 권한 등급을 나누는 것이 좋습니다.

프로젝트 4: CSV·SQL 데이터 분석 보조 도구

네 번째 프로젝트는 데이터 분석 보조 도구입니다. 사용자가 "지난 6개월 동안 지역별 매출 성장률을 표로 보여주고, 눈에 띄는 변화를 설명해줘"라고 물으면 시스템이 필요한 SQL을 만들고, 실행 결과를 표와 요약문으로 돌려줍니다.

이 프로젝트는 매력적이지만 위험합니다. LLM이 생성한 SQL을 그대로 실행하면 민감 데이터 유출, 과도한 쿼리 비용, 실수로 인한 데이터 변경이 발생할 수 있습니다. 따라서 데이터 분석 보조 도구의 핵심은 "자연어를 SQL로 바꾸는 능력"이 아니라 "안전한 범위 안에서만 실행하게 하는 통제"입니다.

안전한 SQL 실행 원칙

분석 보조 도구는 처음부터 제한된 환경에서 설계해야 합니다.

• 읽기 전용 계정을 사용한다.

• SELECT 문만 허용한다.

• 접근 가능한 테이블과 컬럼을 화이트리스트로 제한한다.

• 주민등록번호, 카드번호, 이메일, 전화번호 같은 민감 컬럼은 제외한다.

• LIMIT을 강제하고, 긴 쿼리는 타임아웃을 건다.

• 실행 전 생성 SQL을 사용자에게 보여주거나 정책 검사기를 통과시킨다.

이 원칙을 지키지 않으면 분석 도구는 편리한 만큼 위험해집니다.

SQL 생성과 검증을 분리하기

SQL을 생성하는 체인과 검증하는 함수를 분리해 봅시다.

from pydantic import BaseModel, Field
from typing import List
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate


class SQLPlan(BaseModel):
    question: str
    tables: List[str]
    sql: str = Field(description="실행할 읽기 전용 SQL")
    explanation: str


model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
sql_model = model.with_structured_output(SQLPlan)

schema_description = """
허용 테이블:
1. sales_orders
   - order_id
   - order_date
   - region
   - product_category
   - revenue
   - quantity

2. customers_aggregated
   - region
   - customer_segment
   - customer_count

금지 사항:
- 개인정보 컬럼 조회 금지
- INSERT, UPDATE, DELETE, DROP, ALTER 금지
- SELECT 외 문장 금지
"""

sql_prompt = ChatPromptTemplate.from_messages([
    ("system", """
당신은 데이터 분석 SQL 작성 도우미입니다.
반드시 제공된 스키마 안에서만 SQL을 작성하세요.
읽기 전용 SELECT 문만 작성하세요.
모든 쿼리에는 LIMIT 100을 포함하세요.
"""),
    ("human", """
스키마:
{schema}

사용자 질문:
{question}
""")
])

sql_chain = sql_prompt | sql_model

이제 검증 함수를 둡니다. 간단한 문자열 검사만으로 완벽한 보안을 만들 수는 없지만, 최소한의 1차 방어선은 됩니다. 운영에서는 SQL 파서 라이브러리나 데이터베이스 프록시 정책을 함께 써야 합니다.

import re

ALLOWED_TABLES = {"sales_orders", "customers_aggregated"}
FORBIDDEN_KEYWORDS = [
    "insert", "update", "delete", "drop", "alter", "truncate",
    "create", "grant", "revoke", "merge"
]


def validate_sql(sql: str):
    normalized = sql.strip().lower()

    if not normalized.startswith("select"):
        raise ValueError("SELECT 문만 실행할 수 있습니다.")

    for keyword in FORBIDDEN_KEYWORDS:
        if re.search(rf"b{keyword}b", normalized):
            raise ValueError(f"금지된 SQL 키워드가 포함되어 있습니다: {keyword}")

    found_tables = set(re.findall(r"bfroms+([a-zA-Z_][a-zA-Z0-9_]*)|bjoins+([a-zA-Z_][a-zA-Z0-9_]*)", normalized))
    flat_tables = {name for pair in found_tables for name in pair if name}

    if not flat_tables.issubset(ALLOWED_TABLES):
        raise ValueError(f"허용되지 않은 테이블 접근: {flat_tables - ALLOWED_TABLES}")

    if "limit" not in normalized:
        raise ValueError("LIMIT 절이 필요합니다.")

    return True

실행 결과는 다시 모델에 넣어 인사이트로 바꿀 수 있습니다. 단, 모델에게 원본 전체 데이터를 넘길 필요는 없습니다. 집계 결과, 컬럼명, 표본 몇 줄, 통계 요약만 넘기면 충분합니다.

insight_prompt = ChatPromptTemplate.from_messages([
    ("system", """
당신은 비즈니스 데이터 분석가입니다.
쿼리 결과를 바탕으로 짧은 요약, 표 해석, 주의할 점을 작성하세요.
데이터에 없는 원인은 추측하지 마세요.
"""),
    ("human", """
사용자 질문:
{question}

실행 SQL:
{sql}

쿼리 결과:
{result_table}
""")
])

insight_chain = insight_prompt | model

CSV 분석은 더 작게 시작하기

SQL 데이터베이스 연결이 부담스럽다면 CSV 분석 도구부터 시작해도 좋습니다. CSV를 pandas DataFrame으로 읽고, 허용된 분석 함수만 도구로 제공합니다. 예를 들어 groupby_sum, filter_rows, describe_columns 같은 함수를 만들고, 에이전트가 직접 Python 코드를 실행하지 못하게 합니다.

이 방식은 자유도는 낮지만 안전합니다. 실무에서는 "무엇이든 할 수 있는 코드 실행 에이전트"보다 "정해진 분석 작업을 안정적으로 수행하는 도구 묶음"이 더 빨리 승인됩니다.

네 프로젝트를 하나의 아키텍처로 묶기

이제 네 프로젝트를 멀리서 바라봅시다. 겉으로는 서로 다르지만 공통 부품이 많습니다. 이 공통 부품을 잘 분리하면 다음 프로젝트를 만들 때 속도가 빨라집니다.

app/
  chains/
    rag_answer.py
    question_rewrite.py
    document_extract.py
    report_writer.py
    sql_generation.py
  prompts/
    rag_answer.md
    transcript_analysis.md
    research_report.md
    sql_generation.md
  tools/
    internal_search.py
    web_search.py
    safe_sql.py
    csv_analysis.py
  schemas/
    transcript.py
    research.py
    sql.py
  retrieval/
    loaders.py
    splitters.py
    vectorstores.py
    retrievers.py
  memory/
    chat_history.py
    user_profile.py
  security/
    permissions.py
    pii_masking.py
    approval.py
  evals/
    datasets.py
    evaluators.py
  observability/
    tracing.py
    logging.py

이 구조의 장점은 교체 가능성입니다. 사내 문서 Q&A 챗봇과 리서치 에이전트는 같은 internal_search 도구를 쓸 수 있습니다. 문서 분석기와 리서치 에이전트는 같은 report_writer 체인을 공유할 수 있습니다. SQL 분석 도구와 챗봇은 같은 권한 검사 모듈을 쓸 수 있습니다.

재사용할 것과 교체할 것

모든 것을 공통화하려고 하면 오히려 복잡해집니다. 실전에서는 공통화할 것과 도메인별로 바꿀 것을 나누는 감각이 중요합니다.

공통으로 재사용하기 좋은 것은 체인 실행 방식, 모델 초기화, 로깅, LangSmith tracing, 에러 처리, 권한 검사, PII 마스킹, 출력 파서 호출 방식입니다. 반대로 도메인별로 교체해야 하는 것은 데이터 소스, 문서 스키마, 검색 필터, 평가 기준, 보안 정책, 사용자 권한 모델입니다.

예를 들어 인사 규정 챗봇과 법무 계약서 분석기는 모두 RAG를 쓰지만, 검색 필터와 답변 기준은 다릅니다. 인사 챗봇은 직원이 이해하기 쉬운 설명이 중요하고, 계약서 분석기는 조항 원문과 위험 근거가 중요합니다. 같은 체인을 억지로 쓰기보다, 공통 기반 위에 도메인별 프롬프트와 스키마를 얹는 편이 낫습니다.

LangGraph로 프로젝트를 더 단단하게 만들기

간단한 프로젝트는 LCEL 체인만으로 충분합니다. 하지만 다음 조건이 생기면 LangGraph를 고려해야 합니다.

• 중간에 사람의 승인이 필요하다.

• 검색 결과가 부족하면 재검색해야 한다.

• 여러 도구를 병렬로 호출한 뒤 합쳐야 한다.

• 작업 상태를 저장하고 나중에 이어서 실행해야 한다.

• 실패한 노드만 재시도해야 한다.

• 여러 에이전트가 역할을 나눠 협업해야 한다.

예를 들어 리서치 보고서 생성은 LangGraph로 표현하기 좋습니다.

START
→ 질문 분해
→ 웹 검색 ┐
→ 사내 검색 ├→ 근거 정리
→ 데이터 검색 ┘
→ 초안 작성
→ 품질 평가
→ 기준 미달이면 추가 검색
→ 사람 검토
→ 최종 보고서
→ END

고정된 체인은 길어질수록 예외 처리가 어려워집니다. 반면 그래프는 노드별 책임이 분명해서 어디서 실패했는지, 어떤 상태에서 멈췄는지 추적하기 쉽습니다. 특히 에이전트형 시스템은 실행 경로가 매번 달라질 수 있으므로 그래프 기반 상태 관리가 큰 도움이 됩니다.

LangSmith로 실전 품질을 관리하기

실전 프로젝트가 데모와 달라지는 지점은 관측 가능성입니다. 데모에서는 "잘 답하네요"로 충분할 수 있습니다. 운영에서는 누가, 언제, 어떤 질문을 했고, 어떤 문서를 검색했으며, 어떤 프롬프트가 모델에 들어갔고, 비용과 지연시간이 얼마였는지 알아야 합니다.

LangSmith Deployment 문서는 에이전트 워크로드를 안정적으로 실행하기 위한 런타임과, assistants·threads·runs 같은 실행 모델을 설명합니다. 또한 사용자 피드백 수집, CI/CD 파이프라인, 커스텀 인증, TTL 설정 같은 운영 항목을 다룹니다.⁴ 앞 장에서 배운 관측과 평가를 이번 실전 프로젝트에도 반드시 붙여야 합니다.

처음부터 복잡한 대시보드를 만들 필요는 없습니다. 최소한 다음 항목을 기록하세요.

• 입력: 사용자 질문, 업로드 문서, 분석 요청

• 검색: 검색 쿼리, 검색 결과, 출처, 점수

• 모델 호출: 프롬프트 버전, 모델명, 토큰 사용량, 지연시간

• 출력: 답변, JSON, SQL, 보고서

• 검증: 스키마 통과 여부, SQL 정책 통과 여부, 출처 포함 여부

• 피드백: 사용자 만족도, 수정 요청, 신고 여부

이 로그는 나중에 평가셋이 됩니다. 실제 사용자가 불만족한 사례 100개는 어떤 튜토리얼보다 값진 개선 재료입니다.

운영 체크리스트

실전 프로젝트를 배포하기 전에는 기능보다 위험을 먼저 점검해야 합니다. LangChain은 빠르게 만들 수 있게 해주지만, 빠르게 만든 시스템이 곧 안전한 시스템은 아닙니다.

다음 체크리스트를 프로젝트마다 확인하세요.

• 답변이 근거 문서나 검색 결과에 묶여 있는가.

• 출처를 사용자에게 보여주거나 내부 로그에 남기는가.

• 구조화 출력은 스키마 검증을 통과하는가.

• 실패 시 재시도, 거절, 사람에게 전달 중 하나로 처리되는가.

• 쓰기 도구 실행 전 사용자 승인 절차가 있는가.

• SQL과 API 호출은 권한과 범위가 제한되어 있는가.

• 민감 정보는 프롬프트와 로그에서 마스킹되는가.

• 비용, 지연시간, 오류율을 관측할 수 있는가.

• 평가셋으로 변경 전후 품질을 비교하는가.

• 프롬프트, 모델, 검색 파라미터 변경 이력이 남는가.

이 체크리스트는 프로젝트를 느리게 만들기 위한 것이 아닙니다. 오래 가게 만들기 위한 것입니다. 빠른 프로토타입은 사람을 설득하지만, 안전한 운영 구조는 조직을 설득합니다.

마지막으로: 작은 성공을 연결하라

LangChain을 잘 활용하는 팀은 처음부터 거대한 AI 플랫폼을 만들지 않습니다. 대신 작고 명확한 문제를 하나 고릅니다. 사내 규정 50개 문서를 답하는 챗봇, 상담 transcript에서 액션 아이템을 뽑는 파이프라인, 주간 시장 보고서 초안을 만드는 리서치 에이전트, 매출 CSV를 요약하는 분석 도구처럼 범위가 분명한 것부터 시작합니다.

그리고 성공한 부품을 남깁니다. 잘 작동한 프롬프트는 템플릿이 되고, 안정적인 출력 스키마는 표준이 되고, 검색 품질 평가셋은 다음 RAG 프로젝트의 기준이 됩니다. 한 번 만든 권한 검사와 로깅 구조는 다음 에이전트의 안전망이 됩니다.

LangChain의 진짜 가치는 "LLM을 한 번 호출하기 쉽게 해준다"가 아닙니다. 모델, 프롬프트, 검색기, 도구, 메모리, 파서, 그래프, 관측을 조립 가능한 부품으로 바꿔준다는 데 있습니다. 이 부품들을 업무 문제에 맞게 차분히 연결하면, 챗봇은 지식 창구가 되고, 문서 분석기는 야근을 줄이고, 리서치 에이전트는 초안을 만들고, 데이터 분석 도구는 질문과 숫자 사이의 거리를 좁힙니다.

이제 여러분에게 필요한 것은 더 많은 개념이 아니라 하나의 실제 문제입니다. 문서 하나, 질문 하나, 반복 업무 하나를 고르세요. 작게 만들고, 로그를 보고, 평가하고, 개선하세요. 그렇게 만든 첫 번째 LangChain 프로젝트가 다음 프로젝트의 기반이 됩니다.

참고

¹LangChain Samples · GitHub

²LangChain: Open Source AI Agent Framework

³Workflows and agents - Docs by LangChain

⁴LangSmith Deployment - Docs by LangChain