ch14 chains

문서요약 체인

📋 문서 요약 방식별 선택 가이드

요약 방식	선택 기준
Stuff	문서가 짧고 한 번에 끝내고 싶을 때
Map-Reduce	문서가 매우 길고 빠른 처리가 필요할 때
Map-Refine	문서의 전체 맥락 연결이 중요할 때
Chain of Density	가장 정보 밀도가 높은 완벽한 요약이 필요할 때
Clustering	내용이 중복되는 방대한 자료를 정리할 때

1. Stuff (몽땅 집어넣기)

원리: 모든 문서를 하나의 프롬프트에 통째로 넣고 "요약해줘"라고 시키는 가장 단순한 방식입니다.

장점: 가장 빠르고 비용이 저렴하며, 문맥을 한 번에 파악하기 좋습니다.

단점: 문서가 너무 길면 AI가 읽을 수 있는 한계를 초과하여 에러가 발생합니다.

stuff_chain = create_stuff_documents_chain(llm, prompt)

2. Map-Reduce (분할 요약 후 병합)

원리: 문서를 여러 조각으로 나눠서 각각 요약한 뒤(Map), 그 요약본들을 다시 모아서 최종 요약을 만듭니다(Reduce).

장점: 아주 방대한 양의 문서도 처리할 수 있고, 각 조각을 동시에 요약하므로 속도가 빠릅니다.

단점: 전체를 관통하는 미묘한 맥락이 요약 과정에서 사라질 수 있습니다.

from langchain_core.runnables import chain


@chain
def map_reduce_chain(docs):
    map_llm = ChatOpenAI(
        temperature=0,
        #단순 반복 작업이 많으므로, 속도가 빠르고 가격이 *저렴한 가성비 모델*
        model_name="gpt-4o-mini",
    )

    # map prompt 다운로드
    map_prompt = hub.pull("teddynote/map-prompt")

    # map chain 생성
    map_chain = map_prompt | map_llm | StrOutputParser()

    # *병렬 처리 (batch)*
    doc_summaries = map_chain.batch(docs)

    # reduce prompt 다운로드
    reduce_prompt = hub.pull("teddynote/reduce-prompt")
    reduce_llm = ChatOpenAI(
    	#문맥 파악 능력 위해 성능이 더 좋은 *고성능 모델*
        model_name="gpt-4o",
        temperature=0,
        callbacks=[StreamingCallback()],
        streaming=True,
    )

    reduce_chain = reduce_prompt | reduce_llm | StrOutputParser()

    return reduce_chain.invoke({"doc_summaries": doc_summaries, "language": "Korean"})

Map-Reduce에서 체인을 분리하는 이유는 각 단계의 특성에 맞는 AI 모델을 선택하여 비용을 절감하고, 병렬 처리를 통해 대량의 문서를 빠르게 요약하기 위함입니다.

3. Map-Refine (점진적 보완 요약)

원리: 첫 번째 조각을 요약한 뒤, 그 요약본을 다음 조각과 함께 넘겨서 내용을 업데이트합니다. 이 과정을 마지막 조각까지 반복하며 요약을 완성해 나갑니다.

장점: 앞뒤 맥락이 잘 이어지며, 요약의 디테일이 살아있습니다.

단점: 순차적으로 작업해야 해서 속도가 느리고, 뒤로 갈수록 초기 내용이 희미해질 수 있습니다.

ed!from langchain_core.runnables import chain


@chain
def map_refine_chain(docs):

    # map chain 생성
    map_summary = hub.pull("teddynote/map-summary-prompt")

    map_chain = (
        map_summary
        | ChatOpenAI(
            model_name="gpt-4o-mini",
            temperature=0,
        )
        | StrOutputParser()
    )

    input_doc = [{"documents": doc.page_content, "language": "Korean"} for doc in docs]

    # 첫 번째 프롬프트, ChatOpenAI, 문자열 출력 파서를 연결하여 체인을 생성합니다.
    doc_summaries = map_chain.batch(input_doc)

    refine_prompt = hub.pull("teddynote/refine-prompt")

    refine_llm = ChatOpenAI(
        model_name="gpt-4o-mini",
        temperature=0,
        callbacks=[StreamingCallback()],
        streaming=True,
    )

    refine_chain = refine_prompt | refine_llm | StrOutputParser()

    previous_summary = doc_summaries[0]

    for current_summary in doc_summaries[1:]:

        previous_summary = refine_chain.invoke(
            {
                "previous_summary": previous_summary,
                "current_summary": current_summary,
                "language": "Korean",
            }
        )
        print("\n\n-----------------\n\n")

    return previous_summary

-> 원래 Refine 방식은 이전 요약 + 다음 원문을 결합하는 것이 정석입니다.
하지만 본 실습 코드에서는 토큰(입력량) 제한을 고려하여, 각 페이지를 먼저 요약(batch)한 뒤 요약본끼리 Refine 하는 구조를 취하고 있습니다. 요약본 + 요약본

4. Chain of Density (밀도 보완 반복 요약)

"Chain of Density" (CoD) 프롬프트는 GPT-4를 사용한 요약 생성을 개선하기 위해 개발된 기법입니다.

원리: 요약을 한 번으로 끝내지 않고 여러 번 반복 실행합니다. 이때 핵심 정보(Entity)가 누락되지 않았는지 체크하며 요약문의 밀도를 점점 높여갑니다.

장점: 인간이 작성한 요약과 비슷한 밀도를 가진 고품질 요약본이 나옵니다. 원문의 앞부분에 치우치는 경향(lead bias)이 덜합니다.

단점: 여러 번 반복 실행하므로 비용과 시간이 많이 듭니다.

5. Clustering-Map-Refine (군집화 기반 정예 요약)

원리: 문서 조각(Chunk)들을 비슷한 내용끼리 그룹(클러스터)으로 묶습니다. 각 그룹에서 가장 대표적인 중심 문서들만 골라 Refine 방식으로 요약합니다.

장점: 중복된 내용은 과감히 생략하고 방대한 양의 핵심 주제를 효율적으로 요약할 수 있습니다.

단점: 그룹을 나누는 '클러스터링' 과정이 추가되어 기술적으로 더 복잡합니다.

SQL쿼리 만드는 체인: create_sql_query_chain

사용자의 질문(자연어)을 LLM이 분석하여, 데이터를 추출하기 위한 최적의 SQL 쿼리를 스스로 생성합니다. 이렇게 생성된 쿼리를 통해 데이터베이스를 실시간으로 조회하여 정확한 정보를 찾아냅니다.

from langchain_community.tools.sql_database.tool import QuerySQLDataBaseTool

# SQLite 데이터베이스에 연결
db = SQLDatabase.from_uri("sqlite:///data/finance.db")
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 생성한 쿼리를 실행하는 도구
execute_query = QuerySQLDataBaseTool(db=db)

# SQL 쿼리 생성
# 기본 프롬프트가 내장되어 있지만 매개변수로 Prompt를 추가해서 엉뚱한 쿼리 생성방지에 도움을 줄 수 있음 (아래에 서술)
write_query = create_sql_query_chain(llm, db)

# 생성한 쿼리를 실행하기 위한 체인을 생성합니다.
chain = write_query | execute_query

Prompt를 직접 작성해서 넣는 법
prompt = PromptTemplate.from_template("프롬프트 작성").partial(dialect=db.dialect)
chain = create_sql_query_chain(llm, db, prompt)

chain.invoke({"question": "테디의 이메일을 조회하세요"})

위 방법으로는 답변이 단답형 형식으로 출력되므로 더 친절한 답변을 받기 위해서는 답변을 LLM으로 증강생성 하면 된다.

-965.7 -> '테디의 transaction의 합계는 -965.7 입니다.'

#LCEL 문법의 체인 사용
#친절한 답변 생성
from operator import itemgetter
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough

answer_prompt = PromptTemplate.from_template(
    """Given the following user question, corresponding SQL query, and SQL result, answer the user question.

Question: {question}
SQL Query: {query}
SQL Result: {result}
Answer: """
)

# 지시서(prompt)를 읽고, 모델(llm)이 생각해서, 문자열(Parser)로 답해라
answer = answer_prompt | llm | StrOutputParser()

# 생성한 쿼리를 실행하고 결과를 출력하기 위한 체인을 생성합니다.
chain = (
  RunnablePassthrough
  .assign(query=write_query)
  .assign(result=itemgetter("query") | execute_query)
  | answer
)

RunnablePassthrough.assign(...): 기존에 있던 데이터는 그대로 두고, 옆에 새로운 정보를 '추가'해서 다음 단계로 넘김

assign(query=write_query) 실행 결과: {"question": "...", "query": "SELECT..."}

result=itemgetter("query") | execute_query
#itemgetter로 query를 가져와서 execute_query로 전달 후 result를 가져옴
실행결과: {"question": "...", "query": "...", "result": "[('...')]"} 
-> answer에 저장

SQL Agent

Agent를 활용하여 Sql 쿼리를 생성하고 실행 결과를 답변으로 출력이 가능합니다.

'A와 B를 비교해줘' 같이 질문이 복잡해서 한 번의 쿼리로 안 끝날 때, 에이전트는 스스로 계획을 세워 여러 번 DB를 뒤져보고 최종 답을 냅니다.

[생각 → 행동 → 관찰]의 과정을 반복합니다.

from langchain_openai import ChatOpenAI
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import create_sql_agent

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
db = SQLDatabase.from_uri("sqlite:///data/finance.db")

# Agent 생성
agent_executor = create_sql_agent(llm, db=db, agent_type="openai-tools", verbose=True)
agent_executor.invoke(
    {"input": "테디와 셜리의 transaction 의 합계를 구하고 비교하세요"}
)

실행 과정

Entering new SQL Agent Executor chain...
---1 어떤 테이블 있나 확인
Invoking: `sql_db_list_tables` with `{}`
accounts, customers, transactions

---2 'transactions' 태이블 구조 파악
Invoking: `sql_db_schema` with `{'table_names': 'transactions'}`

CREATE TABLE transactions (
    transaction_id INTEGER, 
    account_id INTEGER, 
    amount REAL, 
    transaction_date TEXT, 
    PRIMARY KEY (transaction_id), 
    FOREIGN KEY(account_id) REFERENCES accounts (account_id)
)

/* 예시 데이터 3줄
3 rows from transactions table:
transaction_id  account_id  amount  transaction_date
1   1   74.79   2024-07-13
2   1   -224.1  2024-05-13
3   1   -128.9  2024-01-25
*/

---3 정확한 쿼리 생성 및 조회
Invoking: `sql_db_query` with `{'query': 'SELECT account_id, SUM(amount) AS total_amount FROM transactions WHERE account_id IN (1, 2) GROUP BY account_id'}`

[(1, -965.7), (2, 743.13)] 
테디의 거래 합계는 -965.7이고, 셜리의 거래 합계는 743.13입니다.

Finished chain.

🛠️ 실무 적용 시 반드시 고려할 점

LLM 기반의 SQL 에이전트는 편리하지만, 실제 서비스에 적용하려면 안정성과 효율성을 꼭 따져봐야 합니다.

1. 보안: "읽기 전용"은 필수
   AI가 실수로 데이터를 지우거나 수정하지 않도록, DB 접속 시 읽기 전용(Read-Only) 계정을 연결해야 합니다. 또한, 위험한 명령어(DROP, DELETE 등)가 포함되었는지 검사하는 Query Validator 단계를 추가하는 것이 일반적입니다.

2. 비용과 속도: 하이브리드 전략
   에이전트는 스스로 판단하는 과정에서 AI 모델을 다회 호출(Multi-turn)하므로 API 비용 상승과 응답 지연이 발생합니다. 단순한 질문은 Chain(고정)으로, 복잡한 분석은 Agent(자율 판단)로 처리하는 설계가 효율적입니다.

3. 컨텍스트 최적화 (Context Management)
   DB의 모든 구조를 AI에게 넘기면 오히려 헷갈려 할 수 있습니다. 꼭 필요한 테이블 정보와 컬럼 설명만 골라 전달하여 AI가 엉뚱한 답변(환각)을 하지 않도록 관리해야 합니다.

🧰 함께 살펴보면 좋은 도구들

랭체인 외에도 현업에서 자주 언급되는 대안들입니다.

LlamaIndex: 랭체인과 양대 산맥입니다. 특히 데이터를 찾고 연결하는 기능(RAG)에 특화되어 있어 SQL 관련 작업에서도 많이 쓰입니다.

Vanna.ai: 'Text-to-SQL'에만 집중한 특화 툴로, 회사의 과거 쿼리 데이터를 학습시켜 정확도를 높이기 좋습니다.

LangGraph: 랭체인의 확장판입니다. 에이전트의 사고 과정을 더 세밀하게 제어하고 싶을 때(예: "먼저 승인을 받고 쿼리를 실행해") 사용합니다.

구조화된 출력 만드는 체인: with_structured_output

AI 답변을 파싱(글자 쪼개기)하느라 고생할 필요 없이, 처음부터 API처럼 규격에 맞는 데이터를 받는 기술.

사람이 아닌 '프로그램'이 AI의 답변을 소비할 때 사용!

예시) 4지선다형 퀴즈를 생성

1. AI에게 줄 '데이터의 틀'를 만드는 작업

Quiz 클래스는 퀴즈의 질문, 난이도, 그리고 네 개의 선택지를 정의

class Quiz(BaseModel):
    question: str = Field(..., description="퀴즈의 질문")
    level: str = Field(...)
    options: List[str] = Field(...)

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You're a world-famous quizzer and generates quizzes in structured formats.",
        ),
        (
            "human",
            "TOPIC 에 제시된 내용과 관련한 4지선다형 퀴즈를 출제해 주세요. 만약, 실제 출제된 기출문제가 있다면 비슷한 문제를 만들어 출제하세요."
            "단, 문제에 TOPIC 에 대한 내용이나 정보는 포함하지 마세요. \nTOPIC:\n{topic}",
        ),
        ("human", "Tip: Make sure to answer in the correct format"),
    ]
)

2. with_structured_output(Quiz) (AI에게 틀 전달)

# 구조화된 출력을 위한 모델 생성
llm_with_structured_output = llm.with_structured_output(Quiz)

# 퀴즈 생성 체인 생성
chain = prompt | llm_with_structured_output

# 퀴즈 생성을 요청합니다.
generated_quiz = chain.invoke({"topic": "ADSP(데이터 분석 준전문가) 자격 시험"})

# 생성된 퀴즈 출력
print(f"{generated_quiz.question} (난이도: {generated_quiz.level})\n")
for i, opt in enumerate(generated_quiz.options):
    print(f"{i+1}) {opt}")

#결과물
다음 중 데이터 분석의 과정에서 가장 먼저 수행해야 하는 단계는 무엇인가요? (난이도: 보통)

1) 데이터 수집
2) 데이터 전처리
3) 문제 정의
4) 모델 평가

3. 데이터의 흐름

① 입력: "ADSP 시험에 대한 퀴즈를 내줘"라고 요청합니다.

② 처리: AI가 내용을 생성한 뒤, Quiz 클래스 양식에 맞게 데이터를 칸칸이 집어넣습니다.

③ 출력: 결과물이 문장이 아니라 generated_quiz.question처럼 변수명으로 바로 접근할 수 있는 형태로 나옵니다.