LCEL (LangChain Expression Language) 3

ML-AI llm
#genai-llm #langchain #lcel #runnable #chain #study-llm-framework

Chat History 관리

from langchain_community.chat_message_histories import SQLChatMessageHistory

def respond(session_id: str, human_message: str) -> str:
    chat_message_history = SQLChatMessageHistory(
        session_id = session_id,
        connection = "sqlite:///sqlite.db",
    )
    messages = chat_message_history.get_messages()

    ai_message = chain.invoke({
        "chat_history": messages,
        "human_message": human_message,
    })

    chat_message_history.add_user_message(human_message)
    chat_message_history.add_ai_message(ai_message)

    return ai_message

# 호출
from uuid import uuid4

session_id = uuid4().hex
output1 = respond(session_id=session_id, human_message="안녕하세요. 제 이름은 나뭇잎입니다.")
print(output1)

output2 = respond(session_id=session_id, human_message="내 이름이 뭘까?")
print(output2)

지원하는 데이터베이스

SQLite 말고 다른 데이터베이스의 통합도 제공합니다:

  • 인메모리
  • SQLAlchemy가 지원하는 각종 관계형 데이터베이스
  • Redis
  • DynamoDB
  • CosmosDB
  • Momento 등

메모리

대화 이력에 대한 고급 처리를 구현하고 싶은 경우 메모리를 사용합니다. 예를 들어 최근 k개의 대화 이력만 포함하고 싶은 경우. ConversationBufferWindowMemory나 ConversationSummaryMemory를 해당 컴포넌트로 사용할 수 있습니다.

LangServe

LangChain의 러너블을 간편하게 REST API로 만들기 위한 패키지입니다.

📚 추가로 알아볼 내용

1. 고급 Runnable 패턴들

RunnableGenerator (스트리밍 생성기):

from langchain_core.runnables import RunnableGenerator

def streaming_splitter(input: Iterator[str]) -> Iterator[str]:
    buffer = ""
    for chunk in input:
        buffer += chunk
        # 문장 단위로 분할하여 yield
        while '. ' in buffer:
            sentence, buffer = buffer.split('. ', 1)
            yield sentence + '. '

    if buffer:
        yield buffer

streaming_chain = model | RunnableGenerator(streaming_splitter)

RunnableRetry (재시도 로직):

from langchain_core.runnables import RunnableRetry

retry_chain = RunnableRetry(
    bound=model,
    max_attempt_number=3,
    wait_exponential_jitter=True,
    stop_after_delay=30
)

RunnableWithMessageHistory (대화 이력 관리):

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import SQLChatMessageHistory

def get_session_history(session_id: str):
    return SQLChatMessageHistory(session_id, "sqlite:///memory.db")

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

2. 복잡한 라우팅 전략

Multi-Modal Routing:

from typing import Literal

class RouteInput(BaseModel):
    query: str
    data_type: Literal["text", "image", "code"] = "text"

def route_by_type(input: RouteInput):
    if input.data_type == "image":
        return vision_chain
    elif input.data_type == "code":
        return code_analysis_chain
    else:
        return general_chain

routing_chain = RunnableLambda(route_by_type)

Semantic Routing:

from langchain_community.utils.math import cosine_similarity

def semantic_router(query: str, route_embeddings: dict):
    query_embedding = embeddings.embed_query(query)

    similarities = {}
    for route, embedding in route_embeddings.items():
        sim = cosine_similarity([query_embedding], [embedding])[0][0]
        similarities[route] = sim

    best_route = max(similarities, key=similarities.get)
    return route_chains[best_route]