Agentic Workflow Patterns Complete Guide 2026: Python Code and Framework Examples

from langgraph.graph import StateGraph, END
from typing import TypedDict

class ChainState(TypedDict):
    query: str
    facts: str
    analysis: str
    summary: str

def extract_facts(state: ChainState) -> ChainState:
    """Step 1: Extract key facts from the query."""
    prompt = f"Extract 3-5 key facts from: {state['query']}"
    state["facts"] = llm.invoke(prompt)  # llm is your configured model
    return state

def analyze(state: ChainState) -> ChainState:
    """Step 2: Analyze the extracted facts."""
    prompt = f"Analyze these facts and identify implications:\n{state['facts']}"
    state["analysis"] = llm.invoke(prompt)
    return state

def format_summary(state: ChainState) -> ChainState:
    """Step 3: Format the final summary."""
    prompt = f"Write a concise summary based on this analysis:\n{state['analysis']}"
    state["summary"] = llm.invoke(prompt)
    return state

# Build the chain graph
graph = StateGraph(ChainState)
graph.add_node("extract", extract_facts)
graph.add_node("analyze", analyze)
graph.add_node("format", format_summary)
graph.set_entry_point("extract")
graph.add_edge("extract", "analyze")
graph.add_edge("analyze", "format")
graph.add_edge("format", END)

chain = graph.compile()
result = chain.invoke({"query": "What are the implications of NPU performance on mobile AI?"})
print(result["summary"])

from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal

class RouterState(TypedDict):
    query: str
    category: str
    response: str

def classify(state: RouterState) -> RouterState:
    """Route the query to the appropriate category."""
    prompt = f"Classify this query as 'technical', 'billing', or 'general':\n{state['query']}"
    state["category"] = llm.invoke(prompt).strip().lower()
    return state

def handle_technical(state: RouterState) -> RouterState:
    state["response"] = llm.invoke(f"Provide technical support for: {state['query']}")
    return state

def handle_billing(state: RouterState) -> RouterState:
    state["response"] = llm.invoke(f"Handle billing inquiry: {state['query']}")
    return state

def handle_general(state: RouterState) -> RouterState:
    state["response"] = llm.invoke(f"Answer general question: {state['query']}")
    return state

def route_query(state: RouterState) -> Literal["tech", "billing", "general"]:
    """Conditional routing function."""
    mapping = {"technical": "tech", "billing": "billing", "general": "general"}
    return mapping.get(state["category"], "general")

graph = StateGraph(RouterState)
graph.add_node("classify", classify)
graph.add_node("tech", handle_technical)
graph.add_node("billing", handle_billing)
graph.add_node("general", handle_general)
graph.set_entry_point("classify")
graph.add_conditional_edges("classify", route_query, {
    "tech": "tech", "billing": "billing", "general": "general"
})
graph.add_edge("tech", END)
graph.add_edge("billing", END)
graph.add_edge("general", END)

router = graph.compile()

from langgraph.graph import StateGraph, END
from typing import TypedDict, List
import asyncio

class ParallelState(TypedDict):
    query: str
    results: List[str]
    final: str

async def search_web(state: ParallelState) -> ParallelState:
    result = llm.invoke(f"Search the web for: {state['query']}")
    state["results"].append(f"Web: {result}")
    return state

async def search_docs(state: ParallelState) -> ParallelState:
    result = llm.invoke(f"Search documentation for: {state['query']}")
    state["results"].append(f"Docs: {result}")
    return state

async def search_code(state: ParallelState) -> ParallelState:
    result = llm.invoke(f"Find code examples for: {state['query']}")
    state["results"].append(f"Code: {result}")
    return state

def synthesize(state: ParallelState) -> ParallelState:
    state["final"] = llm.invoke(
        f"Synthesize these sources into a coherent answer:\n" +
        "\n".join(state["results"])
    )
    return state

# LangGraph supports async nodes for fan-out parallelism
graph = StateGraph(ParallelState)
graph.add_node("web", search_web)
graph.add_node("docs", search_docs)
graph.add_node("code", search_code)
graph.add_node("synthesize", synthesize)
graph.set_entry_point("web")
# All three search nodes run in parallel
graph.add_edge("web", "synthesize")
graph.add_edge("docs", "synthesize")
graph.add_edge("code", "synthesize")
graph.add_edge("synthesize", END)

from langgraph.graph import StateGraph, END
from typing import TypedDict

class EvalState(TypedDict):
    prompt: str
    draft: str
    score: int
    iterations: int

MAX_ITERATIONS = 3
QUALITY_THRESHOLD = 8

def generate(state: EvalState) -> EvalState:
    state["draft"] = llm.invoke(f"Write content based on: {state['prompt']}")
    state["iterations"] += 1
    return state

def evaluate(state: EvalState) -> EvalState:
    prompt = f"Score this output 1-10 on quality, clarity, and accuracy:\n{state['draft']}"
    result = llm.invoke(prompt)
    # Extract numerical score from LLM response
    state["score"] = extract_score(result)
    return state

def optimize(state: EvalState) -> EvalState:
    state["draft"] = llm.invoke(
        f"Improve this draft based on quality score {state['score']}/10:\n{state['draft']}"
    )
    return state

def should_continue(state: EvalState) -> str:
    if state["score"] >= QUALITY_THRESHOLD:
        return "accept"
    if state["iterations"] >= MAX_ITERATIONS:
        return "accept"  # Max iterations reached
    return "optimize"

graph = StateGraph(EvalState)
graph.add_node("generate", generate)
graph.add_node("evaluate", evaluate)
graph.add_node("optimize", optimize)
graph.set_entry_point("generate")
graph.add_edge("generate", "evaluate")
graph.add_conditional_edges("evaluate", should_continue, {
    "accept": END,
    "optimize": "optimize"
})
graph.add_edge("optimize", "generate")

evaluator = graph.compile()

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool

# Define worker agents with specialized roles
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find comprehensive information on the assigned topic",
    backstory="Expert at synthesizing information from multiple sources",
    tools=[SerperDevTool(), ScrapeWebsiteTool()],
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear, well-structured content from research",
    backstory="Experienced at explaining complex topics clearly",
    verbose=True
)

reviewer = Agent(
    role="Quality Reviewer",
    goal="Ensure accuracy, clarity, and completeness of output",
    backstory="Detail-oriented editor with deep domain knowledge",
    verbose=True
)

# Define tasks for each agent
research_task = Task(
    description="Research the topic: {topic}. Gather latest information, statistics, and expert opinions.",
    expected_output="A comprehensive research brief with key findings",
    agent=researcher
)

writing_task = Task(
    description="Write a well-structured article based on the research brief. Use clear headings and examples.",
    expected_output="A complete article draft in markdown format",
    agent=writer
)

review_task = Task(
    description="Review the article for accuracy, clarity, and completeness. Suggest specific improvements.",
    expected_output="A review with actionable feedback and the final approved article",
    agent=reviewer
)

# Assemble the crew with sequential orchestration
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff(inputs={"topic": "Edge AI developments in 2026"})
print(result)