You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
from langchain_core.tools import tool
from langgraph.constants import END, START
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.pydantic_v1 import BaseModel, Field
class State(TypedDict):
messages: Annotated[list, add_messages]
reacted with thumbs up emoji reacted with thumbs down emoji reacted with laugh emoji reacted with hooray emoji reacted with confused emoji reacted with heart emoji reacted with rocket emoji reacted with eyes emoji
Uh oh!
There was an error while loading. Please reload this page.
-
from typing import Annotated
from langchain_core.tools import tool
from langgraph.constants import END, START
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.pydantic_v1 import BaseModel, Field
class State(TypedDict):
messages: Annotated[list, add_messages]
@tool
def add(first_int: int, second_int: int) -> int:
"""
Add two integers together.
@tool
def multiply(first_int: int, second_int: int) -> int:
"""
Multiply two integers together in more fast way.
class AddTask(BaseModel):
"""Add two integers together."""
class MultiplyTask(BaseModel):
"""multiply two integers together."""
tools = [AddTask, MultiplyTask]
llm = ChatOpenAI()
llm_with_tools = llm.bind_tools(tools)
def chatbot(state: State):
message = {"messages": [llm_with_tools.invoke(state["messages"])]}
return message
"""
"""
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
add_node = ToolNode(tools=[AddTask])
multiply_node = ToolNode(tools=[MultiplyTask])
graph_builder.add_node("add", add_node)
graph_builder.add_node("multiply", multiply_node)
-> list[Literal["add", "multiply", "end"]]
def route_tools(state: State):
next_node = tools_condition(state)
# If no tools are invoked, return to the user
if next_node == END:
return END
graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges(
"chatbot",
route_tools,
)
graph_builder.add_edge("add", "chatbot")
graph_builder.add_edge("multiply", "chatbot")
memory = SqliteSaver.from_conn_string(":memory:")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "1"}}
print(graph.get_graph().draw_mermaid())
def _print_event(event: dict, _printed: set, max_length=1500):
message = event.get("messages")
if message:
if isinstance(message, list):
message = message[-1]
if message.id not in _printed:
msg_repr = message.pretty_repr(html=True)
if len(msg_repr) > max_length:
msg_repr = msg_repr[:max_length] + " ... (truncated)"
print(msg_repr)
_printed.add(message.id)
while True:
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
_printed = set()
events = graph.stream(
{"messages": ("user", user_input)}, config, stream_mode="values"
)
for event in events:
_print_event(event, _printed)
Beta Was this translation helpful? Give feedback.
All reactions