LangGraph 簡単なサンプル3つ


:::note 
[LangGraph Quick Start](https://langchain-ai.github.io/langgraph/tutorials/introduction/) （参考:前回の[LangGraph Quick Start(日本語訳)）](https://qiita.com/suzookita/items/8551eb2fd3bd342d670a)のPart1をベースに、普段利用のOpenAI,ChatPromptTemplateを利用した記載に修正しサンプルを残す
:::
- sample1
    - Quick Start Part1のモデル変更
    - Anthropicのmodel:claude-3-5-sonnet-20240620 =>OpenAI(model:gpt-4o-mini)
- sample2
    - state add_messagesを利用しない
    - node(chatbot->agent)
    - graphname(graph_builder => workflow)
    - ChatPromptTemplateの利用
- sample3
    - nodeを２つに(agent_1,agent_2)
    - 結果出力をstreamモードと,invokeの両方を記載


```text:pip list
openai                                   1.58.1
langchain                                0.2.17
langchain-community                      0.2.17
langchain-core                           0.2.43
langchain-experimental                   0.0.64
langchain-openai                         0.1.25
langgraph                                0.2.26
langgraph-checkpoint                     1.0.12
langsmith                                0.1.147
python 3.9.17
langchain-anthropic                      0.1.23 #今回は使ってません
# 訳あって、まだpython 3.9.17、langchain v0.2.17使ってます
```

# 0.環境セットアップ
```python:python
import os
from os.path import join, dirname, abspath
from dotenv import load_dotenv

dir_path = dirname(abspath("__file__"))
dotenv_path = join(dir_path, '../[your_dir_name]/.env')
load_dotenv(dotenv_path, verbose=True)

#print(os.environ['ANTHROPIC_API_KEY'])
#print(os.environ['OPENAI_API_KEY'])
#print(os.environ['TAVILY_API_KEY'])
```

# 1.Sample1
- LangGraph Part1 Part 1: Build a Basic Chatbotをベース
- 利用するStateは、公式そのまま
    - 1つのキー:messages を持つ TypedDict
    - add_messages リデューサー関数
- 変更点
    - anthropicではなくopenai 
    


```python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    # メッセージのタイプは「リスト」です。アノテーション内の `add_messages`
    # 関数は、この状態キーの更新方法を定義します (この場合、メッセージを
    # 上書きするのではなく、リストに追加します)
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)
```

```python
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(temperature=0, model_name="gpt-4o-mini")

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

# 最初の引数は一意のノード名
# 2 番目の引数はノードが使用されるたびに呼び出される関数またはオブジェクト
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

graph = graph_builder.compile()
```
```python
#可視化
from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    # (任意) 追加の依存関係
    pass
```
<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/698129/c103b299-4a57-273c-9e3a-d4656abbe9e2.png" width="200px">

点線枠はグラフ名を記載するために加筆してます


```python
# LangGraph Streamによる実行処理を定義
def stream_graph_updates(user_input: str):
    for event in graph.stream({"messages": [("user", user_input)]}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)
```
```python
#実行
stream_graph_updates("こんにちは")
#Assistant: こんにちは！どういったことをお話ししましょうか？
```

# 2.Sample2
- Sample1との変更点
    - Stateに、add_messagesを利用しない
        - 簡易的なのでField(..., description='hogehoge fugafuga')の記載は省略
    - Chatbotではなく、Agentとして役割を与える（今回は日本語から英語への翻訳） 
    - 命名変更
        - ノード名・関数名:chatbot => node名:agent_1、関数名:agent1
        - graph_builder => workflow

```python:Stateの作成
#from typing import Annotated #使わない
from typing_extensions import TypedDict
from typing import List
from langgraph.graph import StateGraph, START, END
#from langgraph.graph.message import add_messages #使わない


class State(TypedDict):
    """
    グラフの状態を表す(Dict型)
    
    Attributes:
      query:str 質問
      answer:str 日本語の質問を英語に変換した答え
    """
    query:str
    answer:str
```

```python:Agentの作成
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(temperature=0, model_name="gpt-4o-mini")

def agent1(state:State): #英訳する関数（後でノードとして定義し利用する）
    query = state['query']
    
    # Prompt template(書き方1)
    #message = """与えられた以下質問を英語に翻訳してください。\n query:{query}"""
    #prompt1 = ChatPromptTemplate.from_messages(("human",message))

    # Prompt template(書き方2)
    prompt1 = ChatPromptTemplate([
      ('system', 'あなたは優秀な翻訳家です。'),
      ('user', """以下を英語に翻訳してください。\n query:{query}""")
    ])

    agent1 = (
        prompt1
        | llm
        | StrOutputParser()
    )
    
    answer = agent1.invoke(query)
    #print(answer)
    return {"query":query,"answer":answer}
```

```python:LangGraphの構築
# グラフの初期化
workflow = StateGraph(State)

# 最初の引数は一意のノード名
# 2 番目の引数はノードが使用されるたびに呼び出される関数またはオブジェクト
workflow.add_node("agent_1", agent1)
workflow.add_edge(START, "agent_1")
workflow.add_edge("agent_1", END)

graph_app = workflow.compile()
```

```python:グラフの可視化
#可視化
from IPython.display import Image, display
try:
    display(Image(graph_app.get_graph().draw_mermaid_png()))
except Exception:
    # (任意)追加の依存関係
    pass
```

<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/698129/990f8a6a-2a8e-2eb3-2824-b2fdcef572c5.png" width="220px">

点線枠はグラフ名を記載するために加筆してます

```python
# Stateの初期値
initial_state = State(query='こんにちは') #Agentに翻訳してほしい言葉を入れる
```

実行結果の確認
```python
# streamを利用する場合
for chunk in graph_app.stream(initial_state):
    print(chunk)

# {'agent1': {'query': 'こんにちは', 'answer': 'Hello.'}}
```

```python
# streamを利用する場合
for chunk in graph_app.stream(initial_state):
    print(chunk) #--- print(1)
    # さらにvaluesをprintする場合
    # Agent名を変数に格納
    ai_name = list(chunk.keys())[0] 
    for value in chunk.values():
        print("Assistant:", value) #---print(2)
        print(ai_name, value) #----print(3)

# print(1) --- {'agent_1': {'query': 'こんにちは', 'answer': 'Hello.'}}
# print(2) --- Assistant: {'query': 'こんにちは', 'answer': 'Hello.'}
# print(3) --- agent_1 {'query': 'こんにちは', 'answer': 'Hello.'}
```

# 3.Sample3
- Sample2との変更点
    - Agentを追加して、２つのAgentに働いてもらう
        - 英語をフランス語に翻訳するAgent

```python
from typing_extensions import TypedDict
from typing import List
import pandas as pd

from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    """
    グラフの状態を表す(Dict型)
    
    Attributes:
      query:str 質問
      answer:str 日本語の質問を英語に変換した返事
      answer2:str 英語をフランス語に翻訳した返事
    """
    query:str
    answer:str
    answer2:str
```

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

llm = ChatOpenAI(temperature=0, model_name="gpt-4o-mini")
```
```python:agent_1(英語に翻訳)
def agent1(state:State): #英訳する関数（後でノードとして定義し利用する）
    query = state['query']
    
    # Prompt template(書き方1)
    #message = """与えられた以下質問を英語に翻訳してください。\n query:{query}"""
    #prompt1 = ChatPromptTemplate.from_messages(("human",message))

    # Prompt template(書き方2)
    prompt1 = ChatPromptTemplate([
      ('system', 'あなたは優秀な翻訳家です。'),
      ('user', """以下を英語に翻訳してください。\n query:{query}""")
    ])

    agent1 = (
        prompt1
        | llm
        | StrOutputParser()
    )
    
    answer = agent1.invoke(query)
    #print(answer)
    return {"query":query,"answer":answer}
```

```python: agent_2(フランス語に翻訳)
def agent2(state:State): #フランス語訳する関数（後でノードとして定義し利用する）
    answer = state['answer']
    
    # Prompt template(書き方1)
    #message = """与えられた以下質問をフランス語に翻訳してください。\n answer:{answer}"""
    #prompt1 = ChatPromptTemplate.from_messages(("human",message))

    # Prompt template(書き方2)
    prompt2 = ChatPromptTemplate([
      ('system', 'あなたは優秀な翻訳家です。'),
      ('user', """以下をフランス語に翻訳してください。\n answer:{answer}""")
    ])

    agent2 = (
        prompt2
        | llm
        | StrOutputParser()
    )
    
    answer2 = agent2.invoke(answer)
    #print(answer)
    return {"answer":answer,"answer2":answer2}
```

```python:LangGraph構築
# グラフの初期化
workflow2 = StateGraph(State)

# 最初の引数は一意のノード名
# 2 番目の引数はノードが使用されるたびに呼び出される関数またはオブジェクト
workflow2.add_node("agent_1", agent1)
workflow2.add_node("agent_2", agent2)  # <==追加(2つ目のAgent)
workflow2.add_edge(START, "agent_1")
workflow2.add_edge("agent_1","agent_2") #<==追加(agent1からagent2へのedge)
workflow2.add_edge("agent_2", END)     #<==修正(ENDに向かうagentを1から2に修正)

graph_app2 = workflow2.compile()
```

```python:可視化
from IPython.display import Image, display

try:
    display(Image(graph_app2.get_graph().draw_mermaid_png()))
except Exception:
    # This requires some extra dependencies and is optional
    pass
```
<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/698129/5973e721-03db-b610-2fae-f5bb47858323.png" width="200px">



点線枠はグラフ名を記載するために加筆してます

```python:Stateの初期値
# Stateの初期値
initial_state = State(query='こんにちは')
```

実行確認
```python
# stream出力
for chunk in graph_app2.stream(initial_state):
    print(chunk)

# {'agent_1': {'query': 'こんにちは', 'answer': 'Hello.'}}
# {'agent_2': {'answer': 'Hello.', 'answer2': 'Bonjour.'}}
```

- stream_modeには2種類ある
    - stream_mode = "**values**" : 各ノードが呼び出された後の**全ての**State
    - stream_mode = "**updates**" : 各ノードが呼び出された後の**更新された**State
        - keyにnode名を取って更新されたstateのみを返す

```python
# stream出力
# stream_mode="values"の場合
for chunk in graph_app2.stream(initial_state, stream_mode="values"): 
    print(chunk)
    final_chunk = chunk
    
print("final_chunk:",final_chunk)

# {'query': 'こんにちは'}
# {'query': 'こんにちは', 'answer': 'Hello.'}
# {'query': 'こんにちは', 'answer': 'Hello.', 'answer2': 'Bonjour.'}
# final_chunk: {'query': 'こんにちは', 'answer': 'Hello.', 'answer2': 'Bonjour.'}

```

```python
# stream出力
# stream_mode="updates"の場合
for chunk in graph_app2.stream(initial_state, stream_mode="updates"): 
    print(chunk)
    final_chunk = chunk
    
print("final_chunk:",final_chunk)

# {'agent_1': {'query': 'こんにちは', 'answer': 'Hello.'}}
# {'agent_2': {'answer': 'Hello.', 'answer2': 'Bonjour.'}}
# final_chunk: {'agent_2': {'answer': 'Hello.', 'answer2': 'Bonjour.'}}
```

**Tip**:辞書型なのでDataFrameにして表示も可能
```python
import pandas as pd
df=pd.DataFrame([final_chunk['agent_2']])
display(df)
```

|	|answer	|answer2|
|:--|:--|:--|
|0	|Hello.	|Bonjour.|

invokeによる出力
```python
#invoke出力
graph_app2.invoke(initial_state)
#{'query': 'こんにちは', 'answer': 'Hello.', 'answer2': 'Bonjour.'}
```

```python
# ivoke出力
# 最終結果の値だけほしい場合
dict_data = graph_app2.invoke(initial_state)
dict_data['answer2']

# 'Bonjour.'
```

```python
# inovke出力
# 最終のstatusだけ、pandas DataFrame に変換し、HTMLファイル出力する場合
# DF変換
dict_data = graph_app2.invoke(initial_state)
df=pd.DataFrame([dict_data])
display(df)

# HTMLファイル出力
df.to_html("sample_df.html",index=False, encoding='utf-8')
print("sample_df.htmlに出力しました")

# HTML出力
HTML=df.to_html(index=False)
# Flaskでtemplateに入れテーブル表示したい場合などに利用
# ファイル出力しない場合はencoding='utf-8'をつけるとエラーになるので注意
```
出力ファイル
```html:sample_df.html
<table border="1" class="dataframe">
  <thead>
    <tr style="text-align: right;">
      <th>query</th>
      <th>answer</th>
      <th>answer2</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <td>こんにちは</td>
      <td>Hello.</td>
      <td>Bonjour.</td>
    </tr>
  </tbody>
</table>
```
|	|query|	answer|	answer2|
|:--|:--|:--|:--|
|0	|こんにちは	|Hello.	|Bonjour.|

```python
# sample_df.htmlに出力しました
```

# 参考
- 参考にさせていただきました

https://qiita.com/ikedachin/items/2e7197ea87861247425f


https://qiita.com/TKfumi/items/05fc0208014a83fb8079


https://note.com/rrrrrrrrrr_666/n/n3934dfc242b2