cr

langgraph.json

@@ -1,7 +1,16 @@
 {
-  "dependencies": ["./my_agent"],
+  "dockerfile_lines": [],
   "graphs": {
-    "agent": "./my_agent/agent.py:graph"
+    "weather_agent": "./my_agent/agent.py:graph",
+    "guardrail_before": "./my_agent/agent_guardrail_before.py:graph",
+    "guardrail_after": "./my_agent/agent_guardrail_after.py:graph"
   },
-  "env": ".env"
-}
+  "env": [
+    "OPENAI_API_KEY",
+    "TAVILY_API_KEY"
+  ],
+  "python_version": "3.11",
+  "dependencies": [
+    "./my_agent"
+  ]
+}

my_agent/agent_guardrail_after.py

@@ -0,0 +1,66 @@
+from typing import TypedDict, Literal
+
+from langgraph.graph import StateGraph, END
+from my_agent.utils.nodes import call_model, should_continue, tool_node, english_guardrail
+from my_agent.utils.state import AgentGuardrailAfterState
+
+# Define logic to determine whether question is about the weather
+def response_in_english(state: AgentGuardrailAfterState) -> Literal['hardcoded_response', END]:
+    if not state['is_english']:
+        return "hardcoded_response"
+    else:
+        return END
+
+def hardcoded_response(state):
+    return {"messages": [{"role": "assistant", "content": "Unable to process question"}]}
+
+
+# Define the config
+class GraphConfig(TypedDict):
+    model_name: Literal["anthropic", "openai"]
+
+
+# Define a new graph
+workflow = StateGraph(AgentGuardrailAfterState, config_schema=GraphConfig)
+
+# Define the two nodes we will cycle between
+workflow.add_node("agent", call_model)
+workflow.add_node("action", tool_node)
+workflow.add_node(english_guardrail)
+workflow.add_node(hardcoded_response)
+
+# Set the entrypoint as `agent`
+# This means that this node is the first one called
+workflow.set_entry_point("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,
+    # Finally we pass in a mapping.
+    # The keys are strings, and the values are other nodes.
+    # END is a special node marking that the graph should finish.
+    # What will happen is we will call `should_continue`, and then the output of that
+    # will be matched against the keys in this mapping.
+    # Based on which one it matches, that node will then be called.
+    {
+        # If `tools`, then we call the tool node.
+        "continue": "action",
+        # Otherwise we finish.
+        "end": "english_guardrail",
+    },
+)
+
+# 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("action", "agent")
+workflow.add_conditional_edges("english_guardrail", response_in_english)
+workflow.add_edge("hardcoded_response", END)
+
+# Finally, we compile it!
+# This compiles it into a LangChain Runnable,
+# meaning you can use it as you would any other runnable
+graph = workflow.compile()

my_agent/agent_guardrail_before.py

@@ -0,0 +1,67 @@
+from typing import TypedDict, Literal
+
+from langgraph.graph import StateGraph, END
+from my_agent.utils.nodes import call_model, should_continue, tool_node, weather_guardrail
+from my_agent.utils.state import AgentGuardrailBeforeState
+
+# Define logic to determine whether question is about the weather
+def is_about_weather(state: AgentGuardrailBeforeState) -> Literal['hardcoded_response', 'agent']:
+    if not state['about_weather']:
+        return "hardcoded_response"
+    else:
+        return "agent"
+
+def hardcoded_response(state):
+    return {"messages": [{"role": "assistant", "content": "sorry I can only answer questions about weather"}]}
+
+
+# Define the config
+class GraphConfig(TypedDict):
+    model_name: Literal["anthropic", "openai"]
+
+
+# Define a new graph
+workflow = StateGraph(AgentGuardrailBeforeState, config_schema=GraphConfig)
+
+# Define the two nodes we will cycle between
+workflow.add_node("agent", call_model)
+workflow.add_node("action", tool_node)
+workflow.add_node(weather_guardrail)
+workflow.add_node(hardcoded_response)
+
+# Set the entrypoint as `agent`
+# This means that this node is the first one called
+workflow.set_entry_point("weather_guardrail")
+
+workflow.add_conditional_edges("weather_guardrail", is_about_weather)
+
+# 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,
+    # Finally we pass in a mapping.
+    # The keys are strings, and the values are other nodes.
+    # END is a special node marking that the graph should finish.
+    # What will happen is we will call `should_continue`, and then the output of that
+    # will be matched against the keys in this mapping.
+    # Based on which one it matches, that node will then be called.
+    {
+        # If `tools`, then we call the tool node.
+        "continue": "action",
+        # Otherwise we finish.
+        "end": END,
+    },
+)
+
+# 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("action", "agent")
+workflow.add_edge("hardcoded_response", END)
+
+# Finally, we compile it!
+# This compiles it into a LangChain Runnable,
+# meaning you can use it as you would any other runnable
+graph = workflow.compile()

my_agent/utils/nodes.py

@@ -3,6 +3,7 @@ from langchain_anthropic import ChatAnthropic
 from langchain_openai import ChatOpenAI
 from my_agent.utils.tools import tools
 from langgraph.prebuilt import ToolNode
+from typing import TypedDict
 
 
 @lru_cache(maxsize=4)
@@ -35,11 +36,45 @@ system_prompt = """Be a helpful assistant"""
 def call_model(state, config):
     messages = state["messages"]
     messages = [{"role": "system", "content": system_prompt}] + messages
-    model_name = config.get('configurable', {}).get("model_name", "anthropic")
+    model_name = config.get('configurable', {}).get("model_name", "openai")
     model = _get_model(model_name)
     response = model.invoke(messages)
     # We return a list, because this will get added to the existing list
     return {"messages": [response]}
 
 # Define the function to execute tools
-tool_node = ToolNode(tools)
+tool_node = ToolNode(tools)
+
+
+about_weather_prompt = """Determine whether the user's most recent question is about weather."""
+
+class AboutWeather(TypedDict):
+    """Is the user's question about weather?"""
+    about_weather: bool
+
+def weather_guardrail(state, config):
+    messages = state["messages"]
+    messages = [{"role": "system", "content": about_weather_prompt}] + messages
+    model_name = config.get('configurable', {}).get("model_name", "openai")
+    model = _get_model(model_name).with_structured_output(AboutWeather)
+    response = model.invoke(messages)
+    # We return a list, because this will get added to the existing list
+    return {"about_weather": response['about_weather']}
+
+
+responds_in_english_prompt = """Determine whether the final assistant response is in English or not."""
+
+
+class IsEnglish(TypedDict):
+    """Is the final assistant response in English?"""
+    is_english: bool
+
+
+def english_guardrail(state, config):
+    messages = state["messages"]
+    messages = [{"role": "system", "content": responds_in_english_prompt}] + messages
+    model_name = config.get('configurable', {}).get("model_name", "openai")
+    model = _get_model(model_name).with_structured_output(IsEnglish)
+    response = model.invoke(messages)
+    # We return a list, because this will get added to the existing list
+    return {"is_english": response['is_english']}

my_agent/utils/state.py

@@ -1,6 +1,14 @@
-from langgraph.graph import add_messages
+from langgraph.graph import add_messages, MessagesState
 from langchain_core.messages import BaseMessage
 from typing import TypedDict, Annotated, Sequence
 
-class AgentState(TypedDict):
-    messages: Annotated[Sequence[BaseMessage], add_messages]
+class AgentState(MessagesState):
+    pass
+
+
+class AgentGuardrailBeforeState(MessagesState):
+    about_weather: bool
+
+
+class AgentGuardrailAfterState(MessagesState):
+    is_english: bool