LangGraphの使い方メモ

はじめに

手順

# インストール
!pip install -U langchain langchain_community langgraph langchain_openai langchainhub tavily-python 

# 準備
from langchain import hub
from langchain.agents import create_openai_functions_agent
from langchain_openai.chat_models import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults

import os
from google.colab import userdata
os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY")
os.environ["TAVILY_API_KEY"] = userdata.get("TAVILY_API_KEY")
os.environ["LANGCHAIN_API_KEY"] = userdata.get("LANGCHAIN_API_KEY")
os.environ["LANGCHAIN_TRACING_V2"] = "true"

tools = [TavilySearchResults(max_results=1)]
prompt = ""
llm = ChatOpenAI(model="gpt-3.5-turbo")

agent_runnable = create_openai_functions_agent(llm, tools, prompt)

from langchain_core.runnables import RunnablePassthrough
from langchain_core.agents import AgentFinish

agent = RunnablePassthrough.assign(
    agent_outcome = agent_runnable
)

def execute_tools(data):
    agent_action = data.pop('agent_outcome')
    tool_to_use = {t.name: t for t in tools}[agent_action.tool]
    observation = tool_to_use.invoke(agent_action.tool_input)
    data['intermediate_steps'].append((agent_action, observation))
    return data

def should_continue(data):
    if isinstance(data['agent_outcome'], AgentFinish):
        return "exit"
    else:
        return "continue"

# Graphの作成
from langgraph.graph import END, Graph
workflow = Graph()

# Graphを構成するnodeを追加
workflow.add_node("agent", agent)
workflow.add_node("tools", execute_tools)

# グラフの開始ノードを設定するエッジ
workflow.set_entry_point("agent")

# 条件に基づいて遷移先のノードを決定するエッジ
# (conditional_edgesを使うことで、条件分岐が可能となる)
# agent -> continue -> tools
#       -> exit -> END
workflow.add_conditional_edges(
    "agent", # start node
    should_continue,
    # should_continueの内容による条件分岐
    {
        "continue": "tools",
        "exit": END
    }
)

#　特定のノードから無条件で遷移するエッジを指定
# tools -> agent
workflow.add_edge('tools', 'agent')


#　コンパイルを行なってインスタンスへと変換
chain = workflow.compile()

chain.invoke({"input": "AGIとは何かについて簡潔に教えてください", "intermediate_steps": []})

from IPython.display import Image, display

display(Image(chain.get_graph().draw_mermaid_png()))

CheckPointを使う

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# グラフを設定
graph = StateGraph(State)
graph.add_node("add_message", add_message)
graph.add_node("llm_response", llm_response)

graph.set_entry_point("add_message")
graph.add_edge("add_message", "llm_response")
graph.add_edge("llm_response", END)

# チェックポインターを設定
checkpointer = MemorySaver()

# グラフをコンパイル
compiled_graph = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "example-1"}}
user_query = State(query="「M1:犯人の手がかりを見つける」というミッションで、1.ナイフを見つける 2.隠された日記を見つける というタスクを完了しました")
first_response = compiled_graph.invoke(user_query, config)
first_response

for checkpoint in checkpointer.list(config):
    print(checkpoint)

参考にした記事など