Building Cyclical Graphs With Conditional Edges

In this lesson we show how to build a small, agentic workflow using LangGraph’s conditional edges to route between execution paths. The demo is intentionally compact: based on the user’s question, the graph will either perform a live web search via Tavily or answer immediately using an LLM. Both routes converge into a final formatting step. You can also generate a visual representation (Mermaid) of the graph to inspect routing and convergence. Key concepts covered:

State-driven routing using conditional edges
Composable nodes that read/write shared typed state
Converging branches into a single terminal formatter
Integrating external APIs (OpenAI Responses API and Tavily search)

Setup and imports

Install dependencies if needed (uncomment the pip line in a fresh environment), then initialize clients and environment variables.

# If needed in a fresh environment, uncomment:
import os
from typing import List, Literal
from typing_extensions import TypedDict, NotRequired

from openai import OpenAI
from tavily import TavilyClient
from langgraph.graph import StateGraph, START, END

OPENAI_MODEL = os.environ.get("OPENAI_MODEL", "gpt-5")
TAVILY_MAX_RESULTS = int(os.environ.get("TAVILY_MAX_RESULTS", "3"))

# Ensure API keys are set in the environment (or set them here for testing)
os.environ.setdefault("OPENAI_API_KEY", "your-openai-api-key-here")
os.environ.setdefault("TAVILY_API_KEY", "your-tavily-api-key-here")

openai_client = OpenAI()  # reads OPENAI_API_KEY from env
tavily_client = TavilyClient(api_key=os.environ.get("TAVILY_API_KEY"))

Make sure OPENAI_API_KEY and TAVILY_API_KEY are available in your environment before running the examples. Leaving keys in source code is not recommended for production.

This example uses:

OpenAI Responses API client for LLM calls.
Tavily client for web search; the exact method name for searching may vary by SDK version—adapt as needed.

Typed shared state

We define a typed AgentState that nodes will read from and write to. LangGraph’s nodes interact through this shared state rather than direct node-to-node parameter passing.

class AgentState(TypedDict):
    question: str
    intent: NotRequired[Literal["search", "answer"]]
    search_results: NotRequired[list[dict]]  # Tavily results
    draft_answer: NotRequired[str]
    final_answer: NotRequired[str]

The state begins with question and accumulates intent, search_results, draft_answer, and final_answer as the graph runs.

Node overview

We implement four nodes:

Node	Purpose	Example output
`classify_intent`	Decide whether the question needs a web search or can be answered from general knowledge.	`{"intent": "search"}`
`search_web`	Call Tavily and normalize results into `title`, `url`, `content`.	`{"search_results": [{...}]}`
`answer_direct`	Ask the LLM (Responses API) to produce a concise draft answer.	`{"draft_answer": "..."}`
`format_output`	Terminal node: format either the LLM draft or the Tavily results into `final_answer`.	`{"final_answer": "..."}`

Use the table above to quickly see responsibilities and expected state writes.

classify_intent

This node returns a single label — "search" or "answer" — which controls conditional routing.

def classify_intent(state: AgentState) -> dict:
    """Classify whether the question needs a web search."""
    prompt = (
        "Classify the user question into exactly one label: 'search' or 'answer'.\n"
        "Use 'search' if the question likely requires up-to-date facts, sources, or web lookup.\n"
        "Use 'answer' if general knowledge is enough.\n\n"
        f"Question: {state['question']}\n"
        "Return ONLY the label.\n"
    )

    resp = openai_client.responses.create(
        model=OPENAI_MODEL,
        instructions="You are a strict classifier.",
        input=prompt,
    )

    # Many client wrappers expose text via `output_text`; fall back defensively.
    label = (getattr(resp, "output_text", None) or "").strip().lower()
    label = "search" if "search" in label else "answer"
    return {"intent": label}

Use a terse classifier prompt to minimize hallucination and to make the decision deterministic. If you want higher fidelity, consider a small validation step after classification.

search_web

This node fetches results from Tavily and normalizes them into a simple list of dictionaries with title, url, and content.

def search_web(state: AgentState) -> dict:
    """Retrieve web results for the question using Tavily."""
    query = state["question"]

    # Adjust this call to match your tavily-python SDK if necessary.
    # Many clients provide a `.search()` method that returns a list/dict of results.
    resp = tavily_client.search(query, max_results=TAVILY_MAX_RESULTS)

    # Normalize results into a list of dicts with 'title', 'url', 'content'.
    raw_results = getattr(resp, "results", resp) or []
    normalized = []
    for r in raw_results:
        if isinstance(r, dict):
            title = r.get("title", "") or ""
            url = r.get("url", "") or ""
            content = r.get("content", "") or r.get("snippet", "") or ""
        else:
            # Fallback for object-like results
            title = getattr(r, "title", "") or ""
            url = getattr(r, "url", "") or ""
            content = getattr(r, "content", "") or getattr(r, "snippet", "") or ""
        normalized.append({"title": title, "url": url, "content": content})

    return {"search_results": normalized}

Notes:

SDKs differ: if tavily_client.search returns a paged object or resp.results, adapt the extraction accordingly.
Keep the normalized output small and consistent to simplify downstream formatting.

answer_direct

Ask the LLM to answer concisely without performing a web lookup. The draft is stored in draft_answer.

def answer_direct(state: AgentState) -> dict:
    """Answer directly without web search."""
    resp = openai_client.responses.create(
        model=OPENAI_MODEL,
        instructions="Answer concisely and accurately.",
        input=state["question"],
    )
    answer_text = (getattr(resp, "output_text", None) or "").strip()
    return {"draft_answer": answer_text}

format_output

The final node handles both branches:

If intent == "search", it formats Tavily results into a readable summary.
If intent == "answer", it returns the LLM’s draft answer.

def format_output(state: AgentState) -> dict:
    """Final formatting node.
    - If intent == 'answer': return the LLM draft answer.
    - If intent == 'search': format Tavily results into a readable response.
    """
    intent = state.get("intent", "answer")

    if intent == "search":
        results = state.get("search_results", [])
        if not results:
            return {"final_answer": "I couldn't find relevant sources for this query."}

        lines = ["Here are the top results I found:"]
        for i, r in enumerate(results, start=1):
            title = (r.get("title") or "").strip()
            url = (r.get("url") or "").strip()
            snippet = (r.get("content") or "").strip()

            # Keep it demo-friendly: short list + short snippet
            if title and url:
                lines.append(f"- [{i}] {title} — {url}")
            elif url:
                lines.append(f"- [{i}] {url}")
            else:
                lines.append(f"- [{i}] (no title/url)")

            if snippet:
                short = snippet[:180] + ("... " if len(snippet) > 180 else "")
                lines.append(f"{short}")

        return {"final_answer": "\n".join(lines).strip()}

    # Default: direct answer path
    answer = (state.get("draft_answer") or "").strip()
    if not answer:
        return {"final_answer": "I couldn't generate an answer."}
    return {"final_answer": answer}

Wiring the graph

We wire the graph to start at classify_intent, branch conditionally to search_web or answer_direct, and then converge at format_output before transitioning to END.

builder = StateGraph(AgentState)

# Register nodes
builder.add_node("classify_intent", classify_intent)
builder.add_node("search_web", search_web)
builder.add_node("answer_direct", answer_direct)
builder.add_node("format_output", format_output)

# Start -> classify
builder.add_edge(START, "classify_intent")

# Conditional routing: returns either "search" or "answer"
def route_by_intent(state: AgentState):
    return state.get("intent", "answer")

builder.add_conditional_edges(
    "classify_intent",
    route_by_intent,
    path_map={
        "search": "search_web",
        "answer": "answer_direct",
    },
)

# Both branches converge into the same formatter, then end
builder.add_edge("search_web", "format_output")
builder.add_edge("answer_direct", "format_output")
builder.add_edge("format_output", END)

app = builder.compile()

Conditional edges let the graph decide the next node dynamically based on the current state. This makes branching explicit, easier to reason about, and straightforward to visualize.

Run examples

Run two sample invocations to exercise both routes:

# Example 1: general knowledge -> direct answer path
out1 = app.invoke({"question": "Explain conditional edges in LangGraph in one sentence."})
print("Intent:", out1.get("intent"))
print(out1.get("final_answer"))

# Example 2: web-query -> search path
out2 = app.invoke({"question": "What is the latest Tavily Search API update? Provide a short summary."})
print("Intent:", out2.get("intent"))
print(out2.get("final_answer"))

Expected behavior:

The first question should choose answer, produce draft_answer via the LLM, and return it as final_answer.
The second should choose search, fetch results with Tavily, and return a formatted list of top matches.

Visualize the graph

To inspect control flow and conditional edges, export the graph as Mermaid source and render it in any Mermaid-compatible tool (for example, mermaid.live or the VS Code Mermaid preview). Many graph implementations expose a method such as to_mermaid() or get_mermaid()—check your graph object’s API. Recommended rendering steps:

Get the Mermaid source string from your graph object.
Paste the Mermaid code into an external renderer (e.g., https://mermaid.live/).
Inspect branching points and convergence to verify the routing.

Extending this pattern

This pattern scales well:

Add more classifier labels and map them to additional tool nodes via add_conditional_edges.
Insert validation or hallucination-checking nodes before the formatter.
Persist important facts into memory nodes that future queries can read.

References and further reading:

Course Introduction

Foundational Graph Concepts and Toolkit Setup

The Core Workflow: Nodes, Edges, and Routing

State Management and Iterative Loops

Context Management and Short-Term Memory

Long-Term Memory and Stateful Persistence

Human-in-the-Loop (HILT) Architectures

Advanced Control and Debugging UX

Conclusion

Building Cyclical Graphs With Conditional Edges

Setup and imports

Typed shared state

Node overview

classify_intent

search_web

answer_direct

format_output

Wiring the graph

Run examples

Visualize the graph

Extending this pattern

Watch Video

​Setup and imports

​Typed shared state

​Node overview

​classify_intent

​search_web

​answer_direct

​format_output

​Wiring the graph

​Run examples

​Visualize the graph

​Extending this pattern

Watch Video

Setup and imports

Typed shared state

Node overview

classify_intent

search_web

answer_direct

format_output

Wiring the graph

Run examples

Visualize the graph

Extending this pattern