langgraph-compass 1.0.1

Intelligent follow-up question generation for LangGraph agents

Compass

Intelligent follow-up question generation for LangGraph agents.

Transform your AI agents from reactive responders into proactive conversational partners. Compass generates contextually relevant follow-up questions that guide users toward productive outcomes. Built and battle-tested at Cisco.

Why Follow-Ups Matter

Great conversations don't just answer questions—they anticipate what comes next. Contextually relevant follow-ups:

• Increase engagement — Users who receive thoughtful follow-ups continue conversations 2-3x longer
• Surface hidden needs — Users often don't know what to ask; good follow-ups guide discovery
• Catch misunderstandings early — Clarifying questions prevent compounding errors
• Drive business outcomes — In support, sales, and onboarding, follow-ups guide users toward resolution

The Problem with DIY

The naive approach looks simple:

# Just add it to your prompt, right?
system_prompt = """
Answer the user's question.
At the end, suggest a follow-up question.
"""

But this breaks down in enterprise settings:

Problem	What Happens
Unbounded suggestions	Model suggests follow-ups your agent can't actually fulfill
Guardrail-unaware	Suggests follow-ups after guardrails block a request
No novelty tracking	Repeats similar suggestions across conversation turns
Coupled to response	Can't run in parallel—adds latency to every response
Hard to tune	Logic buried in prompts, not configurable or testable
One-size-fits-all	Can't adapt to user workflows or preferences

The Compass Solution

Compass is a dedicated LangGraph node that gives you granular control over follow-up generation:

from compass import CompassNode

compass = CompassNode(
    model=your_llm,
    strategy="adaptive",      # Automatically chooses clarifying/exploratory/deepening
    trigger=DefaultTriggerPolicy(
        skip_on_guardrail=True,   # Don't suggest after blocks
        min_response_length=50,   # Skip for short responses
    ),
)

# Add to your graph - that's it
builder.add_node("compass", compass)
builder.add_edge("agent", "compass")

Feature	What It Enables
Configurable Triggers	Skip follow-ups on guardrails, short responses, or custom conditions
Capability Grounding	Only suggest paths your agent can actually handle via ExampleRetriever
Novelty Filtering	Automatically avoids repetitive suggestions across turns
Parallel Execution	Run via Send() for zero added latency
Separate Model	Use a smaller/cheaper model for follow-ups
Full Tracing	Every decision logged in LangSmith

Installation

pip install langgraph-compass

Quickstart

from compass import CompassNode
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# 1. Create the node
compass = CompassNode(model=ChatOpenAI())

# 2. Add to your graph
builder = StateGraph(MessagesState)
builder.add_node("agent", your_agent)
builder.add_node("compass", compass)
builder.add_edge(START, "agent")
builder.add_edge("agent", "compass")
builder.add_edge("compass", END)

graph = builder.compile()

# 3. Suggestions appear in state
result = graph.invoke({"messages": [HumanMessage(content="What's the weather in SF?")]})
print(result["compass_suggestions"])
# ["Would you like me to compare this with yesterday's weather?"]

How It Works

flowchart LR
    subgraph YourGraph[Your LangGraph]
        A[Agent Node] --> C[CompassNode]
    end

    subgraph CompassNode[CompassNode]
        direction TB
        T[Trigger Policy] -->|should_trigger?| G[Generator]
        G --> R[Ranker]
        R --> O[suggestions]
    end

    C --> E[END]

1. Trigger Policy — Decides if follow-ups should be generated (skip on guardrails, short responses, etc.)
2. Generator — Creates candidate questions using your chosen strategy
3. Ranker — Filters for novelty and diversity against conversation history

Strategies

Strategy	When to Use	Example Output
adaptive	Default. Automatically chooses best approach.	Varies by context
clarifying	When responses might have ambiguities	"Would you like me to clarify the difference between X and Y?"
exploratory	When opening new directions	"Interested in exploring how this applies to Z?"
deepening	When more detail would help	"Should I dive deeper into the technical details?"

Advanced Patterns

Zero-Latency Parallel Execution

Run Compass simultaneously with your agent—suggestions ready the moment your response is:

from langgraph.types import Send

def route_parallel(state):
    return [
        Send("agent", state),
        Send("compass", state),  # Runs at the same time
    ]

builder.add_conditional_edges("plan", route_parallel)

Organic Response Injection

Make follow-ups feel natural by weaving them into your agent's response:

from compass import get_compass_instruction

def synthesize(state):
    instruction = get_compass_instruction(state)
    # Returns: "At the end of your response, naturally include..."

    prompt = f"Answer the question. {instruction}"
    return llm.invoke(prompt)

Custom Trigger Policies

Control exactly when follow-ups are generated:

class MyTriggerPolicy:
    def should_trigger(self, state: dict) -> bool:
        # Skip for greetings
        if state.get("intent") == "greeting":
            return False
        # Skip if user asked not to
        if "no follow-up" in state.get("query", "").lower():
            return False
        return True

compass = CompassNode(model=llm, trigger=MyTriggerPolicy())

Capability-Bounded Suggestions

Only suggest follow-ups your agent can actually fulfill.

class CapabilityRetriever:
    """Ground follow-ups in your agent's actual capabilities."""

    def __init__(self, capability_store):
        self.store = capability_store  # Your vector store of supported queries

    def retrieve(self, query: str, k: int = 5) -> list[str]:
        # MMR: relevant AND diverse paths your agent can handle
        return self.store.max_marginal_relevance_search(
            query, k=k, fetch_k=20, lambda_mult=0.7
        )

compass = CompassNode(
    model=llm,
    example_retriever=CapabilityRetriever(capability_store)
)

This prevents the model from suggesting follow-ups about features that don't exist or data you don't have.

Cost-Effective Model Selection

Since follow-up generation is decoupled from your main agent, you can use a smaller, faster, cheaper model:

from langchain_openai import ChatOpenAI

# Your main agent uses a powerful model
agent_model = ChatOpenAI(model="gpt-5.2")

# Compass can use a lightweight model — follow-ups don't need heavy reasoning
compass_model = ChatOpenAI(model="gpt-5-nano")  # 10x cheaper, faster

compass = CompassNode(model=compass_model)

This separation means follow-up generation doesn't inflate your costs or add latency to your critical path.

Integrations

LangSmith Observability

Compass automatically integrates with LangSmith tracing. Every follow-up generation is logged with full context:

# Just set your environment variables
# LANGCHAIN_TRACING_V2=true
# LANGCHAIN_API_KEY=your-key

# Compass nodes appear in your LangSmith traces automatically
# You can see: trigger decisions, generated candidates, ranking results

In LangSmith, you'll see:

• Whether the trigger fired and why
• The strategy used
• All candidate questions generated
• Which were filtered by the ranker
• The final suggestions returned

User Feedback Loop

Track which follow-ups users engage with using LangSmith feedback:

from langsmith import Client

client = Client()

# After user clicks/uses a follow-up
client.create_feedback(
    run_id=compass_run_id,  # From your trace
    key="followup_clicked",
    score=1,
    comment="User clicked: 'Would you like more details?'"
)

# Use this data to improve your prompts and strategies

Workflow Personalization with LangMem

Integrate with langmem to learn user workflows and suggest their typical next steps:

from langgraph.store.memory import InMemoryStore

store = InMemoryStore(index={"dims": 1536, "embed": "openai:text-embedding-3-small"})

class WorkflowRetriever:
    """Suggest follow-ups based on user's typical workflow patterns."""

    def __init__(self, store, user_id: str):
        self.store = store
        self.user_id = user_id

    def retrieve(self, query: str, k: int = 3) -> list[str]:
        # Search user's workflow history
        memories = self.store.search(
            ("workflows", self.user_id),
            query=query,
        )
        # Return their typical next actions
        # e.g., user always does: check metrics → review alerts → update dashboard
        return [m.content.get("typical_next_step") for m in memories if m.content]

# Compass learns to fit into each user's personal workflow
compass = CompassNode(
    model=llm,
    example_retriever=WorkflowRetriever(store, user_id="user-123"),
)

Instead of generic "Anything else?", Compass can suggest "Ready to review alerts?" because it knows that's what this user typically does next.

Configuration Reference

CompassNode

CompassNode(
    model: BaseChatModel,
    *,
    trigger: TriggerPolicy | None = None,          # When to generate (default: DefaultTriggerPolicy)
    strategy: Literal["adaptive", "clarifying", "exploratory", "deepening"] = "adaptive",
    max_suggestions: int = 1,                       # How many to return
    starters: list[str] | None = None,              # Approved starter phrases
    output_key: str = "compass_suggestions",        # State key for output
    inject_into_messages: bool = False,             # Also add to messages
    query_key: str | None = None,                   # State key for query (auto-detected)
    response_key: str | None = None,                # State key for response (auto-detected)
    messages_key: str = "messages",                 # State key for message history
    example_retriever: ExampleRetriever | None = None,  # Domain examples
    generate_candidates: int = 3,                   # Candidates before ranking
)

DefaultTriggerPolicy

DefaultTriggerPolicy(
    *,
    skip_on_guardrail: bool = True,                 # Skip if guardrails fired
    guardrail_keys: list[str] | None = None,        # Your guardrail state keys
                                                    # Default: ["input_guardrail_fired", "output_guardrail_fired"]
    skip_classifications: list[str] | None = None,  # Query types to skip
    require_agent_response: bool = True,            # Need substantive response
    min_response_length: int = 50,                  # Minimum chars to trigger
    custom_skip_keys: list[str] | None = None,      # Additional state keys to check
)

Example with custom guardrail keys:

# Match your guardrails implementation
trigger = DefaultTriggerPolicy(
    guardrail_keys=["off_topic_detected", "pii_flagged", "toxicity_blocked"]
)

Helper Functions

from compass import get_compass_instruction, extract_previous_followups

# Get instruction for prompt injection
instruction = get_compass_instruction(state, suggestion_key="compass_suggestions")

# Extract previous follow-ups for custom novelty logic
previous = extract_previous_followups(messages, starters=None, max_history=5)

Examples

See the examples/ directory:

• quickstart.py — Minimal working example
• parallel_execution.py — Zero-latency pattern with Send()
• inject_into_response.py — Organic follow-up injection

Development

See CONTRIBUTING.md for development setup and guidelines.

# Quick setup
uv sync --all-extras
uv run pre-commit install

# Run tests
uv run pytest

# Lint & type check
uv run ruff check src tests
uv run mypy src

License

Apache 2.0