【LangGraph入門】基本の使い方と簡単なグラフ作成

from langgraph.graph import StateGraph
from langgraph.graph import END
from pydantic import BaseModel

# ステートの定義
class State(BaseModel):
    value: int
    step1: str = None
    step2: str = None

# ノードのアクションを定義
def node1_action(state: State) -> State:
    value = state.value
    value += 1 # 値を変更
    return {"step1": "Processed by Node 1", "value": value}

def node2_action(state: State) -> State:
    value = state.value
    value *= 2 # 値を変更
    return {"step2": "Processed by Node 2", "value": value}

# グラフを定義
workflow = StateGraph(State)

# node1から処理を開始
workflow.set_entry_point("node1")

# ノードを作成
workflow.add_node("node1", node1_action)
workflow.add_node("node2", node2_action)

# エッジを作成
workflow.add_edge("node1", "node2") # Node 1 → Node 2 の順につなぐ
workflow.add_edge("node2", END)     # Node 2 → END の順につなぐ

# グラフをコンパイルして実行
graph = workflow.compile()
result_state = graph.invoke({"value": 5})

# 結果を表示
print("\nFinal State:", result_state)

class State(BaseModel):
    value: int
    step1: str = None
    step2: str = None

def node1_action(state: State) -> State:
    value = state.value
    value += 1  # 値を変更
    return {"step1": "Processed by Node 1", "value": value}

def node2_action(state: State) -> State:
    value = state.value
    value *= 2  # 値を変更
    return {"step2": "Processed by Node 2", "value": value}

workflow = StateGraph(State)
workflow.set_entry_point("node1")

workflow.add_node("node1", node1_action)
workflow.add_node("node2", node2_action)
workflow.add_edge("node1", "node2")
workflow.add_edge("node2", END)

graph = workflow.compile()
result_state = graph.invoke({"value": 5})
print("\nFinal State:", result_state)

Final State: {'value': 12, 'step1': 'Processed by Node 1', 'step2': 'Processed by Node 2'}

from langgraph.graph import StateGraph
from langgraph.graph import END
import operator
from typing import Annotated
from pydantic import BaseModel, Field
from typing import Any
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.runnables import ConfigurableField
from openai import OpenAI
import os

os.environ['OPENAI_API_KEY'] = 'xxxx'

llm = ChatOpenAI(model="gpt-4o", temperature=0.0)
llm = llm.configurable_fields(max_tokens=ConfigurableField(id='max_tokens'))

class State(BaseModel):
    query: str = Field(..., description="ユーザーからの質問")
    current_role: str = Field(
        default="", description="選定された回答ロール"
    )
    answers: Annotated[list[str], operator.add] = Field(
        default=[], description="回答履歴"
    )


def selection_node(state: State) -> dict[str, Any]:
    query = state.query
    prompt = ChatPromptTemplate.from_template(
    """質問を分析し、最も適切なエージェントを選択してください。

    ### 選択肢
    1.web検索エージェント:AIがWeb検索を行い、最適な情報を提供します。
    2.社員情報検索エージェント：該当の技術スタックに精通する社員を検索し、その情報を提供する。

    ### 質問
     {query}

    ### 出力形式
    回答は選択肢の番号（1または2）のみを返してください。
    """.strip()
    )
    # 選択肢の番号のみを返すことを期待したいため、max_tokensの値を1に変更
    chain = prompt | llm.with_config(configurable=dict(max_tokens=1)) | StrOutputParser()
    role_number = chain.invoke({"query": query})

    return {"current_role": role_number}

def search_perplexity(state: State) -> dict[str, Any]:
    """2024年の時点の最新の情報をウェブ検索する関数"""
    query = state.query
    messages = [
        {
            "role": "system",
            "content": (
                "You are an artificial intelligence assistant and you need to "
                "engage in a helpful, detailed, polite conversation with a user."
            ),
        },
        {
            "role": "user",
            "content": (
                query
            ),
        },
    ]
    client = OpenAI(api_key="xxxx", base_url="https://api.perplexity.ai")

    response = client.chat.completions.create(
        model="llama-3.1-sonar-large-128k-online",
        messages=messages,
    )
    answer = response.choices[0].message.content

    return {"answers": [answer]}

def database_search_employee (state: State) -> dict[str, Any]:
    """社員情報検索を実行する関数"""
    query = state.query
    # 本来は検索機能を実装するべきだか、ここではダミーデータが返ってきたことを想定
    search_result = """Aさん
    - 生産管理システムの最適化
        - 生産ラインの稼働状況をリアルタイムで監視するダッシュボードの開発
        - AIを活用した生産計画の自動化と効率化
        - 現場スタッフ向けの使いやすいモバイルアプリの設計・導入
    
    Bさん
    - 製品検査プロセスのデジタル化
        - 画像認識技術を活用した自動検査システムの構築
        - データ解析による不良品の原因分析と予防策の提案
        - 検査結果の可視化ツールを用いた品質管理レポートの作成

    Cさん
    - IoTを活用した工場のスマート化推進
        - センサーデータを活用した設備稼働状況の予測分析
        - クラウドプラットフォームによる生産データの一元管理
        - 設備保全のためのAI予測保全モデルの導入"""
    
    prompt = ChatPromptTemplate.from_template(
    """あなたは社員情報検索エージェントです。以下の社員情報を元にユーザの質問に回答してください。

    ### 社員情報
    {search_result}

    ### 質問
    {query}
    """.strip()
    )
    chain = prompt | llm | StrOutputParser()
    answer = chain.invoke({"search_result": search_result,"query": query})
    return {"answers": [answer]}

def result_summary(state: State) -> dict[str, Any]:
    """前のノードで得られた結果を統合するノード"""
    summary = f"Question: {state.query}\nAnswer: {state.answers[-1]}"
    return {"answers": [summary]}

from langgraph.graph import StateGraph

workflow = StateGraph(State)

# ノードの追加
workflow.add_node("selection", selection_node)
workflow.add_node("search_perplexity", search_perplexity)
workflow.add_node("database_search_employee", database_search_employee)
workflow.add_node("result_summary", result_summary)

# selectionノードから処理を開始
workflow.set_entry_point("selection")

# エッジの追加
workflow.add_conditional_edges(
    "selection",
    lambda state: state.current_role, #リスト
    {
        "1": "search_perplexity",
        "2": "database_search_employee"
    },
)

workflow.add_edge("search_perplexity", "result_summary")
workflow.add_edge("database_search_employee", "result_summary")
workflow.add_edge("result_summary", END)


initial_state = State(query="製造業の自動化についてwebの情報も踏まえて教えてください")
result = compiled.invoke(initial_state)
print(result["answers"][-1])

def selection_node(state: State) -> dict[str, Any]:
    query = state.query
    prompt = ChatPromptTemplate.from_template(
    """質問を分析し、最も適切なエージェントを選択してください。

    ### 選択肢
    1.web検索エージェント:AIがWeb検索を行い、最適な情報を提供します。
    2.社員情報検索エージェント：該当の技術スタックに精通する社員を検索し、その情報を提供する。

    ### 質問
     {query}

    ### 出力形式
    回答は選択肢の番号（1または2）のみを返してください。
    """.strip()
    )
    # 選択肢の番号のみを返すことを期待したいため、max_tokensの値を1に変更
    chain = prompt | llm.with_config(configurable=dict(max_tokens=1)) | StrOutputParser()
    role_number = chain.invoke({"query": query})

    return {"current_role": role_number}

def search_perplexity(state: State) -> dict[str, Any]:
    """2024年の時点の最新の情報をウェブ検索する関数"""
    query = state.query
    messages = [
        {
            "role": "system",
            "content": (
                "You are an artificial intelligence assistant and you need to "
                "engage in a helpful, detailed, polite conversation with a user."
            ),
        },
        {
            "role": "user",
            "content": (
                query
            ),
        },
    ]
    client = OpenAI(api_key="xxxxx", base_url="https://api.perplexity.ai")

    response = client.chat.completions.create(
        model="llama-3.1-sonar-large-128k-online",
        messages=messages,
    )
    answer = response.choices[0].message.content

    return {"answers": [answer]}

def database_search_employee (state: State) -> dict[str, Any]:
    """社員情報検索を実行する関数"""
    query = state.query
    # 本来は検索機能を実装するべきだか、ここではダミーデータが返ってきたことを想定
    search_result = """Aさん
    - 生産管理システムの最適化
        - 生産ラインの稼働状況をリアルタイムで監視するダッシュボードの開発
        - AIを活用した生産計画の自動化と効率化
        - 現場スタッフ向けの使いやすいモバイルアプリの設計・導入
    
    Bさん
    - 製品検査プロセスのデジタル化
        - 画像認識技術を活用した自動検査システムの構築
        - データ解析による不良品の原因分析と予防策の提案
        - 検査結果の可視化ツールを用いた品質管理レポートの作成

    Cさん
    - IoTを活用した工場のスマート化推進
        - センサーデータを活用した設備稼働状況の予測分析
        - クラウドプラットフォームによる生産データの一元管理
        - 設備保全のためのAI予測保全モデルの導入"""
    
    prompt = ChatPromptTemplate.from_template(
    """あなたは社員情報検索エージェントです。以下の社員情報を元にユーザの質問に回答してください。

    ### 社員情報
    {search_result}

    ### 質問
    {query}
    """.strip()
    )
    chain = prompt | llm | StrOutputParser()
    answer = chain.invoke({"search_result": search_result,"query": query})
    return {"answers": [answer]}

def result_summary(state: State) -> dict[str, Any]:
    """前のノードで得られた結果を統合するノード"""
    summary = f"Question: {state.query}\nAnswer: {state.answers[-1]}"
    return {"answers": [summary]}

# ノードの追加
workflow.add_node("selection", selection_node)
workflow.add_node("search_perplexity", search_perplexity)
workflow.add_node("database_search_employee", database_search_employee)
workflow.add_node("result_summary", result_summary)

# selectionノードから処理を開始
workflow.set_entry_point("selection")

# エッジの追加
workflow.add_conditional_edges(
    "selection",
    lambda state: state.current_role, #リスト
    {
        "1": "search_perplexity",
        "2": "database_search_employee"
    },
)

workflow.add_edge("search_perplexity", "result_summary")
workflow.add_edge("database_search_employee", "result_summary")
workflow.add_edge("result_summary", END)

compiled = workflow.compile()

compiled = workflow.compile()
initial_state = State(query="製造業の自動化についてwebの情報も踏まえて教えてください")
result = compiled.invoke(initial_state)

{'query': '製造業の自動化についてwebの情報も踏まえて教えてください',
 'current_role': '1',
 'answers': ['製造業の自動化は、効率の向上、コストの削減、および品質の改善を目指す重要な戦略です。以下は、ウェブ上の情報を基にした主要なポイントです。\n\n## 自動化技術の利点\n\n### 労働コストの削減と生産率の向上\n自動化設備は人力操作を減少させ、労働コストを削減することができます。自動化により、生産時間とステップが一致性を持つため、生産率が大幅に向上します。例えば、自動化された生産ラインは、24/7の不間断生産を実現し、出荷効率を20%以上提高させることが可能です.\n\n### 品質の向上と誤差の減少\n自動化技術、特にAIと機器学習を活用することで、製品の品質検査が高度化されます。例えば、ブリヂストンタイヤの場合、AIが膨大な品質データを分析し、微細な不良も見逃さず、高品質なタイヤを安定的に生産しています.\n\n### 資源配置の最適化\nAIを用いたデータ分析により、生産流程が最適化され、資源配置がより合理になることが期待されます。AIは需要予測や設備稼働状況を分析し、最適な生産計画を自動生成することができます。これにより、生産効率が飛躍的に向上し、エネルギー消費も削減されます.\n\n## 智慧製造とAIの役割\n\n### 即時監控と予防性維護\n智慧製造では、物联网（IoT）技術とAIを組み合わせて、設備の即時監控と予防性維護を実現します。AIは装置の運行データを分析し、故障を予測し、停機時間を30%減少させることができます。また、生産効率も15%以上向上し、維護コストを削減します.\n\n### 生産計画の最適化\nAIは膨大な製造データを学習し、需要予測や設備稼働状況を分析して、最適な生産計画を自動生成します。これにより、複雑な条件設定も迅速に判断でき、生産効率が大幅に向上します.\n\n### スキルレス化と技能継承\nAIは熟練作業員のノウハウを機械的に抽出し、高精度な分析と最適化を可能にします。例えば、ブリヂストンタイヤでは、AIが熟練工の技術やノウハウをデジタルデータ化し、製造現場の「スキルレス化」を実現しています.\n\n## 制造実行システム（MES）の役割\n\n### 製造工程のリアルタイム管理\n製造実行システム（MES）は、稼働監視、設備管理、生産実績の集計分析などを担うシステム基盤です。MESを利用することで、製造現場の状況をリアルタイムで把握し、ボトルネックとなる工程を洗い出すことができます。また、不良品の削減や製造工程の無理や無駄を減少させることが可能です.\n\n## 自動化の実施方法\n\n### 企業資源と生産程序の計画\n自動化を実施する際には、企業資源の整合と数位化、生産程序の規劃が重要です。IoT技術や感測器を導入し、即時データを収集・分析して、生産作業流程を調整する必要があります.\n\n### 自動化設備の導入\n先進の工業ロボット、自動化装置、電腦視覺システムを導入し、重複性任務を自動化することで、24/7の不間断生産を実現します。また、AIを用いた画像認識技術を活用して、製品の欠陥を自動で検出するシステムも開発されます.\n\nこれらのポイントを踏まえ、製造業の自動化は、効率の向上、コストの削減、品質の改善を実現するための重要な戦略です。AIと物联网技術の活用が、製造業の数位化と智慧製造への進展を支える鍵となります。',
  'Query: 製造業の自動化についてwebの情報も踏まえて教えてください\nAnswer: 製造業の自動化は、効率の向上、コストの削減、および品質の改善を目指す重要な戦略です。以下は、ウェブ上の情報を基にした主要なポイントです。\n\n## 自動化技術の利点\n\n### 労働コストの削減と生産率の向上\n自動化設備は人力操作を減少させ、労働コストを削減することができます。自動化により、生産時間とステップが一致性を持つため、生産率が大幅に向上します。例えば、自動化された生産ラインは、24/7の不間断生産を実現し、出荷効率を20%以上提高させることが可能です.\n\n### 品質の向上と誤差の減少\n自動化技術、特にAIと機器学習を活用することで、製品の品質検査が高度化されます。例えば、ブリヂストンタイヤの場合、AIが膨大な品質データを分析し、微細な不良も見逃さず、高品質なタイヤを安定的に生産しています.\n\n### 資源配置の最適化\nAIを用いたデータ分析により、生産流程が最適化され、資源配置がより合理になることが期待されます。AIは需要予測や設備稼働状況を分析し、最適な生産計画を自動生成することができます。これにより、生産効率が飛躍的に向上し、エネルギー消費も削減されます.\n\n## 智慧製造とAIの役割\n\n### 即時監控と予防性維護\n智慧製造では、物联网（IoT）技術とAIを組み合わせて、設備の即時監控と予防性維護を実現します。AIは装置の運行データを分析し、故障を予測し、停機時間を30%減少させることができます。また、生産効率も15%以上向上し、維護コストを削減します.\n\n### 生産計画の最適化\nAIは膨大な製造データを学習し、需要予測や設備稼働状況を分析して、最適な生産計画を自動生成します。これにより、複雑な条件設定も迅速に判断でき、生産効率が大幅に向上します.\n\n### スキルレス化と技能継承\nAIは熟練作業員のノウハウを機械的に抽出し、高精度な分析と最適化を可能にします。例えば、ブリヂストンタイヤでは、AIが熟練工の技術やノウハウをデジタルデータ化し、製造現場の「スキルレス化」を実現しています.\n\n## 制造実行システム（MES）の役割\n\n### 製造工程のリアルタイム管理\n製造実行システム（MES）は、稼働監視、設備管理、生産実績の集計分析などを担うシステム基盤です。MESを利用することで、製造現場の状況をリアルタイムで把握し、ボトルネックとなる工程を洗い出すことができます。また、不良品の削減や製造工程の無理や無駄を減少させることが可能です.\n\n## 自動化の実施方法\n\n### 企業資源と生産程序の計画\n自動化を実施する際には、企業資源の整合と数位化、生産程序の規劃が重要です。IoT技術や感測器を導入し、即時データを収集・分析して、生産作業流程を調整する必要があります.\n\n### 自動化設備の導入\n先進の工業ロボット、自動化装置、電腦視覺システムを導入し、重複性任務を自動化することで、24/7の不間断生産を実現します。また、AIを用いた画像認識技術を活用して、製品の欠陥を自動で検出するシステムも開発されます.\n\nこれらのポイントを踏まえ、製造業の自動化は、効率の向上、コストの削減、品質の改善を実現するための重要な戦略です。AIと物联网技術の活用が、製造業の数位化と智慧製造への進展を支える鍵となります。']}

initial_state = State(query="製品検査プロセスのデジタル化が得意な社員を教えてください")
result = compiled.invoke(initial_state)

{'query': '製品検査プロセスのデジタル化が得意な社員を教えてください',
 'current_role': '2',
 'answers': ['製品検査プロセスのデジタル化が得意な社員はBさんです。Bさんは、画像認識技術を活用した自動検査システムの構築や、データ解析による不良品の原因分析と予防策の提案、検査結果の可視化ツールを用いた品質管理レポートの作成を担当しています。',
  'Query: 製品検査プロセスのデジタル化が得意な社員を教えてください\nAnswer: 製品検査プロセスのデジタル化が得意な社員はBさんです。Bさんは、画像認識技術を活用した自動検査システムの構築や、データ解析による不良品の原因分析と予防策の提案、検査結果の可視化ツールを用いた品質管理レポートの作成を担当しています。']}

from IPython.display import Image, display

display(Image(compiled.get_graph().draw_mermaid_png()))

mermaid_syntax = compiled.get_graph().draw_mermaid()
print(mermaid_syntax)

%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
	__start__([<p>__start__</p>]):::first
	selection(selection)
	search_perplexity(search_perplexity)
	database_search_employee(database_search_employee)
	result_summary(result_summary)
	__end__([<p>__end__</p>]):::last
	__start__ --> selection;
	database_search_employee --> result_summary;
	result_summary --> __end__;
	search_perplexity --> result_summary;
	selection -.  1  .-> search_perplexity;
	selection -.  2  .-> database_search_employee;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6fc