LangChain - Runnable, Chain 실습

김소은·2025년 8월 21일

🚀 LangChain 실습 정리: Runnable & Chain 활용기

오늘은 LangChain의 Runnable과 Chain을 중심으로 다양한 실습을 진행했습니다.
단순한 모델 호출을 넘어서, 대화 히스토리 관리·스트리밍·Fallback·문서 요약·SQL 처리까지 확장하는 방법을 배웠습니다.

1. @chain 데코레이터로 Runnable 구성하기

LangChain은 @chain 데코레이터를 이용해 함수를 Runnable 객체로 만들 수 있습니다.

@chain
def polite_answer(question: str) -> str:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "정중한 한국어 톤으로 답변하라."),
        ("human", "{q}")
    ])
    return (prompt | llm | StrOutputParser()).invoke({"q": question})

👉 장점: .invoke(), .batch(), .stream() 같은 공통 인터페이스 사용 가능.

2. Redis를 이용한 RunnableWithMessageHistory

대화형 에이전트는 히스토리 관리가 필수입니다.
Redis를 연결해 세션별 대화 기록을 저장/복원하는 방법을 배웠습니다.

def get_history(session_id: str):
    return RedisChatMessageHistory(session_id=session_id, url=REDIS_URL)

prompt = ChatPromptTemplate.from_messages([
    ("system", "너는 한국어 비서다."),
    MessagesPlaceholder("history"),
    ("human", "{input}")
])

chat_chain = prompt | llm | StrOutputParser()

history_chain = RunnableWithMessageHistory(
    chat_chain, get_history,
    input_messages_key="input", history_messages_key="history",
)

3. 제너레이터 함수 & 바인딩 실습

제너레이터 Runnable: yield로 토큰 스트리밍 가능
.bind(): 모델 파라미터(temperature, stop 등) 사전 고정 가능

@chain
def word_streamer(text: str):
    for w in text.split():
        yield w.upper() + " "

fast_llm = llm.bind(temperature=0, stop=["\n"])

4. Fallback 처리하기

LLM 호출 실패 시 대체 체인으로 자동 전환하는 방법도 배웠습니다.

safe_chain = (prompt | primary | StrOutputParser()).with_fallbacks([
    (prompt | fallback | StrOutputParser())
])

👉 안정성 높은 프로덕션 서비스 구축에 필수.

5. 문서 요약 (Map-Reduce 체인)

긴 문서를 작은 청크로 나누고(Map), 각 요약을 합쳐(Reduce) 최종 요약을 생성합니다.

splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=120)
docs = splitter.create_documents([long_text])

map_chain = RunnableLambda(lambda d: {"chunk": d.page_content}) | map_prompt | llm | StrOutputParser()
summaries = [map_chain.invoke({"page_content": doc.page_content}) for doc in docs]

reduce_chain = RunnableLambda(lambda xs: {"points": "\n".join(xs)}) | reduce_prompt | llm | StrOutputParser()
final_summary = reduce_chain.invoke(summaries)

6. SQL Query 처리 (Agent)

자연어 질의를 SQL 쿼리로 변환하고 실행하는 SQL Agent를 만들었습니다.

db = SQLDatabase.from_uri("sqlite:///demo.db")
sql_agent = create_sql_agent(llm=llm, db=db, agent_type="openai-tools", verbose=True)

response = sql_agent.invoke({"input": "제품별 총 매출액을 큰 것부터 보여줘."})
print(response["output"])