[AI Agent] LangChain Expression Language(LCEL)

[AI Agent] LangChain Expression Language(LCEL)

1) 기본 개념

(1) LangChain Expression Language(LCEL)

• 가장 기본적이고 일반적인 사용 사례는 prompt 템플릿과 모델을 함께 연결하는 것
• chain = prompt | model | output_parser

In [ ]:

from dotenv import load_dotenv
load_dotenv("/home1/irteamsu/data_ssd2/users/sjson/projects_NHN/llm.mcp/agents-from-scratch/.env")

In [ ]:

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

In [ ]:

# 1. Prompt 정의 
# template 정의
template = "{country}의 수도는 어디인가요?"

# from_template 메소드를 이용하여 PromptTemplate 객체 생성
prompt_template = PromptTemplate.from_template(template)

# prompt 생성
prompt = prompt_template.format(country="대한민국")

prompt_template, prompt

In [ ]:

# 2. Model 정의 
model = ChatOpenAI(
    model="gpt-4o-mini",
    max_tokens=2048,
    temperature=0.1,
    api_key="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"
)

In [ ]:

# 3. Output Parser 정의
output_parser = StrOutputParser()

In [ ]:

chain = prompt_template | model | output_parser

chain

In [ ]:

chain.get_graph() # 체인의 실행 그래프를 반환

LCEL은 다음과 같은 기본 구성 요소를 제공

• Runnable: 모든 LCEL 컴포넌트의 기본 클래스입니다.
• Chain: 여러 Runnable을 순차적으로 실행합니다.
• RunnableMap: 여러 Runnable을 병렬로 실행합니다.
• RunnableSequence: Runnable의 시퀀스를 정의합니다.
• RunnableLambda: 사용자 정의 함수를 Runnable로 래핑합니다.

(2) Runnable 프로토콜 - 사용자 정의 체인을 가능한 쉽게 만들 수 있도록!

• 사용자 정의 체인을 가능한 쉽게 만들 수 있도록, Runnable 프로토콜을 구현
• Runnable 프로토콜은 대부분의 컴포넌트에 구현되어 있음
• 이는 표준 인터페이스로, 다음의 메소드들이 포함됨
  • 동기 메소드
    • stream: 응답의 청크를 스트리밍합니다.
    • invoke: 입력에 대해 체인을 호출합니다.
    • batch: 입력 목록에 대해 체인을 호출합니다.
  • 비동기 메소드
    • astream: 비동기적으로 응답의 청크를 스트리밍합니다.
    • ainvoke: 비동기적으로 입력에 대해 체인을 호출합니다.
    • abatch: 비동기적으로 입력 목록에 대해 체인을 호출합니다.
    • astream_log: 최종 응답뿐만 아니라 발생하는 중간 단계를 스트리밍합니다.
  • +) 개념 정리
    • 동기 : 작업이 순차적으로 실행되기에, 이전 작업이 끝나야 다음 작업 수행
    • 비동기 : 작업을 요청한 후, 기다리지 않고 다음 작업을 실행
• 위에서 배운 Chain은 이런 여러 Runnable을 순차적으로 실행

In [ ]:

for token in chain.stream({"country": "미국"}):
    # 스트림에서 받은 데이터의 내용을 출력합니다. 줄바꿈 있게 출력하고, 버퍼를 즉시 비웁니다.
    print(token, end="\n", flush=True)

In [ ]:

chain.invoke({"country": "미국"})

In [ ]:

chain.batch([{"country": "미국"}, {"country": "한국"}])

(3) Runnable의 종류

• 기본 제공
  • RunnableMap: 여러 Runnable을 병렬로 실행합니다.
  • RunnableSequence: Runnable의 시퀀스를 정의합니다.
  • RunnableLambda: 사용자 정의 함수를 Runnable로 래핑합니다.
• 그 외
  • RunnablePassthrough 인스턴스: invoke() 메서드를 통해 입력된 데이터를 그대로 반환하거나 추가 키를 더해 전달 가능
    • 데이터를 변경하지 않고 파이프라인의 다음 단계로 전달하는 데 사용될 수 있음
  • RunnableParellel 클래스: 병렬로 실행 가능한 작업을 정의
  • RunnableWithMessageHistory : 특정 유형의 작업(체인)에 메시지 기록을 추가
    • 휘발성 저장
    • 영구 저장

In [ ]:

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

runnable = RunnableParallel(
    # 전달된 입력을 그대로 반환하는 Runnable을 설정합니다.
    passed=RunnablePassthrough(),
    # 입력의 "num" 값에 3을 곱한 결과를 반환하는 Runnable을 설정합니다.
    extra=RunnablePassthrough.assign(mult=lambda x: x["num"] * 3),
    # 입력의 "num" 값에 1을 더한 결과를 반환하는 Runnable을 설정합니다.
    modified=lambda x: x["num"] + 1,
)

# {"num": 1}을 입력으로 Runnable을 실행합니다.
runnable.invoke({"num": 1})

In [ ]:

r = RunnablePassthrough.assign(mult=lambda x: x["num"] * 3)
r.invoke({"num": 1})

In [ ]:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

model = ChatOpenAI(
    model="gpt-4o-mini",
    max_tokens=2048,
    temperature=0.1,
    api_key="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"
)
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "당신은 {ability} 에 능숙한 어시스턴트입니다. 20자 이내로 응답하세요",
        ),
        # 대화 기록을 변수로 사용, history 가 MessageHistory 의 key 가 됨
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),  # 사용자 입력을 변수로 사용
    ]
)
runnable = prompt | model  # 프롬프트와 모델을 연결하여 runnable 객체 생성

In [ ]:

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}  # 세션 기록을 저장할 딕셔너리


# 세션 ID를 기반으로 세션 기록을 가져오는 함수
def get_session_history(session_ids: str) -> BaseChatMessageHistory:
    print(session_ids)
    if session_ids not in store:  # 세션 ID가 store에 없는 경우
        # 새로운 ChatMessageHistory 객체를 생성하여 store에 저장
        store[session_ids] = ChatMessageHistory()
    return store[session_ids]  # 해당 세션 ID에 대한 세션 기록 반환


with_message_history = (
    RunnableWithMessageHistory(  # RunnableWithMessageHistory 객체 생성
        runnable,  # 실행할 Runnable 객체
        get_session_history,  # 세션 기록을 가져오는 함수
        input_messages_key="input",  # 입력 메시지의 키
        history_messages_key="history",  # 기록 메시지의 키
    )
)

In [ ]:

with_message_history.invoke(
    # 수학 관련 질문 "코사인의 의미는 무엇인가요?"를 입력으로 전달합니다.
    {"ability": "math", "input": "What does cosine mean?"},
    # 설정 정보로 세션 ID "abc123"을 전달합니다.
    config={"configurable": {"session_id": "abc123"}},
)

In [ ]:

with_message_history.invoke(
    # 능력과 입력을 설정합니다.
    {"ability": "math", "input": "이전의 내용을 한글로 답변해 주세요."},
    # 설정 옵션을 지정합니다.
    config={"configurable": {"session_id": "abc123"}},
)

In [ ]:

# 새로운 session_id로 인해 이전 대화 내용을 기억하지 않습니다.
with_message_history.invoke(
    # 수학 능력과 입력 메시지를 전달합니다.
    {"ability": "math", "input": "이전의 내용을 한글로 답변해 주세요"},
    # 새로운 session_id를 설정합니다.
    config={"configurable": {"session_id": "def234"}},
)

(4) RunnableSequence : Runnable instance 묶기

In [ ]:

from langchain_core.runnables import RunnableLambda, RunnableSequence

def add_one(x: int) -> int:
    return x + 1

def mul_two(x: int) -> int:
    return x * 2

runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence1 = runnable_1 | runnable_2
sequence1.invoke(1)

In [ ]:

# Or equivalently:
sequence2 = RunnableSequence(first=runnable_1, last=runnable_2)
sequence2.invoke(1)

(5) Runtime Arguments Binding

• 때로는 Runnable 시퀀스 내에서 Runnable을 호출할 때, 이전 Runnable의 출력이나 사용자 입력에 포함되지 않은 상수 인자를 전달해야 할 경우가 있음
  • 이때 Runnable.bind()를 사용하면 인자를 쉽게 전달할 수 있음

In [ ]:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            # 대수 기호를 사용하여 다음 방정식을 작성한 다음 풀이하세요.
            "Write out the following equation using algebraic symbols then solve it. "
            "Use the format\n\nEQUATION:...\nSOLUTION:...\n\n",
        ),
        (
            "human",
            "{equation_statement}",  # 사용자가 입력한 방정식 문장을 변수로 받습니다.
        ),
    ]
)

model = ChatOpenAI(
    model="gpt-4o-mini",
    max_tokens=2048,
    temperature=0.1,
    api_key="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"
)

# 방정식 문장을 입력받아 프롬프트에 전달하고, 모델에서 생성된 결과를 문자열로 파싱합니다.
runnable = (
    {"equation_statement": RunnablePassthrough()} | prompt | model | StrOutputParser()
)

# 예시 방정식 문장을 입력하여 결과를 출력합니다.
print(runnable.invoke("x raised to the third plus seven equals 12"))

In [ ]:

runnable = (
    # 실행 가능한 패스스루 객체를 생성하여 "equation_statement" 키에 할당합니다.
    {"equation_statement": RunnablePassthrough()}
    | prompt  # 프롬프트를 파이프라인에 추가합니다.
    | model.bind(
        stop="SOLUTION"
    )  # 모델을 바인딩하고 "SOLUTION" 토큰에서 생성을 중지하도록 설정합니다.
    | StrOutputParser()  # 문자열 출력 파서를 파이프라인에 추가합니다.
)
# "x raised to the third plus seven equals 12"라는 입력으로 파이프라인을 실행하고 결과를 출력합니다.
print(runnable.invoke("x raised to the third plus seven equals 12"))

• binding 의 특히 유용한 활용 방법 중 하나는 호환되는 OpenAI 모델에 OpenAI Functions 를 연결하는 것임

In [ ]:

openai_function = {
    "name": "solver",  # 함수의 이름
    # 함수의 설명: 방정식을 수립하고 해결합니다.
    "description": "Formulates and solves an equation",
    "parameters": {  # 함수의 매개변수
        "type": "object",  # 매개변수의 타입: 객체
        "properties": {  # 매개변수의 속성
            "equation": {  # 방정식 속성
                "type": "string",  # 방정식의 타입: 문자열
                "description": "The algebraic expression of the equation",  # 방정식의 대수식 표현
            },
            "solution": {  # 해답 속성
                "type": "string",  # 해답의 타입: 문자열
                "description": "The solution to the equation",  # 방정식의 해답
            },
        },
        "required": ["equation", "solution"],  # 필수 매개변수: 방정식과 해답
    },
}

In [ ]:

# 다음 방정식을 대수 기호를 사용하여 작성한 다음 해결하세요.
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "Write out the following equation using algebraic symbols then solve it.",
        ),
        ("human", "{equation_statement}"),
    ]
)
model = ChatOpenAI(model="gpt-4o-mini", temperature=0, api_key="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").bind(
    function_call={"name": "solver"},  # openai_function schema 를 바인딩합니다.
    functions=[openai_function],
)
runnable = {"equation_statement": RunnablePassthrough()} | prompt | model
# x의 세제곱에 7을 더하면 12와 같다
runnable.invoke("x raised to the third plus seven equals 12")

2) Messages

메시지 타입설명

SystemMessage	시스템 지시나 프롬프트를 정의할 때 사용됩니다. LLM의 행동이나 문체를 제어할 수 있습니다. 예: "You are a helpful assistant."
HumanMessage	사람이 보낸 메시지를 나타냅니다. 예: "What's the weather like today?"
AIMessage	LLM이 생성한 응답을 나타냅니다. ***내부적으로 role="assistant"역할** . 예: "The weather is sunny today."
FunctionMessage	함수 호출 후 반환된 응답 메시지입니다. tool calling 이후에 LLM이 응답을 계속 이어나갈 수 있도록 합니다.
ToolMessage	LangChain에서 tool 결과를 전달할 때 사용하는 메시지입니다. 예를 들어 외부 API, DB 질의, 내부 계산 등 다양한 도구의 결과를 LLM에 전달 FunctionMessage와 유사하지만 더 범용적입니다.
ChatMessage(role=..., content=...)	커스텀 역할(role)을 갖는 일반 메시지입니다. tool_call_id와 연결되어 있어 어떤 호출에 대한 응답인지 명확히 구분 가능 예: role="user" 혹은 "moderator" 등 임의 지정 가능

• Tool Calling하는 경우, 핵심 흐름
  1. 사용자 질문 → HumanMessage
  2. LLM 응답 (Tool 호출 지시 포함) → AIMessage(tool_calls=[...])
  3. Tool 실행 결과 응답 → ToolMessage(tool_call_id, content)
  4. LLM이 최종 응답 → AIMessage(content="...")

In [ ]:

from langchain.schema import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content="Tell me a joke."),
    AIMessage(content="Why don't scientists trust atoms? Because they make up everything!")
]

3) Agent란?

• 에이전트는 사전에 정의된 규칙이나 명시적인 프로그래밍 없이도 스스로 결정을 내리고 행동
  • 구성
    • AI Model
    • Capabilities and Tools
• LangChain에서 에이전트는 다음과 같은 구성요소로 이루어져 있습니다:
  • Agent: 의사 결정을 담당하는 핵심 컴포넌트입니다.
  • Tools: 에이전트가 사용할 수 있는 기능들의 집합입니다.
  • Toolkits: 관련된 도구들의 그룹입니다.
  • AgentExecutor: 에이전트의 실행을 관리하는 컴포넌트입니다.