Create langgraph_ai_agent.py

Author: amosgyamfi

AI/Multi-Agent-AI/langgraph_ai_agent.py

@@ -0,0 +1,87 @@
+from typing import Annotated, Literal, TypedDict
+
+from langchain_core.messages import HumanMessage
+from langchain_anthropic import ChatAnthropic
+from langchain_core.tools import tool
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.graph import END, START, StateGraph, MessagesState
+from langgraph.prebuilt import ToolNode
+
+
+# Define the tools for the agent to use
+@tool
+def search(query: str):
+    """Call to surf the web."""
+    # This is a placeholder, but don't tell the LLM that...
+    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."
+
+
+tools = [search]
+
+tool_node = ToolNode(tools)
+
+model = ChatAnthropic(model="claude-3-5-sonnet-20240620", temperature=0).bind_tools(
+    tools
+)
+
+
+# Define the function that determines whether to continue or not
+def should_continue(state: MessagesState) -> Literal["tools", END]:
+    messages = state["messages"]
+    last_message = messages[-1]
+    # If the LLM makes a tool call, then we route to the "tools" node
+    if last_message.tool_calls:
+        return "tools"
+    # Otherwise, we stop (reply to the user)
+    return END
+
+
+# Define the function that calls the model
+def call_model(state: MessagesState):
+    messages = state["messages"]
+    response = model.invoke(messages)
+    # We return a list, because this will get added to the existing list
+    return {"messages": [response]}
+
+
+# Define a new graph
+workflow = StateGraph(MessagesState)
+
+# Define the two nodes we will cycle between
+workflow.add_node("agent", call_model)
+workflow.add_node("tools", tool_node)
+
+# Set the entrypoint as `agent`
+# This means that this node is the first one called
+workflow.add_edge(START, "agent")
+
+# We now add a conditional edge
+workflow.add_conditional_edges(
+    # First, we define the start node. We use `agent`.
+    # This means these are the edges taken after the `agent` node is called.
+    "agent",
+    # Next, we pass in the function that will determine which node is called next.
+    should_continue,
+)
+
+# We now add a normal edge from `tools` to `agent`.
+# This means that after `tools` is called, `agent` node is called next.
+workflow.add_edge("tools", "agent")
+
+# Initialize memory to persist state between graph runs
+checkpointer = MemorySaver()
+
+# Finally, we compile it!
+# This compiles it into a LangChain Runnable,
+# meaning you can use it as you would any other runnable.
+# Note that we're (optionally) passing the memory when compiling the graph
+app = workflow.compile(checkpointer=checkpointer)
+
+# Use the Runnable
+final_state = app.invoke(
+    {"messages": [HumanMessage(content="what is the weather in sf")]},
+    config={"configurable": {"thread_id": 42}},
+)
+final_state["messages"][-1].content