LangChain & LangGraph

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

# 기본 RAG Chain
prompt = ChatPromptTemplate.from_template(
    "다음 컨텍스트를 참고하여 질문에 답하세요.\n\n"
    "컨텍스트: {context}\n\n"
    "질문: {question}"
)
model = ChatOpenAI(model="gpt-4o")

# LCEL 파이프 연산자로 체인 구성
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# 실행
result = chain.invoke("Delta Lake란 무엇인가요?")
print(result)

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_community.tools import TavilySearchResults

# 모델과 도구 정의
model = ChatOpenAI(model="gpt-4o")
tools = [TavilySearchResults(max_results=3)]

# ReAct Agent 생성 (단 2줄!)
agent = create_react_agent(model, tools)

# 실행
result = agent.invoke({
    "messages": [("user", "최신 Databricks 뉴스를 알려줘")]
})

for msg in result["messages"]:
    print(f"[{msg.type}] {msg.content[:200]}")

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

# 1. 상태 정의 -- 그래프 전체에서 공유
class AgentState(TypedDict):
    question: str
    category: str
    answer: str

# 2. 노드 정의 -- 각 단계의 로직
def classify(state: AgentState) -> AgentState:
    """질문을 분류하는 노드"""
    question = state["question"]
    if "가격" in question or "비용" in question:
        return {"category": "pricing"}
    elif "기술" in question or "아키텍처" in question:
        return {"category": "technical"}
    else:
        return {"category": "general"}

def handle_pricing(state: AgentState) -> AgentState:
    """가격 관련 질문 처리"""
    return {"answer": f"가격 팀에서 답변: {state['question']}에 대한 견적 안내입니다."}

def handle_technical(state: AgentState) -> AgentState:
    """기술 관련 질문 처리"""
    return {"answer": f"기술 팀에서 답변: {state['question']}에 대한 아키텍처 가이드입니다."}

def handle_general(state: AgentState) -> AgentState:
    """일반 질문 처리"""
    return {"answer": f"일반 답변: {state['question']}에 대한 정보입니다."}

# 3. 라우터 함수 -- 조건 분기 로직
def route_question(state: AgentState) -> Literal["pricing", "technical", "general"]:
    return state["category"]

# 4. 그래프 조립
graph = StateGraph(AgentState)

# 노드 추가
graph.add_node("classify", classify)
graph.add_node("pricing", handle_pricing)
graph.add_node("technical", handle_technical)
graph.add_node("general", handle_general)

# 엣지 연결
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_question, {
    "pricing": "pricing",
    "technical": "technical",
    "general": "general",
})
graph.add_edge("pricing", END)
graph.add_edge("technical", END)
graph.add_edge("general", END)

# 5. 컴파일 & 실행
app = graph.compile()
result = app.invoke({"question": "Databricks 가격이 어떻게 되나요?"})
print(result["answer"])
# 출력: "가격 팀에서 답변: Databricks 가격이 어떻게 되나요?에 대한 견적 안내입니다."

from langgraph.checkpoint.memory import MemorySaver

# 메모리 기반 Checkpoint (프로덕션에서는 PostgreSQL 등 사용)
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# thread_id로 대화 상태를 구분
config = {"configurable": {"thread_id": "user-123"}}

# 첫 번째 실행 -- 상태가 자동 저장됨
result1 = app.invoke({"question": "기술 아키텍처 설명해주세요"}, config)

# 이후 같은 thread_id로 실행하면 이전 상태를 이어받음
result2 = app.invoke({"question": "좀 더 상세하게"}, config)