LangChain vs LangGraph: 완벽 비교 가이드

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

# 프롬프트 템플릿 정의
prompt = ChatPromptTemplate.from_template(
    "다음 질문에 답하세요: {question}"
)

# 모델 생성
model = ChatOpenAI(model="gpt-4")

# 체인 구축
chain = prompt | model | StrOutputParser()

# 체인 호출
result = chain.invoke({"question": "LangChain이란 무엇인가요?"})
print(result)

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict

# 상태 정의
class State(TypedDict):
    question: str
    answer: str

# 노드 정의
def generate_answer(state: State) -> State:
    model = ChatOpenAI(model="gpt-4")
    response = model.invoke(state["question"])
    return {"answer": response.content}

# 그래프 구축
graph = StateGraph(State)
graph.add_node("generate", generate_answer)
graph.set_entry_point("generate")
graph.add_edge("generate", END)

# 컴파일 및 실행
app = graph.compile()
result = app.invoke({"question": "LangGraph란 무엇인가요?"})
print(result["answer"])

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, AIMessage
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# 모델 및 메모리 생성
model = ChatOpenAI(model="gpt-4")
memory = ConversationBufferMemory(return_messages=True)

# 대화 체인 생성
chain = ConversationChain(llm=model, memory=memory)

# 멀티 턴 대화
response1 = chain.predict(input="안녕하세요, 저는 철수예요")
response2 = chain.predict(input="제 이름이 뭐예요?")
print(response2)  # "철수"를 기억함

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, List

class State(TypedDict):
    messages: List

def chat(state: State) -> State:
    model = ChatOpenAI(model="gpt-4")
    response = model.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

# 그래프 구축
graph = StateGraph(State)
graph.add_node("chat", chat)
graph.set_entry_point("chat")
graph.add_edge("chat", END)

app = graph.compile()

# 상태를 가진 멀티 턴 대화
result1 = app.invoke({
    "messages": [HumanMessage("안녕하세요, 저는 철수예요")]
})
result2 = app.invoke({
    "messages": result1["messages"] + [HumanMessage("제 이름이 뭐예요?")]
})

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain_community.document_loaders import TextLoader

# 문서 로드 및 분할
loader = TextLoader("docs.txt")
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)
texts = text_splitter.split_documents(documents)

# 벡터 저장소 생성
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(texts, embeddings)

# RAG 체인 생성
llm = ChatOpenAI(model="gpt-4")
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever()
)

# 쿼리
result = qa_chain.invoke({"query": "문서가 무엇에 관한 건가요?"})

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from typing import TypedDict, List

class RAGState(TypedDict):
    query: str
    documents: List[str]
    answer: str

def retrieve(state: RAGState) -> RAGState:
    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma(persist_directory="./chroma", embedding_function=embeddings)
    docs = vectorstore.similarity_search(state["query"], k=3)
    return {"documents": [d.page_content for d in docs]}

def generate(state: RAGState) -> RAGState:
    llm = ChatOpenAI(model="gpt-4")
    context = "\n\n".join(state["documents"])
    prompt = f"컨텍스트: {context}\n\n질문: {state['query']}"
    response = llm.invoke(prompt)
    return {"answer": response.content}

# 그래프 구축
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)

app = graph.compile()
result = app.invoke({"query": "문서가 무엇에 관한 건가요?"})

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

# 커스텀 도구 정의
@tool
def get_weather(city: str) -> str:
    """도시의 날씨 가져오기"""
    return f"{city} 날씨: 맑음, 22°C"

@tool
def calculate(expression: str) -> str:
    """수학 식 계산"""
    return str(eval(expression))

# 도구 목록 생성
tools = [get_weather, calculate]

# 에이전트 생성
llm = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 도움이 되는 어시스턴트입니다."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# 실행
result = agent_executor.invoke({"input": "서울 날씨 어때요?"})

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from typing import TypedDict, List, Any

# 도구 정의
@tool
def get_weather(city: str) -> str:
    """도시의 날씨 가져오기"""
    return f"{city} 날씨: 맑음, 22°C"

@tool
def calculate(expression: str) -> str:
    """수학 식 계산"""
    return str(eval(expression))

tools = [get_weather, calculate]

# 상태 정의
class AgentState(TypedDict):
    messages: List[Any]

# 도구가 바인딩된 모델 생성
llm = ChatOpenAI(model="gpt-4")
llm_with_tools = llm.bind_tools(tools)

# 에이전트 노드 정의
def agent(state: AgentState) -> AgentState:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

# 그래프 구축
graph = StateGraph(AgentState)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(tools))
graph.set_entry_point("agent")

# 조건부 라우팅
def should_continue(state):
    if state["messages"][-1].tool_calls:
        return "tools"
    return END

graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "서울 날씨 어때요?"}]})

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import Tool
from langchain_core.prompts import ChatPromptTemplate

# 전문 에이전트 생성
def create_agent(role: str, expertise: str):
    llm = ChatOpenAI(model="gpt-4")
    prompt = ChatPromptTemplate.from_messages([
        ("system", f"당신은 {role}입니다. {expertise}"),
        ("human", "{input}"),
    ])
    return prompt | llm

# 에이전트 생성
researcher = create_agent("연구원", "정보를 수집합니다.")
analyst = create_agent("분석가", "데이터를 분석합니다.")
writer = create_agent("작성자", "보고서를 작성합니다.")

# 순차 실행 (단순화)
topic = "2024년 AI 트렌드"
research = researcher.invoke({"input": f"{topic} 연구"})
analysis = analyst.invoke({"input": f"분석: {research.content}"})
report = writer.invoke({"input": f"분석을 바탕으로 보고서 작성: {analysis.content}"})

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, List, Optional

class MultiAgentState(TypedDict):
    topic: str
    research: Optional[str]
    analysis: Optional[str]
    report: Optional[str]
    feedback: Optional[str]

llm = ChatOpenAI(model="gpt-4")

def researcher_node(state: MultiAgentState) -> MultiAgentState:
    prompt = f"이 주제를 철저히 연구하세요: {state['topic']}"
    response = llm.invoke(prompt)
    return {"research": response.content}

def analyst_node(state: MultiAgentState) -> MultiAgentState:
    prompt = f"이 연구를 분석하세요: {state['research']}"
    response = llm.invoke(prompt)
    return {"analysis": response.content}

def writer_node(state: MultiAgentState) -> MultiAgentState:
    prompt = f"이 분석을 바탕으로 보고서를 작성하세요: {state['analysis']}"
    response = llm.invoke(prompt)
    return {"report": response.content}

def reviewer_node(state: MultiAgentState) -> MultiAgentState:
    prompt = f"이 보고서를 검토하고 피드백을 제공하세요: {state['report']}"
    response = llm.invoke(prompt)
    return {"feedback": response.content}

# 조건: 수정 필요 여부
def should_revise(state):
    if "수정" in state["feedback"].lower():
        return "writer"
    return END

# 그래프 구축
graph = StateGraph(MultiAgentState)
graph.add_node("researcher", researcher_node)
graph.add_node("analyst", analyst_node)
graph.add_node("writer", writer_node)
graph.add_node("reviewer", reviewer_node)

graph.set_entry_point("researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", "reviewer")
graph.add_conditional_edges("reviewer", should_revise)

app = graph.compile()
result = app.invoke({"topic": "2024년 AI 트렌드"})

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 스트리밍이 가능한 모델 생성
model = ChatOpenAI(model="gpt-4", streaming=True)
prompt = ChatPromptTemplate.from_template("{topic}에 대한 이야기를 해주세요")
chain = prompt | model

# 스트리밍 출력
for chunk in chain.stream({"topic": "용감한 기사"}):
    print(chunk.content, end="", flush=True)

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict

class StreamState(TypedDict):
    topic: str
    story: str

async def generate_story(state: StreamState) -> StreamState:
    model = ChatOpenAI(model="gpt-4")
    prompt = f"{state['topic']}에 대한 이야기를 해주세요"

    story_parts = []
    async for chunk in model.astream(prompt):
        story_parts.append(chunk.content)
        print(chunk.content, end="", flush=True)

    return {"story": "".join(story_parts)}

# 그래프 구축
graph = StateGraph(StreamState)
graph.add_node("generate", generate_story)
graph.set_entry_point("generate")
graph.add_edge("generate", END)

app = graph.compile()

# 스트리밍 실행
import asyncio
asyncio.run(app.ainvoke({"topic": "용감한 기사"}))

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from tenacity import retry, stop_after_attempt, wait_exponential

# 재시도 데코레이터로 래핑
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_llm_with_retry(prompt: str) -> str:
    model = ChatOpenAI(model="gpt-4")
    try:
        response = model.invoke(prompt)
        return response.content
    except Exception as e:
        print(f"오류: {e}, 재시도 중...")
        raise

# 사용
try:
    result = call_llm_with_retry("LangChain이란 무엇인가요?")
except Exception as e:
    print(f"재시도 후 실패: {e}")

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Optional

class RetryState(TypedDict):
    prompt: str
    response: Optional[str]
    error: Optional[str]
    attempts: int

MAX_RETRIES = 3

def call_llm(state: RetryState) -> RetryState:
    model = ChatOpenAI(model="gpt-4")
    try:
        response = model.invoke(state["prompt"])
        return {"response": response.content, "error": None}
    except Exception as e:
        return {"error": str(e), "attempts": state.get("attempts", 0) + 1}

def handle_error(state: RetryState) -> RetryState:
    print(f"{state['attempts']}번째 시도 실패: {state['error']}")
    return state

def should_retry(state: RetryState):
    if state.get("response"):
        return END
    if state.get("attempts", 0) >= MAX_RETRIES:
        return END
    return "call_llm"

# 그래프 구축
graph = StateGraph(RetryState)
graph.add_node("call_llm", call_llm)
graph.add_node("handle_error", handle_error)

graph.set_entry_point("call_llm")
graph.add_conditional_edges("call_llm", should_retry)
graph.add_edge("handle_error", "call_llm")

app = graph.compile()
result = app.invoke({"prompt": "LangGraph란 무엇인가요?", "attempts": 0})