LangChain 고급 패턴: LCEL과 LangGraph 완전 정복

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

prompt = ChatPromptTemplate.from_template("'{topic}'에 대해 간략히 설명해주세요.")
model = ChatOpenAI(model="gpt-4o")
output_parser = StrOutputParser()

# LCEL Pipe 연산자로 체인 구성
chain = prompt | model | output_parser

result = chain.invoke({"topic": "양자 컴퓨팅"})
print(result)

from langchain_core.runnables import RunnableParallel

# 방법 1: 딕셔너리 리터럴 (자동 변환)
chain = {
    "summary": prompt_summary | model | output_parser,
    "keywords": prompt_keywords | model | output_parser,
} | combine_results

# 방법 2: 명시적 RunnableParallel
parallel = RunnableParallel(
    summary=prompt_summary | model | output_parser,
    keywords=prompt_keywords | model | output_parser,
)
result = parallel.invoke({"text": "LangChain은 LLM 애플리케이션 프레임워크입니다."})
# result = {"summary": "...", "keywords": "..."}

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "코드" in x["topic"], code_chain),
    (lambda x: "수학" in x["topic"], math_chain),
    general_chain,  # default
)

# "코드"가 topic에 포함되면 code_chain 실행, "수학"이면 math_chain, 그 외에는 general_chain
result = branch.invoke({"topic": "코드 리팩토링 방법"})

from langchain_core.runnables import RunnablePassthrough

chain = RunnableParallel(
    context=retriever,                    # 검색 결과
    question=RunnablePassthrough(),       # 원본 질문 그대로 전달
) | prompt | model | output_parser

chain = RunnablePassthrough.assign(
    context=lambda x: retriever.invoke(x["question"])
)
# 입력: {"question": "LangChain이란?"}
# 출력: {"question": "LangChain이란?", "context": [Document(...), ...]}

from langchain_core.runnables import RunnableLambda

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 함수를 Runnable로 변환
format_runnable = RunnableLambda(format_docs)

# 체인에서 사용
chain = retriever | format_runnable | prompt | model | output_parser

from langchain_core.runnables import chain

@chain
def format_and_classify(input_dict):
    text = input_dict["text"]
    formatted = text.strip().lower()
    category = "technical" if "api" in formatted else "general"
    return {"formatted": formatted, "category": category}

result = chain.invoke({"topic": "머신러닝"})

results = chain.batch(
    [{"topic": "AI"}, {"topic": "블록체인"}, {"topic": "양자컴퓨팅"}],
    config={"max_concurrency": 2}
)
# results = ["AI는...", "블록체인은...", "양자컴퓨팅은..."]

for chunk in chain.stream({"topic": "딥러닝"}):
    print(chunk, end="", flush=True)

import asyncio

async def main():
    result = await chain.ainvoke({"topic": "강화학습"})
    print(result)

    async for chunk in chain.astream({"topic": "강화학습"}):
        print(chunk, end="", flush=True)

asyncio.run(main())

async for event in chain.astream_events(
    {"topic": "트랜스포머"},
    version="v2"
):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        # LLM 토큰 스트리밍
        print(event["data"]["chunk"].content, end="", flush=True)
    elif kind == "on_chain_start":
        print(f"\n--- 체인 시작: {event['name']} ---")
    elif kind == "on_chain_end":
        print(f"\n--- 체인 종료: {event['name']} ---")

async for log_patch in chain.astream_log(
    {"topic": "어텐션 메커니즘"},
    include_names=["ChatOpenAI"],
):
    for op in log_patch.ops:
        print(op)

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_community.vectorstores import FAISS

# 1. 벡터 스토어 및 Retriever 구성
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_texts(
    ["LangChain은 LLM 애플리케이션 프레임워크다.",
     "LCEL은 LangChain Expression Language의 약자이다.",
     "LangGraph는 상태 기반 에이전트 프레임워크다."],
    embeddings
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 2. 문서 포맷팅 함수
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 3. Prompt 템플릿
prompt = ChatPromptTemplate.from_template("""
다음 컨텍스트를 기반으로 질문에 답변하세요.

컨텍스트:
{context}

질문: {question}

답변:
""")

# 4. 모델 및 파서
model = ChatOpenAI(model="gpt-4o", temperature=0)
output_parser = StrOutputParser()

# 5. LCEL RAG 체인 구성
rag_chain = (
    RunnablePassthrough.assign(
        context=lambda x: format_docs(retriever.invoke(x["question"]))
    )
    | prompt
    | model
    | output_parser
)

# 6. 실행
result = rag_chain.invoke({"question": "LCEL이 무엇인가요?"})
print(result)

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add

# 1. 상태 정의
class AgentState(TypedDict):
    messages: Annotated[list[str], add]   # add reducer: 리스트를 누적
    current_step: str

# 2. 노드 함수 정의
def analyze_node(state: AgentState):
    return {
        "messages": ["분석 완료: 데이터를 처리했습니다."],
        "current_step": "analyze"
    }

def summarize_node(state: AgentState):
    return {
        "messages": ["요약 완료: 결과를 정리했습니다."],
        "current_step": "summarize"
    }

# 3. 그래프 구성
workflow = StateGraph(AgentState)
workflow.add_node("analyze", analyze_node)
workflow.add_node("summarize", summarize_node)

# 4. 엣지 정의
workflow.add_edge(START, "analyze")
workflow.add_edge("analyze", "summarize")
workflow.add_edge("summarize", END)

# 5. 컴파일 및 실행
graph = workflow.compile()
result = graph.invoke({
    "messages": ["시작"],
    "current_step": ""
})
print(result)
# {"messages": ["시작", "분석 완료: ...", "요약 완료: ..."], "current_step": "summarize"}

from langgraph.graph import StateGraph, START, END

class RouterState(TypedDict):
    query: str
    category: str
    result: str

def classify_node(state: RouterState):
    query = state["query"]
    if "코드" in query or "프로그래밍" in query:
        category = "code"
    elif "수학" in query or "계산" in query:
        category = "math"
    else:
        category = "general"
    return {"category": category}

def code_expert(state: RouterState):
    return {"result": f"코드 전문가가 답변합니다: {state['query']}"}

def math_expert(state: RouterState):
    return {"result": f"수학 전문가가 답변합니다: {state['query']}"}

def general_expert(state: RouterState):
    return {"result": f"일반 전문가가 답변합니다: {state['query']}"}

# 라우팅 함수
def route_query(state: RouterState) -> str:
    return state["category"]

# 그래프 구성
workflow = StateGraph(RouterState)
workflow.add_node("classify", classify_node)
workflow.add_node("code", code_expert)
workflow.add_node("math", math_expert)
workflow.add_node("general", general_expert)

workflow.add_edge(START, "classify")

# Conditional Edge: classify 노드 이후 라우팅
workflow.add_conditional_edges(
    "classify",           # 소스 노드
    route_query,          # 라우팅 함수
    {                     # 반환값 -> 대상 노드 매핑
        "code": "code",
        "math": "math",
        "general": "general",
    }
)

workflow.add_edge("code", END)
workflow.add_edge("math", END)
workflow.add_edge("general", END)

graph = workflow.compile()
result = graph.invoke({"query": "Python 코드 리팩토링 방법", "category": "", "result": ""})
print(result["result"])  # "코드 전문가가 답변합니다: ..."

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from typing import TypedDict, Annotated
from operator import add

class ConversationState(TypedDict):
    messages: Annotated[list[str], add]

def chat_node(state: ConversationState):
    last_message = state["messages"][-1]
    response = f"'{last_message}'에 대한 응답입니다."
    return {"messages": [response]}

workflow = StateGraph(ConversationState)
workflow.add_node("chat", chat_node)
workflow.add_edge(START, "chat")
workflow.add_edge("chat", END)

# Checkpointer 적용
checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

# Thread ID를 지정하여 실행
config = {"configurable": {"thread_id": "user-session-1"}}

# 첫 번째 대화
graph.invoke({"messages": ["안녕하세요"]}, config)

# 두 번째 대화 (같은 thread_id -- 이전 상태가 유지됨)
result = graph.invoke({"messages": ["LangGraph에 대해 알려주세요"]}, config)
print(result["messages"])
# ["안녕하세요", "'안녕하세요'에 대한 응답입니다.",
#  "LangGraph에 대해 알려주세요", "'LangGraph에 대해 알려주세요'에 대한 응답입니다."]

# 현재 상태 조회
snapshot = graph.get_state(config)
print(snapshot.values)   # 현재 상태 값
print(snapshot.next)     # 다음 실행될 노드

# 상태 히스토리 조회
for state in graph.get_state_history(config):
    print(state.config["configurable"]["checkpoint_id"], state.values)

# 특정 checkpoint에서 재실행
fork_config = {
    "configurable": {
        "thread_id": "user-session-1",
        "checkpoint_id": "0c62ca34-ac19-445d-bbb0-5b4984975b2a"
    }
}
graph.invoke(None, config=fork_config)

from langgraph.types import interrupt, Command

class ApprovalState(TypedDict):
    action: str
    approved: bool
    result: str

def plan_node(state: ApprovalState):
    return {"action": "중요 데이터베이스 마이그레이션 실행"}

def approval_node(state: ApprovalState):
    # 실행 일시 정지 -- 사람의 승인을 기다림
    response = interrupt({
        "question": "다음 작업을 승인하시겠습니까?",
        "action": state["action"]
    })
    return {"approved": response == "승인"}

def execute_node(state: ApprovalState):
    if state["approved"]:
        return {"result": "작업이 성공적으로 실행되었습니다."}
    return {"result": "작업이 취소되었습니다."}

workflow = StateGraph(ApprovalState)
workflow.add_node("plan", plan_node)
workflow.add_node("approval", approval_node)
workflow.add_node("execute", execute_node)

workflow.add_edge(START, "plan")
workflow.add_edge("plan", "approval")
workflow.add_edge("approval", "execute")
workflow.add_edge("execute", END)

checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "migration-1"}}

# 1단계: 그래프 실행 -> approval 노드에서 일시 정지
result = graph.invoke(
    {"action": "", "approved": False, "result": ""},
    config
)
# result에 __interrupt__ 정보가 포함됨

# 2단계: 사람이 검토 후 승인
final_result = graph.invoke(
    Command(resume="승인"),
    config
)
print(final_result["result"])  # "작업이 성공적으로 실행되었습니다."

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated, Literal
from operator import add

class MultiAgentState(TypedDict):
    query: str
    messages: Annotated[list[str], add]
    next_agent: str
    final_answer: str

# Supervisor: 어떤 에이전트에게 작업을 할당할지 결정
def supervisor_node(state: MultiAgentState):
    query = state["query"].lower()
    if "검색" in query or "찾아" in query:
        return {"next_agent": "researcher", "messages": ["Supervisor: 연구 에이전트에게 할당"]}
    elif "작성" in query or "써줘" in query:
        return {"next_agent": "writer", "messages": ["Supervisor: 작성 에이전트에게 할당"]}
    else:
        return {"next_agent": "analyst", "messages": ["Supervisor: 분석 에이전트에게 할당"]}

# 전문 에이전트들
def researcher_node(state: MultiAgentState):
    return {
        "messages": ["Researcher: 관련 자료를 검색하여 수집했습니다."],
        "final_answer": f"검색 결과: '{state['query']}'에 대한 자료를 찾았습니다."
    }

def writer_node(state: MultiAgentState):
    return {
        "messages": ["Writer: 요청에 맞게 콘텐츠를 작성했습니다."],
        "final_answer": f"작성 결과: '{state['query']}'에 대한 글을 작성했습니다."
    }

def analyst_node(state: MultiAgentState):
    return {
        "messages": ["Analyst: 데이터를 분석하여 인사이트를 도출했습니다."],
        "final_answer": f"분석 결과: '{state['query']}'에 대한 분석을 완료했습니다."
    }

def route_to_agent(state: MultiAgentState) -> str:
    return state["next_agent"]

# 그래프 구성
workflow = StateGraph(MultiAgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_node("analyst", analyst_node)

workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {
        "researcher": "researcher",
        "writer": "writer",
        "analyst": "analyst",
    }
)
workflow.add_edge("researcher", END)
workflow.add_edge("writer", END)
workflow.add_edge("analyst", END)

graph = workflow.compile()

result = graph.invoke({
    "query": "최신 AI 트렌드를 검색해줘",
    "messages": [],
    "next_agent": "",
    "final_answer": ""
})
print(result["final_answer"])

{
  "graphs": {
    "my_agent": "./agent.py:graph"
  },
  "dependencies": ["langchain", "langgraph", "langchain-openai"],
  "env": ".env"
}

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from typing import TypedDict, Annotated
from operator import add

# 상태 정의
class DocWorkflowState(TypedDict):
    document: str
    summary: str
    questions: Annotated[list[str], add]
    answers: Annotated[list[str], add]
    current_mode: str  # "summarize" | "qa" | "done"

# LLM 및 체인 준비
llm = ChatOpenAI(model="gpt-4o", temperature=0)

summary_prompt = ChatPromptTemplate.from_template(
    "다음 문서를 3줄로 요약하세요:\n\n{document}"
)
summary_chain = summary_prompt | llm | StrOutputParser()

qa_prompt = ChatPromptTemplate.from_template(
    "다음 문서를 참고하여 질문에 답변하세요.\n\n"
    "문서:\n{document}\n\n"
    "요약:\n{summary}\n\n"
    "질문: {question}\n\n답변:"
)
qa_chain = qa_prompt | llm | StrOutputParser()

# 노드 정의
def summarize_node(state: DocWorkflowState):
    summary = summary_chain.invoke({"document": state["document"]})
    return {"summary": summary, "current_mode": "qa"}

def ask_question_node(state: DocWorkflowState):
    # Human-in-the-Loop: 사용자에게 질문을 받음
    user_input = interrupt({
        "message": "문서에 대해 궁금한 점을 질문하세요. '종료'를 입력하면 끝납니다.",
        "summary": state["summary"]
    })
    if user_input == "종료":
        return {"current_mode": "done"}
    return {"questions": [user_input], "current_mode": "qa"}

def answer_node(state: DocWorkflowState):
    question = state["questions"][-1]
    answer = qa_chain.invoke({
        "document": state["document"],
        "summary": state["summary"],
        "question": question,
    })
    return {"answers": [answer]}

# 라우팅 함수
def route_after_answer(state: DocWorkflowState) -> str:
    return state["current_mode"]

def route_after_question(state: DocWorkflowState) -> str:
    if state["current_mode"] == "done":
        return "done"
    return "answer"

# 그래프 구성
workflow = StateGraph(DocWorkflowState)
workflow.add_node("summarize", summarize_node)
workflow.add_node("ask_question", ask_question_node)
workflow.add_node("answer", answer_node)

workflow.add_edge(START, "summarize")
workflow.add_edge("summarize", "ask_question")

workflow.add_conditional_edges(
    "ask_question",
    route_after_question,
    {"answer": "answer", "done": END}
)

# 답변 후 다시 질문 단계로 돌아감 (순환 구조)
workflow.add_edge("answer", "ask_question")

# 컴파일
checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

# 실행
config = {"configurable": {"thread_id": "doc-workflow-1"}}

# 1단계: 문서 전달 -> 요약 생성 -> 질문 대기 (interrupt)
result = graph.invoke(
    {
        "document": "LangGraph는 LangChain 팀이 개발한 상태 기반 에이전트 프레임워크로...",
        "summary": "",
        "questions": [],
        "answers": [],
        "current_mode": "summarize",
    },
    config
)

# 2단계: 사용자 질문 -> 답변 -> 다시 질문 대기
result = graph.invoke(Command(resume="LangGraph의 주요 장점은?"), config)

# 3단계: 추가 질문
result = graph.invoke(Command(resume="LCEL과의 차이점은?"), config)

# 4단계: 종료
result = graph.invoke(Command(resume="종료"), config)