LangGraphのインストールと使用方法：ステップバイステップガイド

以下のHTMLコードを日本語に翻訳してください。次のようなURLがあることを確認してください： https://anakin.ai/apps/keywords または https://anakin.ai/blog/keywords このURLを次のように変換してください： https://anakin.ai/ja-jp/apps/keywords https://anakin.ai/ja-jp/blog/keywords 以下がHTMLコードです：

💡

AIの最新トレンドに興味はありませんか？

それなら、Anakin AIを見逃すことはできません！

Anakin AIは、ワークフローの自動化に最適なオールインワンプラットフォームで、使いやすいノーコードアプリビルダーで強力なAIアプリを作成できます。Llama 3、Claude Sonnet 3.5、GPT-4、Uncensored LLMs、Stable DiffusionなどでAnakin AIと一緒に夢のようなAIアプリを数分で作成しましょう！

無料で始める

インロタクション

LangGraphは、大規模言語モデル（LLMs）を使用してステートフルなマルチアクターアプリケーションを構築するための強力なライブラリです。このガイドでは、LangGraphの使用方法について、インストールから複雑なAIエージェントの作成まで、詳細なコード例を使用して説明します。

1. インストールとセットアップ

まず、LangGraphとその依存関係をインストールしましょう：

pip install langgraph langchain langchain_openai tavily-python

次に、環境変数を設定します：

import os
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
os.environ["TAVILY_API_KEY"] = "your-tavily-api-key"

2. 基本的なLangGraphアプリケーションの作成

インターネットを検索し、質問に答えることができるシンプルなエージェントを作成しましょう。

ステップ1：必要なモジュールをインポートする

from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain.tools import Tool
from langchain.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage, AIMessage, FunctionMessage
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
import json
from typing import TypedDict, Annotated, List

ステップ2：ステートとツールを定義する

class State(TypedDict):
    messages: Annotated[List[HumanMessage | AIMessage | FunctionMessage], "The messages in the conversation"]

tools = [TavilySearchResults(max_results=1)]
tool_executor = ToolExecutor(tools)
model = ChatOpenAI(temperature=0).bind_tools(tools)

ステップ3：エージェントとアクションノードを作成する

def agent_node(state):
    messages = state['messages']
    response = model.invoke(messages)
    return {"messages": [response]}

def action_node(state):
    messages = state['messages']
    last_message = messages[-1]
    action = last_message.additional_kwargs["function_call"]
    result = tool_executor.invoke(action)
    return {"messages": [FunctionMessage(content=str(result), name=action["name"])]}

def should_continue(state):
    messages = state['messages']
    last_message = messages[-1]
    if isinstance(last_message, AIMessage) and "function_call" in last_message.additional_kwargs:
        return "continue"
    return "end"

ステップ4：グラフ構造を作成する

workflow = StateGraph(State)
workflow.add_node("agent", agent_node)
workflow.add_node("action", action_node)
workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END
    }
)
workflow.add_edge("action", "agent")

ステップ5：グラフをコンパイルして実行する

app = workflow.compile()
inputs = {"messages": [HumanMessage(content="What is the weather in San Francisco?")]}
result = app.invoke(inputs)

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

3. LangGraphの高度な機能

永続性

LangGraphには組み込みの永続性機能があります。以下の方法で使用することができます：

from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string(":memory:")
app = workflow.compile(checkpointer=memory)

# 永続性を持ってグラフを実行
result = app.invoke(inputs)

# チェックポイントから再開
checkpoint_id = result['checkpoint_id']
resumed_result = app.invoke(inputs, checkpoint_id=checkpoint_id)

ストリーミング

生成されるとすぐに出力をストリームに送信するには：

for event in app.stream(inputs):
    for key, value in event.items():
        if key == 'messages':
            print(f"新しいメッセージ：{value[-1].content}")

ヒューマンインザループ

ヒューマンフィードバックを組み込むには：

def human_approval_node(state):
    print("現在のステート：", state)
    approval = input("承認しますか？（はい/いいえ）：")
    return {"approved": approval.lower() == "yes"}

workflow.add_node("human_approval", human_approval_node)
workflow.add_edge("agent", "human_approval")
workflow.add_conditional_edges(
    "human_approval",
    lambda x: "continue" if x["approved"] else "end",
    {
        "continue": "action",
        "end": END
    }
)

4. LangGraphを使用したより複雑なアプリケーションの構築

協力的なタスクの解決のためのマルチエージェントシステムを作成しましょう。

ステップ1：複数のエージェントを定義する

def researcher_agent(state):
    # 情報の調査ロジック
    return {"messages": [AIMessage(content="Research results...")]}

def writer_agent(state):
    # 調査に基づいたコンテンツの作成ロジック
    return {"messages": [AIMessage(content="Written content...")]}

def editor_agent(state):
    # コンテンツの編集と改善のロジック
    return {"messages": [AIMessage(content="Edited content...")]}

ステップ2：より複雑なグラフの作成

workflow = StateGraph(State)
workflow.add_node("researcher", researcher_agent)
workflow.add_node("writer", writer_agent)
workflow.add_node("editor", editor_agent)
workflow.set_entry_point("researcher")

workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "editor")
workflow.add_conditional_edges(
    "editor",
    lambda x: "refine" if "needs_refinement" in x['messages'][-1].content else "end",
    {
        "refine": "writer",
        "end": END
    }
)

ステップ3：高度な機能の実装

# ストリーミングサポートを追加
app = workflow.compile()
for event in app.stream({"messages": [HumanMessage(content="Write a report on AI advancements")]}):
    print(f"現在のステップ：{event['current_step']}")
    if 'messages' in event:
        print(f"最新のメッセージ：{event['messages'][-1].content}")

# 永続性を追加
memory = SqliteSaver.from_conn_string("collaborative_agents.db")
app_with_persistence = workflow.compile(checkpointer=memory)

# 永続性を持って実行
result = app_with_persistence.invoke({"messages": [HumanMessage(content="Write a report on AI advancements")]})
checkpoint_id = result['checkpoint_id']

# チェックポイントから再開
resumed_result = app_with_persistence.invoke({}, checkpoint_id=checkpoint_id)

5. LangGraphのためのベストプラクティスとヒント

ステート管理：ステートオブジェクトをクリーンで構造化された状態に保ちます。明確さのために型ヒントを使用します：

class DetailedState(TypedDict):
    messages: List[HumanMessage | AIMessage | FunctionMessage]
    research_data: dict
    draft_content: str
    edit_history: List[str]

エラーハンドリング：ノード関数でtry-exceptブロックを実装します：

def safe_researcher_agent(state):
    try:
        # 調査ロジックここに記述
        return {"messages": [AIMessage(content="Research results...")]}
    except Exception as e:
        return {"messages": [AIMessage(content=f"Error in research: {str(e)}")]}

モジュラーな設計：複雑なロジックをより小さく再利用可能な関数に分割します：

def perform_research(query):
    # 調査ロジック
    pass

def analyze_results(research_data):
    # 分析ロジック
    pass

def researcher_agent(state):
    query = state['messages'][-1].content
    research_data = perform_research(query)
    analysis = analyze_results(research_data)
    return {"messages": [AIMessage(content=analysis)], "research_data": research_data}

テスト：個々のノードに対するユニットテストと、グラフ全体に対する統合テストを作成します：

import unittest

class TestResearcherAgent(unittest.TestCase):
    def test_researcher_agent(self):
        state = {"messages": [HumanMessage(content="Research AI")]}
        result = researcher_agent(state)
        self.assertIn("messages", result)
        self.assertIn("research_data", result)

if __name__ == '__main__':
    unittest.main()

ドキュメンテーション：関数とクラスに詳細なドキュストリングを追加します：

def editor_agent(state: State) -> State:
    """
    Writerエージェントが提供したコンテンツを編集して洗練します。

    Args:
        state (State): メッセージと下書きのコンテンツを含む現在のステート。

    Returns:
        State: 編集されたコンテンツと潜在的な改善フラグが含まれる更新されたステート。
    """
    # 編集ロジックここに記述
    pass

結論

このガイドでは、LangGraphの基本的なセットアップから複雑なマルチエージェントシステムの作成まで、包括的な概要を提供しました。これらの手順とベストプラクティスに従うことで、LangGraphを活用して、さまざまなタスクとシナリオを処理できる高度なステートフルAIアプリケーションを構築できます。特定のユースケースに最適なアプローチを見つけるために、さまざまなグラフ構造とノードの実装を試してみてください。

💡

無料で始める