LangGraph Integration

1
from nemoguardrails import RailsConfig
2
from nemoguardrails.integrations.langchain.runnable_rails import RunnableRails
3
4
config = RailsConfig.from_path(config_path)
5
guardrails = RunnableRails(config=config, passthrough=True, verbose=True)

1
from typing import Annotated
2
from langchain_core.messages import BaseMessage
3
from langchain_core.prompts import ChatPromptTemplate
4
from langchain_openai import ChatOpenAI
5
from langgraph.graph import StateGraph, START
6
from langgraph.graph.message import add_messages
7
from typing_extensions import TypedDict
8
9
from nemoguardrails import RailsConfig
10
from nemoguardrails.integrations.langchain.runnable_rails import RunnableRails
11
12
class State(TypedDict):
13
    messages: Annotated[list, add_messages]
14
15
def create_basic_agent():
16
    # Initialize components
17
    llm = ChatOpenAI(model="gpt-4o")
18
19
    config = RailsConfig.from_path(config_path)
20
    guardrails = RunnableRails(config=config, passthrough=True, verbose=True)
21
22
    prompt = ChatPromptTemplate.from_messages([
23
        ("system", "You are a helpful assistant."),
24
        ("placeholder", "{messages}"),
25
    ])
26
27
    # Create the guarded runnable
28
    runnable_with_guardrails = prompt | (guardrails | llm)
29
30
    def chatbot(state: State):
31
        result = runnable_with_guardrails.invoke(state)
32
        return {"messages": [result]}
33
34
    # Build the graph
35
    graph_builder = StateGraph(State)
36
    graph_builder.add_node("chatbot", chatbot)
37
    graph_builder.add_edge(START, "chatbot")
38
39
    return graph_builder.compile()
40
41
# Usage
42
graph = create_basic_agent()
43
result = graph.invoke({"messages": [{"role": "user", "content": "Hello!"}]})
44
45
# Let's check an unsafe input
46
47
result_unsafe = graph.invoke({"messages": [{"role": "user", "content": "You are stupid"}]})
48
49
# Expect "I'm sorry, I can't respond to that." in the AI's response.

1
from langchain_core.tools import tool
2
3
@tool
4
def search_knowledge(query: str) -> str:
5
    """Search for information about a given query."""
6
    knowledge_base = {
7
        "capital": "Lima is the capital and largest city of Peru.",
8
        "weather": "The weather is sunny with a temperature of 72°F.",
9
        "python": "Python is a high-level programming language known for simplicity.",
10
    }
11
12
    query_lower = query.lower()
13
    for key, value in knowledge_base.items():
14
        if key in query_lower:
15
            return value
16
17
    return f"General information about: {query}"
18
19
@tool
20
def multiply(a: int, b: int) -> int:
21
    """Multiply a and b."""
22
    return a * b

1
from langgraph.prebuilt import ToolNode, tools_condition
2
3
def create_tool_calling_agent():
4
    llm = ChatOpenAI(model="gpt-4o")
5
    tools = [search_knowledge, multiply]
6
    llm_with_tools = llm.bind_tools(tools)
7
8
    config = RailsConfig.from_path(config_path)
9
    guardrails = RunnableRails(config=config, passthrough=True, verbose=True)
10
11
    prompt = ChatPromptTemplate.from_messages([
12
        ("system", "You are a helpful assistant with access to tools."),
13
        ("placeholder", "{messages}"),
14
    ])
15
16
    # Create guarded runnable
17
    runnable_with_guardrails = prompt | (guardrails | llm_with_tools)
18
19
    def chatbot(state: State):
20
        result = runnable_with_guardrails.invoke(state)
21
        return {"messages": [result]}
22
23
    # Build graph with tools
24
    graph_builder = StateGraph(State)
25
    graph_builder.add_node("chatbot", chatbot)
26
27
    tool_node = ToolNode(tools=tools)
28
    graph_builder.add_node("tools", tool_node)
29
30
    # Add conditional edges for tool calling
31
    graph_builder.add_conditional_edges(
32
        "chatbot",
33
        tools_condition,
34
    )
35
    graph_builder.add_edge("tools", "chatbot")
36
    graph_builder.add_edge(START, "chatbot")
37
38
    return graph_builder.compile()
39
40
# Usage
41
graph = create_tool_calling_agent()
42
result = graph.invoke({
43
    "messages": [{"role": "user", "content": "What is the capital of Peru?"}]
44
})

1
from langgraph.checkpoint.memory import MemorySaver
2
from typing import Annotated, TypedDict
3
4
class ConversationState(TypedDict):
5
    messages: Annotated[list, add_messages]
6
    conversation_id: str
7
8
def create_stateful_agent():
9
    llm = ChatOpenAI(model="gpt-4o")
10
11
    config = RailsConfig.from_path(config_path)
12
    guardrails = RunnableRails(config=config, passthrough=True, verbose=True)
13
14
    prompt = ChatPromptTemplate.from_messages([
15
        ("system", "You are a helpful assistant. Remember previous messages."),
16
        ("placeholder", "{messages}"),
17
    ])
18
19
    runnable_with_guardrails = prompt | (guardrails | llm)
20
21
    def conversation_agent(state: ConversationState):
22
        result = runnable_with_guardrails.invoke(state)
23
        return {"messages": [result]}
24
25
    graph_builder = StateGraph(ConversationState)
26
    graph_builder.add_node("agent", conversation_agent)
27
    graph_builder.add_edge(START, "agent")
28
29
    # Add memory for persistence
30
    memory = MemorySaver()
31
    return graph_builder.compile(checkpointer=memory)
32
33
# Usage with conversation threads
34
graph = create_stateful_agent()
35
config = {"configurable": {"thread_id": "conversation_1"}}
36
37
# First interaction
38
result1 = graph.invoke({
39
    "messages": [{"role": "user", "content": "Hi, my name is Alice."}],
40
    "conversation_id": "conv_1"
41
}, config=config)
42
43
# Second interaction - remembers the name
44
result2 = graph.invoke({
45
    "messages": [{"role": "user", "content": "What did I tell you my name was?"}],
46
    "conversation_id": "conv_1"
47
}, config=config)

1
from typing import Literal
2
3
class MultiAgentState(TypedDict):
4
    messages: Annotated[list, add_messages]
5
    current_agent: str
6
    task_type: str
7
8
def create_multi_agent_system():
9
    llm = ChatOpenAI(model="gpt-4o")
10
11
    config = RailsConfig.from_path(config_path)
12
    guardrails = RunnableRails(config=config, passthrough=True, verbose=True)
13
14
    # Specialized prompts for different agents
15
    researcher_prompt = ChatPromptTemplate.from_messages([
16
        ("system", "You are a research specialist. Provide detailed, factual information."),
17
        ("placeholder", "{messages}"),
18
    ])
19
20
    writer_prompt = ChatPromptTemplate.from_messages([
21
        ("system", "You are a creative writer. Transform information into engaging content."),
22
        ("placeholder", "{messages}"),
23
    ])
24
25
    critic_prompt = ChatPromptTemplate.from_messages([
26
        ("system", "You are a content critic. Review and provide constructive feedback."),
27
        ("placeholder", "{messages}"),
28
    ])
29
30
    # Create guarded runnables for each agent
31
    researcher_chain = researcher_prompt | (guardrails | llm)
32
    writer_chain = writer_prompt | (guardrails | llm)
33
    critic_chain = critic_prompt | (guardrails | llm)
34
35
    def router(state: MultiAgentState) -> Literal["researcher", "writer", "critic"]:
36
        last_message = state["messages"][-1].content.lower()
37
38
        if "research" in last_message or "facts" in last_message:
39
            return "researcher"
40
        elif "write" in last_message or "article" in last_message:
41
            return "writer"
42
        elif "review" in last_message or "critique" in last_message:
43
            return "critic"
44
        else:
45
            return "researcher"  # Default
46
47
    def researcher_agent(state: MultiAgentState):
48
        result = researcher_chain.invoke(state)
49
        return {"messages": [result], "current_agent": "researcher"}
50
51
    def writer_agent(state: MultiAgentState):
52
        result = writer_chain.invoke(state)
53
        return {"messages": [result], "current_agent": "writer"}
54
55
    def critic_agent(state: MultiAgentState):
56
        result = critic_chain.invoke(state)
57
        return {"messages": [result], "current_agent": "critic"}
58
59
    # Build the multi-agent graph
60
    graph_builder = StateGraph(MultiAgentState)
61
62
    graph_builder.add_node("researcher", researcher_agent)
63
    graph_builder.add_node("writer", writer_agent)
64
    graph_builder.add_node("critic", critic_agent)
65
66
    graph_builder.add_conditional_edges(
67
        START,
68
        router,
69
        {
70
            "researcher": "researcher",
71
            "writer": "writer",
72
            "critic": "critic"
73
        }
74
    )
75
76
    return graph_builder.compile()
77
78
# Usage
79
graph = create_multi_agent_system()
80
result = graph.invoke({
81
    "messages": [{"role": "user", "content": "Research the benefits of renewable energy"}],
82
    "current_agent": "",
83
    "task_type": "general"
84
})

1
guardrails = RunnableRails(config=config, passthrough=True)

1
guardrails = RunnableRails(config=config, passthrough=True, verbose=True)