We're evaluating LangGraph for automating a support ticket routing workflow. The core logic depends on a database lookup to determine the next node. The docs are heavy on toy examples with hardcoded strings, but light on real-world conditional logic based on external data.
Our prototype needs to:
1. Fetch a ticket's `category_id` from a PostgreSQL table.
2. Branch to a specialized handler node (e.g., `billing_node`, `technical_node`) based on that value.
3. If the lookup fails, route to a human_escalation node.
The naive approach is to do the lookup inside a node's function, then return `"next_node": "some_node"` based on the result. This seems to couple the database I/O with the routing logic tightly. It works, but it feels like the graph's conditional edges (`conditional_edge`) should be able to handle this more cleanly.
Here's the current working pattern we're using, but I'm looking for a more idiomatic review:
```python
from langgraph.graph import StateGraph, END
from typing import TypedDict
import asyncpg
class State(TypedDict):
ticket_id: str
category_id: str | None
next_node: str
async def fetch_ticket_category(state: State):
conn = await asyncpg.connect(DATABASE_URL)
try:
record = await conn.fetchrow(
"SELECT category_id FROM tickets WHERE id = $1", state["ticket_id"]
)
if record:
state["category_id"] = record["category_id"]
else:
state["category_id"] = None
finally:
await conn.close()
# Routing logic inside the node function
if state["category_id"] == "BILL":
state["next_node"] = "billing_node"
elif state["category_id"] == "TECH":
state["next_node"] = "technical_node"
else:
state["next_node"] = "human_escalation_node"
return state
def route_by_next_node(state: State):
# This router function is used by a conditional_edge
return state.get("next_node")
builder = StateGraph(State)
builder.add_node("fetch_ticket_category", fetch_ticket_category)
# ... add other nodes
# Add conditional routing *after* the lookup node
builder.add_conditional_edges(
"fetch_ticket_category",
route_by_next_node,
{
"billing_node": "billing_node",
"technical_node": "technical_node",
"human_escalation_node": "human_escalation_node",
END: END
}
)
```
The question: Is this the intended pattern? Should the database lookup and the routing decision be split into separate nodes for cleaner separation of concerns? How are you handling external data dependencies for branching in production?
I'm concerned about visibility in our observability stack—having the routing logic inside the node function makes it harder to trace the decision point versus having it as an explicit graph edge.
-shift
shift left or go home