Not So Agentic Workflow Automation Patterns
In a recent blog post, I summarized some of the implementation guardrails for workflow automation with agentic systems. Soon after, I came across this epic blog post by Anthropic that covered a set of large language model (LLM) based automation patterns that can be implemented without agents. This blog post is helping many AI developers consider the extent of automation and the variation of the system inputs before choosing an agentic implementation. In other words, many times a seemingly complex workflow may automatically trigger the thought of using agents, but upon closer inspection it can be easily implemented as one of the workflow automation patterns that the Anthropic blog suggested. In this blog, let me go over the patterns with some more detail on implementation like appropriate use cases and pseudo code on implementing the patterns.
These patterns aren't just theoretical constructs. They're practical solutions that can significantly simplify LLM implementations while maintaining robust functionality. In fact, the first two patterns are very obvious and I have used them extensively for building VS Code Snippet extensions and generating use case based IaC documentation. As part of building hands-on-lab for Dell factory use cases, I am using the parallel execution pattern for model benchmarking. Whether you're building developer tools, content generators, or analysis systems, understanding these patterns helps you make better architectural decisions.
In this post, I'll dive deep into each pattern, sharing:
- Concrete implementation examples with “pseudo” code
- Real-world use cases and applications
- Key considerations for each pattern
- Implementation tips from hands-on experience
Note on “pseudo” code: My goal is to help you identify which pattern best fits your use case (and even quickly building on the pseudo code) before jumping into more complex agent-based solutions. In the pseudo-code snippets, I assumed an LLM endpoint behind something like LLMService() method. Once you attach your “LLMService” implementation (LLMs served using Ollama, LMStudio etc. or your enterprise end points with API keys) and install the required Python libraries the examples should run well with minimal changes. There are multiple frameworks like langchain, langgraph, llamaindex, pydantiAI and others to build LLM-based workflows and systems. Please note that the pseudo code doesn’t use any particular framework and provides the workflow pattern in plain Python. You can refer the documentation of the frameworks to build the workflow using the constructs provided by them. Now let's explore the workflow automation patterns.
1. LLM Augmentation: Enhancing Language Models with External Capabilities
The Augmented LLM pattern represents a fundamental approach to extending the capabilities of LLMs by integrating them with external components. In this pattern, the LLM serves as the central orchestrator while interfacing with three key augmentations: retrieval systems, tools, and memory components. This architecture allows the LLM to overcome its inherent limitations by accessing current information, performing specific actions, and maintaining contextual awareness.
Pattern Overview
At its core, this pattern follows a simple yet powerful flow:
- The input is first processed by the LLM to determine required augmentations
- The LLM interacts with external components as needed:
- Retrieval systems for accessing up-to-date information
- Tools for performing specific actions
- Memory for maintaining conversation context
- The LLM then synthesizes information from these sources to generate the final output
Common Applications
This pattern proves particularly valuable in several scenarios:
- Customer service systems that need to access product documentation, process orders, and maintain conversation context
- Research assistants that combine knowledge retrieval with analytical tools
- Personal productivity assistants that interact with calendar tools while remembering user preferences
- Educational tutors that access course materials and track student progress
Use Case: Customer Service Implementation
Let's explore a practical implementation of this pattern in a customer service context. This example demonstrates how the pattern handles documentation lookup, order processing, and conversation memory:
class AugmentedCustomerServiceLLM: def __init__(self): self.llm = LLMService() # Base LLM service self.retrieval = DocumentStore() # Vector DB for company docs self.tools = OrderManagementSystem() # Order processing tools self.memory = ConversationMemory() # Conversation state async def process_query(self, user_input: str) -> str: # Step 1: Update conversation memory self.memory.add_user_message(user_input) # Step 2: Create enriched prompt with context prompt = self._build_prompt( user_input=user_input, conversation_history=self.memory.get_recent_messages(), ) # Step 3: Get LLM's initial analysis llm_analysis = await self.llm.analyze(prompt) # Step 4: Handle different types of requests response = None if llm_analysis.needs_documentation: # Query documentation docs = self.retrieval.search(llm_analysis.doc_query) response = await self.llm.generate_response(prompt, docs) elif llm_analysis.needs_order_action: # Use tools to process order-related requests order_data = await self.tools.process_order_action( action=llm_analysis.action_type, order_id=llm_analysis.order_id ) response = await self.llm.generate_response(prompt, order_data) else: # Direct LLM response for general queries response = llm_analysis.response # Step 5: Update memory with response self.memory.add_assistant_message(response) return response
Example Usage
# Customer interaction example queries = [ "What's your return policy for electronics?", # Triggers documentation retrieval "I want to return order #12345", # Activates order processing tools "Can you confirm what we discussed?" # Utilizes conversation memory ] async def demonstration(): assistant = AugmentedCustomerServiceLLM() for query in queries: response = await assistant.process_query(query) print(f"User: {query}\nAssistant: {response}\n")
Advantages
- Combines the LLM's natural language understanding with practical capabilities
- Maintains single responsibility for each component while enabling complex interactions
- Scales easily by adding new tools or expanding retrieval sources
Implementation Considerations
- Requires careful prompt engineering for effective orchestration
- Need for robust error handling across all components
- Memory management must be optimized for long conversations
2. Prompt Chaining: Sequential LLM Processing with Quality Gates
The Prompt Chaining pattern implements a sequential workflow where multiple LLM calls are arranged in a pipeline, with each step building upon the previous one's output. A key feature of this pattern is the inclusion of quality gates between steps to ensure the output meets specific criteria before proceeding.
Pattern Overview
The pattern consists of three main components:
- Sequential LLM calls, each specialized for a specific task
- Quality gates that validate outputs before proceeding
- Conditional branching based on quality checks
Common Applications
This pattern is particularly effective for:
- Content creation pipelines (outline → draft → polish)
- Multi-step analysis (extract → analyze → summarize)
- Translation workflows (translate → verify → localize)
- Code generation (design → implement → test)
Use Case: Technical Content Creation Pipeline
Let's explore an implementation that generates technical blog posts through a three-stage process:
class TechnicalContentPipeline:
def __init__(self):
self.llm = LLMService()
async def generate_content(self, topic: str) -> str:
# Stage 1: Generate Outline
outline = await self._generate_outline(topic)
# Gate 1: Validate Outline
if not self._validate_outline(outline):
return self._handle_failure("Outline validation failed")
# Stage 2: Create Draft
draft = await self._create_draft(outline)
# Gate 2: Validate Draft
if not self._validate_draft(draft):
return self._handle_failure("Draft validation failed")
# Stage 3: Polish Content
final_content = await self._polish_content(draft)
return final_content
async def _generate_outline(self, topic: str) -> dict:
prompt = f"""
Create a detailed outline for a technical blog post about {topic}.
Include:
- Main sections
- Key points for each section
- Technical concepts to be covered
"""
return await self.llm.generate(prompt)
def _validate_outline(self, outline: dict) -> bool:
# Quality gate checks for outline
required_sections = ['introduction', 'main_content', 'conclusion']
min_sections = 3
min_points_per_section = 2
validation_prompt = f"""
Validate this outline against the following criteria:
1. Contains required sections: {required_sections}
2. Has at least {min_sections} sections
3. Each section has {min_points_per_section} or more points
Outline:
{outline}
Return only True or False.
"""
return self.llm.validate(validation_prompt)
async def _create_draft(self, outline: dict) -> str:
prompt = f"""
Create a detailed draft based on this outline:
{outline}
Requirements:
- Technical accuracy
- Code examples where relevant
- Clear explanations
- Logical flow between sections
"""
return await self.llm.generate(prompt)
def _validate_draft(self, draft: str) -> bool:
validation_criteria = {
'technical_accuracy': 0.8,
'completeness': 0.9,
'coherence': 0.85
}
validation_prompt = f"""
Evaluate this draft against these criteria:
{validation_criteria}
Draft:
{draft}
Return a score for each criterion (0-1).
"""
scores = self.llm.evaluate(validation_prompt)
return all(scores[k] >= v for k, v in validation_criteria.items())
async def _polish_content(self, draft: str) -> str:
prompt = f"""
Polish this draft into final content:
{draft}
Focus on:
- Professional tone
- Technical precision
- Flow and readability
- Engaging style
"""
return await self.llm.generate(prompt)
def _handle_failure(self, error: str) -> str:
return f"Content generation failed: {error}"
Example Usage
# Initialize the pipeline content_pipeline = TechnicalContentPipeline() # Generate content async def create_blog_post(): topic = "Implementing GraphQL in Python" final_content = await content_pipeline.generate_content(topic) return final_content
Advantages
- Breaks complex tasks into manageable steps
- Ensures quality at each stage
- Allows for specialized optimization of each step
- Provides clear failure points for debugging
Implementation Considerations
- Need to balance gate strictness with pass rates
- Prompt design crucial for each stage
- Error handling strategy important for failed stages
- Cost implications of multiple LLM calls
3. LLM Routing: Dynamic Task Distribution in AI Systems
The LLM Routing pattern represents an intelligent approach to directing inputs to specialized LLM instances based on the nature of the task. This pattern uses a central router LLM to analyze incoming requests and direct them to the most appropriate specialized LLM for processing.
Pattern Overview
The pattern consists of three main components:
- A router LLM that analyzes and categorizes incoming requests
- Multiple specialized LLM instances optimized for specific tasks
- A routing mechanism that directs requests to appropriate LLMs
Common Applications
This pattern is particularly effective for:
- Multi-lingual support systems
- Technical support with varied domains (hardware, software, network)
- Content generation across different formats (articles, code, poetry)
- Educational systems handling different subjects
Use Case: Technical Support System
Let's explore an implementation of a technical support system that routes queries to specialized LLMs:
class TechnicalSupportRouter:
def __init__(self):
self.router_llm = LLMService(model="router-model")
self.specialized_llms = {
"hardware": LLMService(model="hardware-specialist"),
"software": LLMService(model="software-specialist"),
"network": LLMService(model="network-specialist")
}
async def process_query(self, user_query: str) -> str:
# Step 1: Analyze query with router LLM
routing_analysis = await self._analyze_query(user_query)
# Step 2: Route to appropriate specialist
response = await self._route_to_specialist(
category=routing_analysis['category'],
query=user_query,
confidence=routing_analysis['confidence']
)
return response
async def _analyze_query(self, query: str) -> dict:
routing_prompt = f"""
Analyze this technical support query and determine:
1. The primary category (hardware/software/network)
2. Confidence level in the categorization (0-1)
3. Any additional context needed
Query: {query}
Return JSON with category, confidence, and context.
"""
return await self.router_llm.analyze(routing_prompt)
async def _route_to_specialist(self, category: str, query: str, confidence: float) -> str:
# If confidence is low, use multiple specialists
if confidence < 0.7:
return await self._handle_uncertain_query(query, category)
specialist = self.specialized_llms.get(category)
if not specialist:
return "Unable to find appropriate specialist for query"
specialist_prompt = f"""
As a {category} specialist, address this query:
{query}
Provide a detailed, technically accurate response.
"""
return await specialist.generate(specialist_prompt)
async def _handle_uncertain_query(self, query: str, primary_category: str) -> str:
# Gather responses from multiple specialists
responses = {}
for category, llm in self.specialized_llms.items():
if category == primary_category:
continue
responses[category] = await llm.generate(query)
# Synthesize responses
synthesis_prompt = f"""
Synthesize these specialist responses into a comprehensive answer:
{responses}
Original query: {query}
"""
return await self.router_llm.generate(synthesis_prompt)
Example Usage
# Initialize the router
support_system = TechnicalSupportRouter()
# Example queries
async def handle_support_queries():
queries = [
"My laptop won't turn on after a power outage", # Hardware
"Getting 404 errors on our web service", # Network
"Python package dependencies are conflicting" # Software
]
for query in queries:
response = await support_system.process_query(query)
print(f"Query: {query}\nResponse: {response}\n")
Advantages
- Improved response quality through specialization
- Efficient resource utilization
- Graceful handling of uncertain cases
- Scalable through addition of new specialists
Implementation Considerations
- Router LLM quality crucial for system performance
- Balance between specialization and maintenance complexity
- Cost implications of multiple LLM instances
- Need for regular specialist model updates
4. Parallelization: Concurrent LLM Processing with Aggregation
The Parallelization pattern leverages multiple LLM instances running concurrently to process the same input from different perspectives, followed by an aggregation step to combine their outputs. This pattern is particularly useful when diverse viewpoints or multiple specialized analyses are needed for a comprehensive response.
Pattern Overview
The pattern consists of three main components:
- Multiple concurrent LLM calls with different prompts/perspectives
- Parallel execution management
- An aggregator that combines and synthesizes the results
Common Applications
This pattern excels in scenarios requiring:
- Multi-perspective analysis (technical, business, user experience)
- Competitive analysis and comparison
- Complex decision-making processes
- Comprehensive content generation
Use Case: Product Review Analysis System
Let's implement a system that analyzes product reviews from multiple angles simultaneously:
import asyncio
from typing import List, Dict
class ProductReviewAnalyzer:
def __init__(self):
self.llm = LLMService()
async def analyze_reviews(self, reviews: List[str]) -> Dict:
# Run all analyses in parallel
technical_analysis, sentiment_analysis, market_analysis = await asyncio.gather(
self._analyze_technical_aspects(reviews),
self._analyze_sentiment(reviews),
self._analyze_market_positioning(reviews)
)
# Aggregate results
final_analysis = await self._aggregate_analyses(
technical_analysis,
sentiment_analysis,
market_analysis
)
return final_analysis
async def _analyze_technical_aspects(self, reviews: List[str]) -> Dict:
prompt = f"""
Analyze these product reviews focusing on technical aspects:
- Feature functionality
- Performance metrics
- Technical issues mentioned
- Compatibility concerns
Reviews: {reviews}
"""
return await self.llm.analyze(prompt)
async def _analyze_sentiment(self, reviews: List[str]) -> Dict:
prompt = f"""
Analyze the emotional content and user satisfaction in these reviews:
- Overall sentiment scoring
- Key positive points
- Major complaints
- User satisfaction trends
Reviews: {reviews}
"""
return await self.llm.analyze(prompt)
async def _analyze_market_positioning(self, reviews: List[str]) -> Dict:
prompt = f"""
Analyze market positioning based on these reviews:
- Competitive comparisons
- Price-value perception
- Market segment alignment
- Brand perception
Reviews: {reviews}
"""
return await self.llm.analyze(prompt)
async def _aggregate_analyses(self, technical: Dict, sentiment: Dict, market: Dict) -> Dict:
aggregation_prompt = f"""
Synthesize these three analyses into a comprehensive report:
Technical Analysis: {technical}
Sentiment Analysis: {sentiment}
Market Analysis: {market}
Provide:
1. Executive summary
2. Key findings from each perspective
3. Cross-perspective insights
4. Recommended actions
"""
return await self.llm.generate(aggregation_prompt)
Example Usage
# Initialize the analyzer
analyzer = ProductReviewAnalyzer()
# Example reviews
reviews = [
"Great battery life but software is buggy. Still better than competitor X.",
"Premium price but worth it. Build quality is exceptional.",
"Missing basic features that cheaper alternatives have."
]
async def analyze_product_feedback():
analysis = await analyzer.analyze_reviews(reviews)
print("Comprehensive Analysis:", analysis)
Advantages
- Reduced total processing time through parallelization
- More comprehensive analysis through multiple perspectives
- Better handling of complex, multi-faceted problems
- Improved objectivity through diverse viewpoints
Implementation Considerations
- Resource management for concurrent LLM calls
- Cost implications of multiple simultaneous calls
- Aggregation logic complexity
- Handling conflicting outputs
- Load balancing and rate limiting
5. The Orchestrator-Workers Pattern: Coordinated LLM Task Distribution
The Orchestrator-Workers pattern implements a hierarchical approach where a central orchestrator LLM breaks down complex tasks and coordinates multiple worker LLMs, followed by a synthesis of their outputs. This pattern is particularly effective for complex tasks that require multiple specialized sub-tasks to be executed and coordinated.
Pattern Overview
The pattern consists of three main components:
- An orchestrator LLM that analyzes tasks and creates subtask assignments
- Worker LLMs that execute specialized subtasks
- A synthesizer that combines worker outputs into a coherent result
Common Applications
This pattern is particularly effective for:
- Research projects requiring multiple types of analysis
- Complex content creation (e.g., full marketing campaigns)
- Multi-step problem solving
- Comprehensive data analysis projects
Use Case: Research Report Generator
Let's implement a system that generates comprehensive research reports by coordinating multiple research tasks:
class ResearchOrchestrator:
def __init__(self):
self.llm = LLMService()
async def generate_research_report(self, topic: str) -> str:
# Step 1: Orchestrator analyzes and plans
research_plan = await self._create_research_plan(topic)
# Step 2: Distribute tasks to workers
worker_results = await self._execute_research_tasks(research_plan)
# Step 3: Synthesize results
final_report = await self._synthesize_research(worker_results, topic)
return final_report
async def _create_research_plan(self, topic: str) -> dict:
planning_prompt = f"""
Create a research plan for: {topic}
Break down into specialized tasks including:
1. Background research and context
2. Current state analysis
3. Future trends prediction
4. Impact analysis
For each task specify:
- Key questions to answer
- Required data sources
- Analysis approach
"""
return await self.llm.plan(planning_prompt)
async def _execute_research_tasks(self, plan: dict) -> List[dict]:
tasks = []
for task in plan['tasks']:
worker_prompt = self._create_worker_prompt(task)
tasks.append(self.llm.analyze(worker_prompt))
return await asyncio.gather(*tasks)
def _create_worker_prompt(self, task: dict) -> str:
return f"""
Execute this research task:
Task: {task['description']}
Questions: {task['key_questions']}
Approach: {task['analysis_approach']}
Provide:
1. Detailed findings
2. Supporting evidence
3. Key insights
4. Limitations of analysis
"""
async def _synthesize_research(self, results: List[dict], topic: str) -> str:
synthesis_prompt = f"""
Create a comprehensive research report on {topic}.
Individual research components:
{results}
Synthesize into:
1. Executive Summary
2. Comprehensive Analysis
3. Key Findings
4. Future Implications
5. Recommendations
Ensure coherent narrative and clear connections between different research components.
"""
return await self.llm.generate(synthesis_prompt)
Example Usage
# Initialize the orchestrator
research_system = ResearchOrchestrator()
# Generate research report
async def generate_report():
topic = "Impact of Artificial Intelligence on Healthcare Delivery"
report = await research_system.generate_research_report(topic)
print("Research Report:", report)
Advantages
- Complex task management through hierarchical structure
- Specialized focus for each worker
- Coordinated execution of related tasks
- Coherent final output through structured synthesis
Implementation Considerations
- Task decomposition strategy
- Worker prompt design
- Resource allocation
- Synthesis complexity
- Error handling across multiple components
6. The Evaluator-Optimizer Pattern: Self-Improving LLM Output Generation
The Evaluator-Optimizer pattern implements a feedback loop where generated content is evaluated against specific criteria, and rejected content is refined based on detailed feedback. This pattern is particularly useful for ensuring high-quality outputs that meet precise requirements.
Pattern Overview
The pattern consists of two main components:
- A generator LLM that creates initial content
- An evaluator LLM that assesses content quality and provides improvement feedback
Common Applications
This pattern is particularly effective for:
- Code generation with specific quality requirements
- Technical documentation creation
- Academic writing and research papers
- Legal document drafting
- Marketing copy that must meet brand guidelines
Use Case: Technical Documentation Generator
Let's implement a system that generates and refines technical documentation until it meets quality standards:
class TechnicalDocGenerator:
def __init__(self):
self.generator_llm = LLMService(model="generator")
self.evaluator_llm = LLMService(model="evaluator")
self.max_iterations = 3
async def generate_documentation(self, specification: dict) -> str:
iteration = 0
while iteration < self.max_iterations:
# Generate documentation
documentation = await self._generate_docs(
specification,
iteration_feedback=None if iteration == 0 else feedback
)
# Evaluate the documentation
evaluation = await self._evaluate_docs(documentation, specification)
if evaluation['accepted']:
return documentation
feedback = evaluation['feedback']
iteration += 1
raise Exception("Failed to generate acceptable documentation")
async def _generate_docs(self, spec: dict, iteration_feedback: Optional[str]) -> str:
prompt = f"""
Generate technical documentation based on this specification:
{spec}
Requirements:
- Clear structure
- Comprehensive coverage
- Technical accuracy
- Code examples where relevant
Previous feedback to address:
{iteration_feedback if iteration_feedback else 'Initial generation'}
"""
return await self.generator_llm.generate(prompt)
async def _evaluate_docs(self, documentation: str, spec: dict) -> dict:
evaluation_prompt = f"""
Evaluate this technical documentation against the specification:
Specification:
{spec}
Documentation:
{documentation}
Evaluate against these criteria:
1. Completeness (all requirements covered)
2. Technical accuracy
3. Clarity and structure
4. Code example quality
5. Target audience appropriateness
Return:
1. Pass/fail for each criterion
2. Detailed feedback for failed criteria
3. Overall acceptance decision
4. Specific improvement suggestions
"""
evaluation = await self.evaluator_llm.analyze(evaluation_prompt)
return {
'accepted': evaluation['meets_all_criteria'],
'feedback': evaluation['improvement_suggestions']
}
Example Usage
# Initialize the documentation generator
doc_generator = TechnicalDocGenerator()
# Generate documentation
async def create_documentation():
spec = {
"component": "Authentication Service",
"audience": "Senior Developers",
"requirements": [
"API endpoints documentation",
"Security considerations",
"Implementation examples",
"Performance characteristics"
]
}
try:
documentation = await doc_generator.generate_documentation(spec)
print("Final Documentation:", documentation)
except Exception as e:
print("Failed to generate acceptable documentation:", str(e))
Pattern Benefits and Considerations
Advantages
- Ensures high-quality output meeting specific criteria
- Provides detailed feedback for improvements
- Self-improving through iteration
- Maintains consistency with requirements
Implementation Considerations
- Balance between strictness and acceptance criteria
- Maximum iteration limit determination
- Resource usage in feedback loop
- Evaluation criteria design
- Feedback quality and specificity
Conclusion
The combination of the patterns discussed above with the latest open-source models and local deployment creates a powerful foundation for building robust AI applications. The key is matching the right pattern with appropriate models (for example, small, large, reasoning enabled) and infrastructure. Whether you're implementing a simple augmented LLM pattern or a complex orchestrator-workers system, today's open-source ecosystem provides the tools needed for successful deployment.
For those starting their implementation journey, I recommend:
- Start with simpler patterns and evolve as needed
- Experiment with different model sizes and architectures
- Pay attention to your specific use case requirements
- Monitor and optimize based on real usage patterns
- Stay informed about new model releases and capabilities
The future of LLM implementation lies in this intersection of implementation patterns, powerful open-source models, and optimized local deployments. The tools and patterns we've explored provide a solid foundation for building the next generation of AI applications.
Author: Parasar Kodati, Staff Technologist, ISG (LinkedIn)