LangGraph
Last updated
Was this helpful?
Last updated
Was this helpful?
pip install langgraph langchain_openai
pip install arize-phoenix arize-otel
pip install openinference-instrumentation-openai openinference-instrumentation-langchain
from arize.otel import register
tracer_provider = register(
space_id="",
api_key="",
project_name="langgraph-tracing",
)
from openinference.instrumentation.langchain import LangChainInstrumentor
LangChainInstrumentor().instrument(tracer_provider=tracer_provider)
from typing import Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
import os
os.environ["OPENAI_API_KEY"] = "" # Fill in here
@tool
def search(query: str):
if "sf" in query.lower() or "san francisco" in query.lower():
return "It's 60 degrees and foggy."
return "It's 90 degrees and sunny."
def should_continue(state: MessagesState) -> Literal["tools", END]:
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
return "tools"
return END
def call_model(state: MessagesState):
messages = state["messages"]
response = model.invoke(messages)
return {"messages": [response]}
tools = [search]
tool_node = ToolNode(tools)
model = ChatOpenAI(temperature=0).bind_tools(tools)
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
"agent",
should_continue,
)
workflow.add_edge("tools", "agent")
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)
final_state = app.invoke(
{"messages": [HumanMessage(content="what is the weather in sf")]},
config={"configurable": {"thread_id": 42}},
)
final_state["messages"][-1].content
The same instrumentations for .