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"]:
        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})