diff --git a/langgraph.json b/langgraph.json index 2d7c855..dc128d2 100644 --- a/langgraph.json +++ b/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" + ] +} \ No newline at end of file diff --git a/my_agent/agent_guardrail_after.py b/my_agent/agent_guardrail_after.py new file mode 100644 index 0000000..18c73b9 --- /dev/null +++ b/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() diff --git a/my_agent/agent_guardrail_before.py b/my_agent/agent_guardrail_before.py new file mode 100644 index 0000000..be9db55 --- /dev/null +++ b/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() diff --git a/my_agent/utils/nodes.py b/my_agent/utils/nodes.py index c44ba6e..3e0ea54 100644 --- a/my_agent/utils/nodes.py +++ b/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) \ No newline at end of file +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']} \ No newline at end of file diff --git a/my_agent/utils/state.py b/my_agent/utils/state.py index 7d21270..63208d0 100644 --- a/my_agent/utils/state.py +++ b/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