에이전트(Agent) - 랭체인 에이전트 심화 (3)

1. 라우터 체인 (RouterChain)

• 조건부 분기 패턴으로써, 입력에 따라 적절한 체인 혹은 Runnable으로 분기하는 패턴이다.
• 분류기가 입력을 분석하고 적절한 체인으로 라우팅 해줌.
• 분류기는 LLM 기반, 규칙 기반으로 분석
• LCEL 방식 : RunnableLambda로 분기  함수를 정의하고 | 으로 흐름 연결
• 예시 : 질문 유형 (코딩 / 수학 / 일반)에 따라 각기 다른 전문 프롬프트 적용
• RunnableBranch -> 조건에 따른 다른 Runnable 선택하는 LCEL 라우팅 객체

동작 흐름

 

RunnableLambda 분기

parser = StrOutputParser()

# 수학체인
math_chain = ChatPromptTemplate.from_messages([
    ("system", "당신은 수학 전문가입니다. 풀이 과정을 단계별로 설명하세요."),
    ("human","{question}")
]) | openai_llm | parser

# 코드체인
code_chain = ChatPromptTemplate.from_messages([
    ("system", "당신은 시니어 개발자입니다. 코드와 주석을 함께 제공하세요"),
    ("human", "{question}")
]) | openai_llm | parser

# 일반체인
general_chain = ChatPromptTemplate.from_messages([
    ("system", "당신은 친절한 AI 어시스턴트입니다. 한국어로 답변하세요"),
    ("human", "{question}")
]) | openai_llm | parser

# 분류기 : 질문을 읽고 유형을 반환
classify_chain = ChatPromptTemplate.from_messages([
    ("system", "질문 유형을 math/code/general 중 하나로만 답하세요"),
    ("human", "{question}")
]) | openai_llm | parser

# 라우터 함수
def route(inputs: dict):
    category = classify_chain.invoke(inputs).strip().lower()
    print(f" => 분류 결과 {category}")

    if 'math' in category: return math_chain
    elif 'code' in category: return code_chain
    else: return general_chain

# 라우터 체인 조합
router_chain = RunnableLambda(route) | RunnableLambda(lambda chain:chain)

# 실행
questions = [
    {"question": "피타고라스 정리를 증명해줘"},
    {"question": "파이썬으로 버블정렬 구현해줘"},
    {"question": "오늘 기분 좋아지는 말 해줘"}
]

for q in questions:
    print(f"Q: {q['question']}")
    print(f"A: {router_chain.invoke(q)[:80]}...\n")

 

RunnableBranch 선언형 분기

from langchain_core.runnables import RunnableBranch

# RunnableBranch((조건1,체인1),(조건2,체인2).....)
branch = RunnableBranch(
    (lambda x:'math' in x.get('topic',''),math_chain),
    (lambda x:'code' in x.get('topic',''), code_chain),
    (lambda x:'cooking' in x.get('topic',''),ChatPromptTemplate.from_messages([
        ("system","당신은 요리 전문가입니다"),
        ("human","{question}")
    ]) | openai_llm | parser),
    general_chain
)

print(branch.invoke({'topic':'math', 'question': '미분이란?'}))
print(branch.invoke({'topic':'cooking', 'question': '김치찌개 레시피'}))
print(branch.invoke({'topic':'other', 'question': '안녕하세요'}))
print(branch.invoke({'topic':'code', 'question': '파이썬으로 선택 정렬 구현'}))

 

2. 단계별 순차 파이프라인(SequentialChain)

• 체인의 기본 흐름은 앞 단계 출력이 뒷 단계의 입력으로 흘러 들어간다
• 연산자 | 자체가 SequentialChain 역할이다.
• assign() 메소드로 중간 결과를 보존하며, 여러 단계를 누적할 수 있다.
• 각 단계의 출력 타입과 입력 타입이 일치해야 한다.
• 예시 -> 번역 => 요약 => 감정 분석 => 보고서 생성 등 다단계 처리

다단계 체인 연동 : 번역 -> 요약 -> 감정 분석 -> 보고서 생성

# 체인 정의
translate_chain = ChatPromptTemplate.from_messages([
    ("system", "다음 텍스트를 한국어로 번역하세요. 번역문만 출력: \n{text}"),
]) | openai_llm | parser

summarize_chain = ChatPromptTemplate.from_messages([
    ("system", "다음 텍스트를 3문장으로 요약하세요.\n{text}"),
]) | openai_llm | parser

sentiment_chain = ChatPromptTemplate.from_messages([
    ("system", "다음 텍스트의 감정을 긍정/부정/중립 중 하나로만 답하세요:\n {summary}"),
]) | openai_llm | parser

report_chain = (
    ChatPromptTemplate.from_template(
        '아래 분석 결과를 바탕으로 한 줄 최종 보고서를 작성하세요.\n'
        '원문: {text}\n번역: {translated}\n요약: {summary}\n감정: {sentiment}'
    ) | openai_llm | parser
)

pipeline = (RunnablePassthrough.assign(translated=translate_chain).assign(summary=summarize_chain).assign(sentiment=sentiment_chain).assign(report=report_chain))
result = pipeline.invoke({"text" : "Python is a versatile language loved by developers worldwide."})

print("번역 : ", result['translated'])
print("요약 : ", result['summary'])
print("감정 : ", result['sentiment'])
print("보고서 : ", result['report'])

 

조건부 단계 삽입

# 조건부 단계 삽입
# 언어 감지 → 필요 시 번역 → 처리 패턴
detect_lang_chain = (
    ChatPromptTemplate.from_template(
        '다음 텍스트의 언어를 korean/english/other 중 하나로만 답하세요:\n{text}'
    ) | openai_llm | parser
)
 
def maybe_translate(inputs: dict) -> dict:
    lang = detect_lang_chain.invoke(inputs).strip().lower()
    if 'english' in lang or 'other' in lang:
        translated = translate_chain.invoke(inputs)
        return {**inputs, 'text': translated, 'was_translated': True}
    return {**inputs, 'was_translated': False}
 
smart_pipeline = (
    RunnableLambda(maybe_translate)
| RunnablePassthrough.assign(summary=summarize_chain)
| RunnablePassthrough.assign(sentiment=sentiment_chain)
)
r1 = smart_pipeline.invoke({'text': 'Python is great!'})  # 번역됨
r2 = smart_pipeline.invoke({'text': '파이썬은 최고야!'})   # 번역 생략
print(r1['was_translated'], r1['summary'][:50])
print(r2['was_translated'], r2['summary'][:50])

 

3. 대용량 문서 처리 (MapReduceChain)

 

Map과 Reduce라는 2단계로 나누어진다.

 

Map은 문서를 청크로 나눠 각각 추출한다 (요약, 추출) 또한 병렬 실행이 가능하다.

 

Reduce는 Map 한 것을 합쳐서 최종 답변을 생성한다.

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

map_chain = ChatPromptTemplate.from_messages([
    ("system", "다음 텍스트를 두문장으로 요약하세요"),
    ("human", "{chunk}")
]) | openai_llm | parser

reduce_chain = ChatPromptTemplate.from_messages([
    ("system", "다음은 긴 문서의 섹션별 요약입니다. 전체를 5문장으로 통합 요약하세요"),
    ("user", "{summaries}")
]) | openai_llm | parser

def map_reduce_summarize(file_path):
    loader = PyPDFLoader(file_path)
    splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
    chunks = splitter.split_documents(loader.load())
    
    print(f"총 청크 수 : {len(chunks)}")

    # ------- Map: 청크별 요약
    chunk_inputs = [{"chunk": c.page_content} for c in chunks]
    summaries = map_chain.batch(chunk_inputs, config={'max_concurrency': 5})
    print(f"{len(summaries)}개 요약 생성")

    # ------- Reduce : 요약 통합

    # 문서 결합
    combined = "\n\n".join(f"[섹션 {i+1}]" for i, s in enumerate(summaries))
    final = reduce_chain.invoke({'summaries': combined})
    return final


result = map_reduce_summarize("직무기술서/2026 상 삼성E&A 직무기술서.pdf")
print(result)