Create New Subgraph Agent in guv3

Overview

guv3 uses a subgraph agent pattern where each specialized assistant is an independent LangGraph StateGraph with its own internal loop. Creating a new agent requires modifying 8+ files in a specific order.

File Structure to Create

src/guv3/gu/agents/{agent_name}/
├── __init__.py
├── agent.py                    # Internal StateGraph (subgraph)
└── nodes/
    ├── __init__.py
    ├── node.py                 # Main runnable function
    ├── tools.py                # Tool definitions
    └── prompt.py               # Prompt templates

src/guv3/gu/graphs/{agent_name}.py   # Graph integration with main builder

Files to Modify

src/guv3/gu/state/state.py                    # Add to dialog_state Literal
src/guv3/gu/core/assistant.py                  # Add delegation class + router mappings (4 places)
src/guv3/gu/graphs/primary_assistant.py        # Import graph integration + add to route_to_workflow
src/guv3/gu/nodes/primary_assistant/node.py    # Import delegation class + add to primary_assistant_tools

Step-by-Step Instructions

Step 1: Create nodes/prompt.py

from langchain_core.prompts import ChatPromptTemplate

{AGENT_NAME_UPPER}_PROMPT = """
# IDENTIDAD Y ROL
Eres **Gu**, asistente inmobiliario virtual especializado en **{specialty}**.
Trabajas para la oficina inmobiliaria de **{{owner_name}}**.

# DELEGACION SILENCIOSA - REGLA CRITICA
- NUNCA digas "Te paso con...", "Te comunico con..."
- Cuando delegues, ejecuta la herramienta SIN texto adicional

# RESUMEN DE CONVERSACIONES ANTERIORES
{{context_history}}

# INSTRUCCIONES
[Your specific instructions here]

# HERRAMIENTAS DISPONIBLES
[Describe each tool and when to use it]
"""

def create_{agent_name}_prompt(**kwargs) -> ChatPromptTemplate:
    prompt_content = {AGENT_NAME_UPPER}_PROMPT.format(**kwargs) if kwargs else {AGENT_NAME_UPPER}_PROMPT
    return ChatPromptTemplate.from_messages([
        ("system", prompt_content),
        ("placeholder", "{messages}"),
    ])

Key rules:

• Use double braces {{variable}} for LangChain template variables
• Always include ("placeholder", "{messages}") as the second message
• Include "DELEGACION SILENCIOSA" section in all prompts

Step 2: Create nodes/tools.py

from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
from loguru import logger
from gu.helpers.helpers import get_user, get_bot

@tool
def your_tool_name(
    param1: str,
    param2: str = None,
    *,
    config: RunnableConfig,
):
    """Description of what this tool does.

    Args:
        param1: Description
        param2: Description
    """
    try:
        user = get_user(config)
        if not user:
            logger.error("[your_tool_name] No user found")
            return ""
        bot = get_bot(config)

        # Tool logic here

        return "Response message"
    except Exception as e:
        logger.error(f"Error in your_tool_name: {e}")
        return "Error message"

tools = [
    your_tool_name,
]

Critical rules:

• config: RunnableConfig MUST be keyword-only (after *)
• Always return strings, never raise exceptions
• Always validate user with get_user(config)
• Export a tools list at the bottom

Step 3: Create nodes/node.py

import functools
from loguru import logger
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import AIMessage
from gu.state.state import State
from gu.core.assistant import agent_invoke
from gu.helpers.helpers import get_user
from gu.agents.{agent_name}.nodes.tools import tools
from gu.agents.{agent_name}.nodes.prompt import create_{agent_name}_prompt


def {agent_name}_fn(state: State, config: RunnableConfig):
    try:
        user_info = state.get("user_info")
        if not user_info:
            user_info = get_user(config)
            if not user_info:
                logger.error("[{agent_name}_fn] No user found")
                return {"messages": AIMessage(content="")}

        prompt = create_{agent_name}_prompt(
            # Pass your template variables here
        )

        return agent_invoke(
            prompt,
            tools,
            state,
            config,
            "{agent_name}",
        )
    except Exception as e:
        logger.error(f"Error in {agent_name}_fn: {e}")
        return


{agent_name}_runnable = functools.partial({agent_name}_fn)

Critical rules:

• Use functools.partial() to create the runnable export
• The name string in agent_invoke() MUST match the agent name exactly
• CompleteOrEscalate is added automatically by agent_invoke — do NOT add it to tools

Step 4: Create agent.py (the subgraph)

from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import tools_condition
from langchain_core.runnables import RunnableConfig
from loguru import logger
from gu.state.state import State
from gu.utils.utilities import create_tool_node_with_fallback
from gu.agents.{agent_name}.nodes.node import {agent_name}_runnable
from gu.agents.{agent_name}.nodes.tools import tools as executable_tools

DELEGATION_TOOLS = {"CompleteOrEscalate"}


def is_delegation_tool_call(state: State) -> bool:
    messages = state.get("messages", [])
    if not messages:
        return False
    last_message = messages[-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        for tool_call in last_message.tool_calls:
            if tool_call.get("name", "") in DELEGATION_TOOLS:
                logger.info(f"[{agent_name}] Delegation tool detected: {tool_call.get('name')}")
                return True
    return False


builder = StateGraph(State)

builder.add_node("{agent_name}", {agent_name}_runnable)
builder.add_node(
    "{agent_name}_tools",
    create_tool_node_with_fallback(executable_tools)
)


def route_{agent_name}(
    state: State,
    config: RunnableConfig,
) -> Literal["{agent_name}_tools", "__end__"]:
    if is_delegation_tool_call(state):
        logger.info("[{agent_name}] Ending subgraph for delegation")
        return END
    route = tools_condition(state)
    if route == END:
        return END
    return "{agent_name}_tools"


builder.add_edge(START, "{agent_name}")
builder.add_conditional_edges("{agent_name}", route_{agent_name})
builder.add_edge("{agent_name}_tools", "{agent_name}")

{agent_name}_agent = builder.compile()

This file is 95% boilerplate. Only replace {agent_name} throughout.

Step 5: Create __init__.py files

agents/{agent_name}/__init__.py:

from gu.agents.{agent_name}.agent import {agent_name}_agent

__all__ = ["{agent_name}_agent"]

agents/{agent_name}/nodes/__init__.py:

from gu.agents.{agent_name}.nodes.node import {agent_name}_runnable
from gu.agents.{agent_name}.nodes.tools import tools as {agent_name}_tools

__all__ = ["{agent_name}_runnable", "{agent_name}_tools"]

Step 6: Create graph integration graphs/{agent_name}.py

from typing import Literal
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import AIMessage
from loguru import logger
from gu.agents.{agent_name}.agent import {agent_name}_agent
from gu.utils.utilities import create_entry_node
from gu.graphs.init_graph import builder
from gu.state.state import State

DELEGATION_TOOLS = {"CompleteOrEscalate"}


def route_after_subgraph(
    state: State,
    config: RunnableConfig,
) -> Literal["leave_skill", "guard_translator"]:
    messages = state.get("messages", [])
    if not messages:
        return "guard_translator"
    last_message = messages[-1]
    if isinstance(last_message, AIMessage) and hasattr(last_message, "tool_calls") and last_message.tool_calls:
        for tool_call in last_message.tool_calls:
            if tool_call.get("name", "") in DELEGATION_TOOLS:
                logger.info(f"[{agent_name}] Routing to leave_skill")
                return "leave_skill"
    return "guard_translator"


{agent_name}_possible_routes = ["leave_skill", "guard_translator"]

builder.add_node("enter_{agent_name}", create_entry_node("Agent Display Name", "{agent_name}"))
builder.add_node("{agent_name}", {agent_name}_agent)
builder.add_edge("enter_{agent_name}", "{agent_name}")
builder.add_conditional_edges("{agent_name}", route_after_subgraph, {agent_name}_possible_routes)

Step 7: Modify existing files (4 changes)

7a. state.py — Add to dialog_state Literal:

# In State class, add to the Literal list:
"{agent_name}",

7b. core/assistant.py — 4 places to modify:

1. Add delegation class:

class To{AgentNamePascal}Assistant(BaseModel):
    """Description of when to delegate to this assistant."""
    query: str

2. Add to Router.POSSIBLE_ROUTES:

"enter_{agent_name}",

3. Add to Router.map_tool_calls():

To{AgentNamePascal}Assistant.__name__: "enter_{agent_name}",

4. Add to Router.get_role_tool_mapping():

"{agent_name}": To{AgentNamePascal}Assistant.__name__,

5. Add to get_agents_tools() — both the list and the tool_mapping dict:

# In agents_tools list:
To{AgentNamePascal}Assistant,
# In tool_mapping dict:
"enter_{agent_name}": To{AgentNamePascal}Assistant,

7c. graphs/primary_assistant.py:

# Add import at top:
import gu.graphs.{agent_name}  # noqa: F401

# Add to route_to_workflow() Literal type hint:
"{agent_name}",

7d. nodes/primary_assistant/node.py:

# Add to imports:
from gu.core.assistant import (
    ...,
    To{AgentNamePascal}Assistant,
)

# Add to primary_assistant_tools list:
To{AgentNamePascal}Assistant,

Naming Conventions

Component	Format	Example
Folder	snake_case	visit_tracker_assistant
Function	{name}_fn	visit_tracker_assistant_fn
Runnable	{name}_runnable	visit_tracker_assistant_runnable
Compiled graph	{name}_agent	visit_tracker_assistant_agent
Graph file	graphs/{name}.py	graphs/visit_tracker_assistant.py
Dialog state	"{name}"	"visit_tracker_assistant"
Entry node	"enter_{name}"	"enter_visit_tracker_assistant"
Tools node	"{name}_tools"	"visit_tracker_assistant_tools"
Delegation class	To{PascalCase}Assistant	ToVisitTrackerAssistant

Checklist

• Created agents/{name}/nodes/prompt.py
• Created agents/{name}/nodes/tools.py
• Created agents/{name}/nodes/node.py
• Created agents/{name}/agent.py
• Created agents/{name}/__init__.py and agents/{name}/nodes/__init__.py
• Created graphs/{name}.py
• Added to state.py dialog_state Literal
• Added delegation class to core/assistant.py
• Added to Router.POSSIBLE_ROUTES
• Added to Router.map_tool_calls()
• Added to Router.get_role_tool_mapping()
• Added to get_agents_tools()
• Imported in graphs/primary_assistant.py
• Added to route_to_workflow() type hint
• Added to nodes/primary_assistant/node.py imports and tools list