LangChain

LLM 을 갖고 뭔갈 만들건데 (ex. RAG, AGent), 그걸 만드는 수단 (다른 수단도 있음 ex. Langraph)

부품(prompt, Model Parser) -> LCEL 체인 (파이프로 연결) -> 메모리 (대화 기억) -> 도구 (외부 기능) -> 에이전트 (스스로 도구 선택)

LLM (Large Language Model)

LLM : 언어 생성 - open ai (vs 언어 이해: google)
ㄴ 우리가 통상 ai를 활용해서 쓰는 기능은 LLM 자체 기능은 아니고 Tool을 활용한거임 (LLM + Tool = Agent)

참고 사이트

• https://bbycroft.net/llm
• https://poloclub.github.io/transformer-explainer/
• https://gpt-tokenizer.dev/

• LLM은 다음 토큰을 확률로 예측한다 : 정해진 답을 저장해 두고 꺼내는 것이 아니라, 매 순간 가장 그럴듯한 다음 토큰을 고른다.

• 토큰과 컨텍스트 윈도우 ** 토큰: 모델이 텍스트를 처리하는 최소 단위
  -> 입력, 출력의 토큰 수에 비례해 시간과 비용이 든다 ** 컨텍스트 윈도우: 한번에 입력할 수 있는 토큰 상한선 -> 넘는 분량은 넣을 수 없으므로, 긴 문서를 통째로 제공하는 데엔 한계

한계

• 학습 컷오프 : 학습 시점 이후의 사실, 그리고 학습에 포함되지 않은 사내•비공개 지식은 알지 못한다. --> But 요즘 보완, Reasoning Model : 추론 모델 (스스로에게 질문 다시) --> '사실'은 아니지만 논리적인 예상을 함
• 환각 (Halluciation) : 모르는 것을 모른다고 하지 않고, 사실처럼 보이는 틀린 내용 생성 --> But 요즘 보완, GR : LLM 뒤 Input, Output에 가드레일을 설치 ㄴ 이렇게 LLM 더 잘 활용하고 관리하는 기법 : 'LLM Ops' (+ 프롬프트 엔지니어링, RAG, 파인 튜닝)

Prompt Engineering

: LLM 개발자가 아닌 사용자로써, LLM 성능을 최대한 잘 할 수 있는 방법은 프롬포트가 유일 ('AI 활용은 Prompt 싸움이다')

LLM Application 구성 요소

1. 모델 : 텍스트를 생성하는 핵심 엔진
2. 프롬포트 : 입력 설계 - 지시•맥락•예시
3. 도구(tool) : 검색, 계산 등 모델이 직접 못하는 일
4. 메모리: 이전 대화를 이어가는 상태
5. 검색: 외부 지식을 찾아 입력에 결합

LangChain

LLM 앱의 구성요소(프롬포트•모델•파서•도구)를 표준 부품으로 만들어 연결(|)하는 프레임워크 --> 각 부품이 같은 방식으로 입출력하므로, 자유롭게 갈아끼우고 이어 붙일 수 있다.

• 사용 이유

1. LLM 제공 회사의 고유 SDK에 종속되지 않기 위해
2. LLM App을 더 쉽게 만들기 위해

+) Lang 시리즈

• DeepAgent: 사용성 굉장히 편함
• Langchain : 기본 뼈대
• Langraph: 자유도 높음, 요즘 Agent 개발에 필수 (Langraph 각 노드가 LangChain 으로 이루어져있음)
• LangSmith(Agent Ops)

ㄴ 보통 Agent 개발 = Langraph 기반 + Langchain ㄴ But 속도에 문제가 있다면, SDK를 쓸 수 밖에 없는 상황이 올 수도 있다. (자유도는 가장 높음)

• 패키지 구성 (생태계) ** langchain : 체인, 에이전트 등 조립 가능 ** langchain-openai : OpenAI 모델 연결(ChatOpenAI) ** langchain-core : 프롬포트, 파서 등 기본 부품

PromptTemplate

: 자주 쓰는 프롬포트 형식을 활용 ('템플릿화') --> 변수 슬롯을 포함하여 입력 수정 => 같은 프롬포트에 입력만 바뀔 때, 여러 곳에서 재사용 할 때

ChatPromptTemplate

= system-human 메세지 + {변수} 슬롯 ; 역할(system)은 고정하고, 바뀌는 입력만 {변수}로 비워두어 system-human 메세지 구조가 그대로 템플릿이 됨

• system prompt : 역할•규칙 (고정) ex. LLM에 role 부여, 인풋으로 받으면 안되는 금지어, LLM을 시스템적으로 구동시킬 때 필요한 규칙

참고 : https://platform.claude.com/docs/ko/release-notes/system-prompts => Agent 만들 때도 매우 중요

• human(user) prompt: 사용자가 입력하는 프롬포트

from langchain_core.prompts import ChatPromptTemplate

# 역할(system)은 고정, 바뀌는 입력만 {중괄호} 슬롯으로 비워 둔다
prompt = ChatPromptTemplate.from_messages([
    ("system", "너는 번역가다. 입력 문장을 {language}로 번역해라."),
    ("human", "{text}"),
])

# 슬롯에 값을 채워 메시지를 완성한다 (아직 모델 호출 전 단계)
prompt.invoke({"text": "좋은 아침입니다", "language": "영어"})

# .partial() — 일부 슬롯을 미리 고정한 새 프롬프트를 만든다
to_english = prompt.partial(language="영어")
to_english.invoke({"text": "반갑습니다"})   # language는 '영어'로 이미 고정

FewShotChatMessagePromptTemplate

예시를 몇 개 보여 주기 ; 원하는 형식이나 스타일을 예시 몇 개로 보여 주면, 모델이 그 패턴을 따라 답합니다. 규칙을 길게 설명하기 어려울 때 특히 유용합니다.

shot: 예시 개수 -> zero-shot, 1-shot, 2-shot

from langchain_core.prompts import FewShotChatMessagePromptTemplate

examples = [
    {"input": "행복", "output": "행복 ↔ 불행"},
    {"input": "성공", "output": "성공 ↔ 실패"},
]
example_prompt = ChatPromptTemplate.from_messages([("human", "{input}"), ("ai", "{output}")])
few_shot = FewShotChatMessagePromptTemplate(examples=examples, example_prompt=example_prompt)

final_prompt = ChatPromptTemplate.from_messages([
    ("system", "너는 단어의 반대말을 '단어 ↔ 반대말' 형식으로 답한다."),
    few_shot,
    ("human", "{input}"),
])
(final_prompt | model).invoke({"input": "사랑"}).content

--> 잘 안쓰고 ChatPromptTemplate Context에 system prompt로 넣어줌

+) invoke() : 내가 너한테 데이터를 줄 테니까, 네가 맡은 역할을 수행해서 다음 단계로 넘길 수 있는 결과물을 만들어내! --> 모든 컴포넌트(Prompt, LLM, OutputParser 등)를 '동일한 방식으로 실행 가능한 체(Runnable)'으로 통일

# 레고 블록 조립하듯이 체인을 만듦
chain = prompt | model | output_parser

# 체인 전체를 실행!
chain.invoke({"text": "좋은 아침입니다", "language": "영어"})

이렇게 chain.invoke()를 딱 한 번만 호출하면, LangChain 내부에서는 다음과 같은 연쇄 반응이 일어난다.

1. 사용자가 입력한 {"text": ...} 딕셔너리가 prompt.invoke()로 들어감.
2. 프롬프트가 완성된 메시지를 생성해서 model.invoke()로 넘겨줌.
3. 모델이 생성한 AI 답변 Raw 데이터가 output_parser.invoke()로 들어가서 깔끔한 텍스트로 가공됨.

Model

모델도 부품, 한 줄로 호출 가능하다

Temperature

: 답의 다양성을 정한다 (0 : 일관 --> high : 자유로움 --> 확률 분포가 점차 균일해짐)

# temperature 0 vs 1.3 — 같은 질문에 답이 얼마나 달라지나
cold = ChatOpenAI(model="gpt-4o-mini", temperature=0)
hot = ChatOpenAI(model="gpt-4o-mini", temperature=1.3)
q = "가을을 한 문장으로 표현해줘."
print("[0.0]", cold.invoke(q).content)
print("[1.3]", hot.invoke(q).content)

[0.0] 가을은 나뭇잎이 황금빛으로 물들고, 서늘한 바람이 불어오는 따스한 추억의 계절이다.
[1.3] 가을은 오색 단풍이 우아하게 나뭇잎을 물들이고, 서늘한 바람이 감성을 자극하며 수확의 기쁨이 가득한 계절이다.

OutputParser

: 모델 출력은 문자열이나, 그대로 쓰려면 json.loads로 직접 파싱하고 오류까지 처리해야 한다.

직접 파싱 : text -> json.loads (try/except) LangChain : 체인 끝에 | Parser

• StrOutputParser : AI Message에서 순수 문자열만 꺼낸다.
• JsonOutputParser : JSON 형태 출력을 dict 로 바로 변환한다.
• model.with_structured_output(Review) : Pydantic 모델 객체로 직접 받는다. ** model 답변 자체에 with_structured_output()

from pydantic import BaseModel

# 받고 싶은 형태를 클래스로 정의 (필드 + 타입)
class Person(BaseModel): # 상속 
    name: str  # Python의 type 검사
    job: str

# with_structured_output → 검증된 Person 객체로 직접 받는다
structured_chain = info_prompt | model.with_structured_output(Person)
person = structured_chain.invoke({"text": "안녕하세요, 저는 김민수이고 데이터 분석가로 일합니다."})
print(person)
print("이름만 꺼내기:", person.name)

"""
name='김민수' job='데이터 분석가'
이름만 꺼내기: 김민수
"""

• Parser로 분리: PydanticOutputParser

from langchain_core.output_parsers import PydanticOutputParser

parser = PydanticOutputParser(pydantic_object=Person)   # 섹션 3의 Person(name, job) 재사용

pyd_prompt = ChatPromptTemplate.from_messages([
    ("system", "문장에서 정보를 뽑아라.\n{format_instructions}"),
    ("human", "{text}"),
]).partial(format_instructions=parser.get_format_instructions())

chain = pyd_prompt | model | parser
chain.invoke({"text": "저는 박지수이고 디자이너입니다."})

"""
Person(name='박지수', job='디자이너')
"""

단순 텍스트면 충분 → StrOutputParser 가벼운 dict면 OK → JsonOutputParser 타입까지 보장된 객체(가장 권장) → with_structured_output 지시문을 직접 넣어 객체로 파싱 → PydanticOutputParser

Chain

Chain 실행 방법

• invoke() : 한 건씩
• batch() : 여러 건 한꺼번에
• stream() : 토큰 실시간

Chain 입력 방법

• 문자열: 단일 입력
• dict : 슬롯 여럿
• RunnablePassthrough : 입력 그대로 전달 ** assign(...) : 기존 유지 + 새 키 추가

from langchain_core.runnables import RunnablePassthrough

# 입력(dict)을 그대로 유지하면서 새 키(length)를 추가한다
add_len = RunnablePassthrough.assign(length=lambda x: len(x["text"]))
add_len.invoke({"text": "안녕하세요"})

# {'text': '안녕하세요', 'length': 5}

• RunnableParallel : 여러 소스로 동시 구성

from langchain_core.runnables import RunnableParallel

summary_prompt = ChatPromptTemplate.from_messages([
    ("system", "너는 한 줄 요약가다. 핵심만 한 문장으로 요약해라."),
    ("human", "{text}"),
])

summary = summary_prompt | model | StrOutputParser()
keywords = (ChatPromptTemplate.from_messages([("human", "다음에서 키워드 3개만 콤마로: {text}")])
            | model | StrOutputParser())

# 두 체인을 동시에 실행해 결과를 dict로 합친다
both = RunnableParallel(summary=summary, keywords=keywords)
both.invoke({"text": "LangChain은 부품을 표준화해 LLM 앱을 쉽게 만들게 해 준다."})

# 두 체인을 동시에 실행해 결과를 dict로 합친다
both2 = RunnableParallel(요약=summary, 키워드=keywords)
both2.invoke({"text": "LangChain은 부품을 표준화해 LLM 앱을 쉽게 만들게 해 준다."})

"""
{'요약': 'LangChain은 LLM 앱 개발을 용이하게 하는 부품 표준화 도구이다.',
 '키워드': 'LangChain, 부품, LLM 앱'}
"""

• RunnableLambda : 함수로 가공해 넣기

from langchain_core.runnables import RunnableLambda

summary_prompt = ChatPromptTemplate.from_messages([
    ("system", "너는 한 줄 요약가다. 핵심만 한 문장으로 요약해라."),
    ("human", "{text}"),
])

summary = summary_prompt | model | StrOutputParser()

# 평범한 함수를 체인 부품으로 끼운다
annotate = RunnableLambda(lambda s: f"[{len(s)}자] {s}")
(summary | annotate).invoke({"text": "오늘은 날씨가 맑고 따뜻합니다."})

""" 
[17자] 오늘은 맑고 따뜻한 날씨입니다.
"""

+) 'Runnable' : Chain의 한 부품으로 포함할 수 있음

Chain 신뢰도

• .with_retry 일시 오류 시 자동 재시도
• .with_fallbacks 실패하면 대체 부품으로

Memory (대화 기억)

체인은 매 호출이 독립적 : 체인과 모델은 상태를 저장하지 않는다 <-> Stateless (vs Statefull : State Storage O) => '기억'을 위해서는 이전 대화를 입력에 함께 전달 ('Context')

ㄴ HOW? MessagesPlaceholder (이력 슬롯) 에 메세지를 쌓아넣는다

from langchain_core.prompts import MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
("system", "너는 친절한 비서야"),
MessagesPlaceholder("history"), # 이력 슬롯
("human", "{input}"),
])

# 실행 시 슬롯에 이력이 펼쳐진다
prompt.invoke({"input": "내 이름은?",
"history": history.messages})
# → system + (이력 전체) + human

• 수동: 이력을 직접 넣기

from langchain_core.chat_history import (
InMemoryChatMessageHistory)

history = InMemoryChatMessageHistory()
history.add_user_message("내 이름은 Kim") 

# 프롬프트의 history 슬롯에 직접 전달
chain.invoke({"input": "내 이름은?",
"history": history.messages}

• 자동: RunnableWithMessageHistory

from langchain_core.runnables.history import (
RunnableWithMessageHistory)

chat = RunnableWithMessageHistory(
chain, get_session_history,
history_messages_key="history")

chat.invoke({"input": "내 이름은 Kim"},
{"configurable": {"session_id": "a"}})

+) 응용

from langchain_core.prompts import MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# 이력 슬롯(history)이 있는 프롬프트
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "너는 친절한 비서다."),
    MessagesPlaceholder("history"),
    ("human", "{input}"),
])
chat_chain = chat_prompt | model | StrOutputParser()


#session_id = 카톡방들을 저장하는 저장소
# session_id별 이력 저장소
store = {}
def get_session_history(session_id):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory() #카톡방 자체
    return store[session_id]

# 체인을 감싸면 이력이 자동으로 쌓이고 들어간다
chat = RunnableWithMessageHistory(
    chat_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

chat.invoke({"input": "안녕"}, config={"configurable": {"session_id": "방1"}})

Tools (도구 연결)

• LLM : 최신 정보 조회, 계산, 외부 시스템 호출을 스스로 하지 못한다. --> Prompt에 tool, skill 목록 다 같이 전달

1. @tool로 도구 정의

from langchain.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """두 정수 a와 b를 곱한다."""
    return a * b

• @tool : 함수를 LLM이 Tool로 인식하도록 해주는 decorator
• Parameters : 자료형 함께 표시 필수
• docstring(도구 설명) : LLM이 읽어 도구 내용(기능)을 파악하므로 가장 중요

※ 도구가 많아지면 LLM이 기억을 못함 (권장 20~25개) --> 해결 방법: Multi-Agent (Context Engineering)

2. bind_tools로 모델 연결

# 모델에 도구 연결 → 모델은 '어떤 도구를 어떤 인자로' 부를지만 정한다
model_with_tools = model.bind_tools([multiply])
resp = model_with_tools.invoke("12 곱하기 7은 얼마야?")
resp.tool_calls

tabily: https://www.tavily.com/

3. 도구를 실행 LLM은 '어떤 도구를 부를지(tool_calls)' 만 정함 (모델이 도구 선택). 실제로 답을 얻으려면 1) 우리가 직접 코드를 실행하거나, 2) Agent 가 실행시켜줘야 함

from langchain_core.messages import ToolMessage

messages = [HumanMessage("12 곱하기 7은 얼마야?")]
ai = model_with_tools.invoke(messages)        # ① 모델이 도구 선택
messages.append(ai)

for call in ai.tool_calls:                     # ② 우리가 직접 실행
    result = multiply.invoke(call["args"])
    messages.append(ToolMessage(content=str(result), tool_call_id=call["id"]))   # ③ 결과 회신

final = model_with_tools.invoke(messages)     # ④ 모델이 최종 답
print(final.content)

toolcall -> toolmessage -> aimessage

단일 Agent (create_agent)

• ReAct Pattern ; LLM이 스스로 "생각 → 도구 호출 → 관찰"을 충분해질 때까지 반복한다. (vs Chain: 정해진 순서로만 흐름) --> 무엇을 몇 번 호출할지 미리 정하기 어려울 때 사용

=> create_agent: 모델이 스스로 도구를 골라 반복 실행하는 단일 에이전트 생성

from langchain.agents import create_agent

# 모델 + 도구 + 행동 규칙(system) → 단일 에이전트
agent = create_agent(
    model,
    tools=[multiply],
    system_prompt="너는 계산을 도와주는 도우미다. 계산이 필요하면 도구를 써라.",
)

# 에이전트는 스스로 multiply 도구를 골라 쓴다
result = agent.invoke({"messages": [{"role": "user", "content": "23 곱하기 19는 얼마야?"}]})
result["messages"][-1].content

• 전체 메시지 흐름 : 사용자 -> 도구 호출 -> 도구 결과 -> 최종 답

for m in r["messages"]:
    print(type(m).__name__, "|", m.content or getattr(m, "tool_calls", ""))

"""
HumanMessage | 6 곱하기 9는?
AIMessage | [{'name': 'multiply', 'args': {'a': 6, 'b': 9}, 'id': 'call_4mFHqTzNhkcBAG4CpdUNqUQ9', 'type': 'tool_call'}]
ToolMessage | 54
AIMessage | 6 곱하기 9는 54입니다.
"""

• checkpointer: 에이전트에 기억 붙이기 -> thread_id별로 대화를 기억

from langgraph.checkpoint.memory import InMemorySaver

# 체크포인터를 붙이면 thread_id별로 대화를 기억한다
mem_agent = create_agent(
    model,
    tools=[multiply],
    system_prompt="너는 계산 도우미다.",
    checkpointer=InMemorySaver(),
)

cfg = {"configurable": {"thread_id": "t1"}}
print(mem_agent.invoke({"messages": [{"role": "user", "content": "5 곱하기 6을 계산해줘."}]}, cfg)["messages"][-1].content)
print("---")
print(mem_agent.invoke({"messages": [{"role": "user", "content": "방금 결과에 2를 곱하면?"}]}, cfg)["messages"][-1].content)

InMemorySaver() 외에도 SqliteSaver 등 다양하게 존재

• ReAct Process : ex. SQL 에이전트

1. Agent 정의

from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import SQLDatabaseToolkit

db = SQLDatabase.from_uri("sqlite:///company.db")
sql_tools = SQLDatabaseToolkit(db=db, llm=model).get_tools()   # 스키마 조회·쿼리 실행 등
print("SQL 도구:", [t.name for t in sql_tools])

sql_agent = create_agent(
    model, tools=sql_tools,
    system_prompt="너는 SQL 분석가다. 질문을 employees 테이블 SQL로 바꿔 조회하고 결과를 한국어로 답해라.",
)
print(sql_agent.invoke({"messages": [{"role": "user", "content": "개발 부서 평균 연봉은 얼마야?"}]})["messages"][-1].content)

2. Agent 실행 (사용자가 질문)

# 질문을 던지고, 에이전트가 만든 SQL과 실행 과정을 .stream으로 들여다본다.
for step in sql_agent.stream(
    {"messages": [{"role": "user", "content": "2020년 이후 입사자 중 연봉이 가장 높은 사람은?"}]},
    stream_mode="values",
):
    step["messages"][-1].pretty_print()   # 테이블/스키마 조회 -> SQL 생성·실행 -> 한국어 답변

================================ Human Message =================================

2020년 이후 입사자 중 연봉이 가장 높은 사람은?
================================== Ai Message ==================================
Tool Calls:
  sql_db_list_tables (call_vAGPikQUeTpSCNnLeglwd0oH)
 Call ID: call_vAGPikQUeTpSCNnLeglwd0oH
  Args:
================================= Tool Message =================================
Name: sql_db_list_tables

employees
================================== Ai Message ==================================
Tool Calls:
  sql_db_query_checker (call_j7ApodwSzsjdYvWe6ps8ZqGD)
 Call ID: call_j7ApodwSzsjdYvWe6ps8ZqGD
  Args:
    query: SELECT * FROM employees WHERE hire_date > '2020-01-01' ORDER BY salary DESC LIMIT 1;
================================= Tool Message =================================
Name: sql_db_query_checker


SELECT * FROM employees WHERE hire_date > '2020-01-01' ORDER BY salary DESC LIMIT 1;
================================== Ai Message ==================================
Tool Calls:
  sql_db_query (call_SHEVz4WQ0VByUha0A6KKZXj4)
 Call ID: call_SHEVz4WQ0VByUha0A6KKZXj4
  Args:
    query: SELECT * FROM employees WHERE hire_date > '2020-01-01' ORDER BY salary DESC LIMIT 1;
================================= Tool Message =================================
Name: sql_db_query

Error: (sqlite3.OperationalError) no such column: hire_date
[SQL: SELECT * FROM employees WHERE hire_date > '2020-01-01' ORDER BY salary DESC LIMIT 1;]
(Background on this error at: https://sqlalche.me/e/20/e3q8)
================================== Ai Message ==================================
Tool Calls:
  sql_db_schema (call_deZuwuCDe5hX52ETwDvJbJ8m)
 Call ID: call_deZuwuCDe5hX52ETwDvJbJ8m
  Args:
    table_names: employees
================================= Tool Message =================================
Name: sql_db_schema


CREATE TABLE employees (
	id INTEGER, 
	name TEXT, 
	dept TEXT, 
	salary INTEGER, 
	hire_year INTEGER, 
	PRIMARY KEY (id)
)

/*
3 rows from employees table:
id	name	dept	salary	hire_year
1	김민수	개발	6500	2021
2	이서연	개발	7200	2019
3	박지훈	영업	5400	2022
*/
================================== Ai Message ==================================
Tool Calls:
  sql_db_query_checker (call_hi2gq8gpTShQ6iYaeniYBCXX)
 Call ID: call_hi2gq8gpTShQ6iYaeniYBCXX
  Args:
    query: SELECT * FROM employees WHERE hire_year > 2020 ORDER BY salary DESC LIMIT 1;
================================= Tool Message =================================
Name: sql_db_query_checker

SELECT * FROM employees WHERE hire_year > 2020 ORDER BY salary DESC LIMIT 1;
================================== Ai Message ==================================
Tool Calls:
  sql_db_query (call_AW07ELCzB5QUWEX3DPeCXInp)
 Call ID: call_AW07ELCzB5QUWEX3DPeCXInp
  Args:
    query: SELECT * FROM employees WHERE hire_year > 2020 ORDER BY salary DESC LIMIT 1;
================================= Tool Message =================================
Name: sql_db_query

[(1, '김민수', '개발', 6500, 2021)]
================================== Ai Message ==================================

2020년 이후에 입사한 사람 중 연봉이 가장 높은 사람은 김민수입니다. 그의 연봉은 6500입니다.

=> create_agent 워크플로보다 더 복잡한 것을 만들고 싶을 때, Langgraph 필요 : 흐름을 그래프 형식으로 직접 설계할 수 있음