pythonadvanced

LangGraph Stateful AI Workflow

Build a multi-node AI workflow with conditional routing using LangGraph's StateGraph.

python
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

class State(TypedDict):
    question: str
    category: str
    answer:   str

llm    = ChatOpenAI(model='gpt-4o-mini', temperature=0)
parser = StrOutputParser()

def classify(state: State) -> State:
    prompt = ChatPromptTemplate.from_template('Classify as "math" or "general": {question}. One word.')
    state['category'] = (prompt | llm | parser).invoke({'question': state['question']}).strip().lower()
    return state

def answer_math(state: State) -> State:
    prompt = ChatPromptTemplate.from_template('Solve step by step: {question}')
    state['answer'] = (prompt | llm | parser).invoke({'question': state['question']})
    return state

def answer_general(state: State) -> State:
    prompt = ChatPromptTemplate.from_template('Answer concisely: {question}')
    state['answer'] = (prompt | llm | parser).invoke({'question': state['question']})
    return state

def route(state: State) -> Literal['math','general']:
    return 'math' if 'math' in state['category'] else 'general'

graph = StateGraph(State)
graph.add_node('classify', classify)
graph.add_node('math', answer_math)
graph.add_node('general', answer_general)
graph.set_entry_point('classify')
graph.add_conditional_edges('classify', route)
graph.add_edge('math', END)
graph.add_edge('general', END)

app = graph.compile()
result = app.invoke({'question': 'What is 17 * 23?', 'category': '', 'answer': ''})
print(result['answer'])

Use Cases

  • AI workflows
  • conditional routing
  • multi-node agents

Tags

Related Snippets

Similar patterns you can reuse in the same workflow.