coderplex-tech · Feb 17, 2025
diff --git a/‎.cursorlogs
+62 b/‎.cursorlogs
+62
diff --git a/‎agents/autonomous-agent/README.md
+220 b/‎agents/autonomous-agent/README.md
+220
diff --git a/‎agents/autonomous-agent/agent.py
+294 b/‎agents/autonomous-agent/agent.py
+294
diff --git a/‎agents/autonomous-agent/examples/research_assistant.py
+175 b/‎agents/autonomous-agent/examples/research_assistant.py
+175
diff --git a/‎agents/domain-specific/README.md
+282 b/‎agents/domain-specific/README.md
+282
diff --git a/‎agents/domain-specific/agent.py
+260 b/‎agents/domain-specific/agent.py
+260
diff --git a/‎agents/domain-specific/examples/medical_assistant.py
+278 b/‎agents/domain-specific/examples/medical_assistant.py
+278
diff --git a/‎building-block/augmented-llm/README.md
+129 b/‎building-block/augmented-llm/README.md
+129
diff --git a/‎examples/building-blocks/augmented-llms/augmented.py renamed to ‎building-block/augmented-llm/augmented.py b/‎examples/building-blocks/augmented-llms/augmented.py renamed to ‎building-block/augmented-llm/augmented.py
diff --git a/‎examples/building-blocks/augmented-llms/examples/__init__.py renamed to ‎building-block/augmented-llm/examples/__init__.py b/‎examples/building-blocks/augmented-llms/examples/__init__.py renamed to ‎building-block/augmented-llm/examples/__init__.py
diff --git a/‎examples/building-blocks/augmented-llms/examples/anthropic.py renamed to ‎building-block/augmented-llm/examples/anthropic.py b/‎examples/building-blocks/augmented-llms/examples/anthropic.py renamed to ‎building-block/augmented-llm/examples/anthropic.py
diff --git a/‎building-block/augmented-llm/examples/basic_usage.py
+39 b/‎building-block/augmented-llm/examples/basic_usage.py
+39
diff --git a/‎examples/building-blocks/augmented-llms/examples/openai.py renamed to ‎building-block/augmented-llm/examples/openai.py b/‎examples/building-blocks/augmented-llms/examples/openai.py renamed to ‎building-block/augmented-llm/examples/openai.py
diff --git a/‎examples/building-blocks/augmented-llms/images/augmented-llm-diagram renamed to ‎building-block/augmented-llm/images/augmented-llm-diagram b/‎examples/building-blocks/augmented-llms/images/augmented-llm-diagram renamed to ‎building-block/augmented-llm/images/augmented-llm-diagram
diff --git a/‎examples/building-blocks/augmented-llms/requirements.txt renamed to ‎building-block/augmented-llm/requirements.txt b/‎examples/building-blocks/augmented-llms/requirements.txt renamed to ‎building-block/augmented-llm/requirements.txt
diff --git a/‎examples/building-blocks/augmented-llms/tools/__init__.py renamed to ‎building-block/augmented-llm/tools/__init__.py b/‎examples/building-blocks/augmented-llms/tools/__init__.py renamed to ‎building-block/augmented-llm/tools/__init__.py
diff --git a/‎building-block/augmented-llm/tools/basic_tools.py
+43 b/‎building-block/augmented-llm/tools/basic_tools.py
+43
diff --git a/‎examples/building-blocks/augmented-llms/tools/calculator.py renamed to ‎building-block/augmented-llm/tools/calculator.py b/‎examples/building-blocks/augmented-llms/tools/calculator.py renamed to ‎building-block/augmented-llm/tools/calculator.py
diff --git a/‎examples/building-blocks/augmented-llms/tools/search_tool.py renamed to ‎building-block/augmented-llm/tools/search_tool.py b/‎examples/building-blocks/augmented-llms/tools/search_tool.py renamed to ‎building-block/augmented-llm/tools/search_tool.py
diff --git a/‎examples/building-blocks/augmented-llms/README.md
-57 b/‎examples/building-blocks/augmented-llms/README.md
-57
diff --git a/‎requirements.txt
+11 b/‎requirements.txt
+11
diff --git a/‎workflows/evaluator-optimizer/README.md
+204 b/‎workflows/evaluator-optimizer/README.md
+204
diff --git a/‎workflows/evaluator-optimizer/examples/code_optimization.py
+98 b/‎workflows/evaluator-optimizer/examples/code_optimization.py
+98
diff --git a/‎workflows/evaluator-optimizer/optimizer.py
+314 b/‎workflows/evaluator-optimizer/optimizer.py
+314
diff --git a/‎workflows/orchestrator-workers/README.md
+192 b/‎workflows/orchestrator-workers/README.md
+192
diff --git a/‎workflows/orchestrator-workers/examples/document_processing.py
+149 b/‎workflows/orchestrator-workers/examples/document_processing.py
+149
diff --git a/‎workflows/orchestrator-workers/orchestrator.py
+223 b/‎workflows/orchestrator-workers/orchestrator.py
+223
diff --git a/‎workflows/parallelization/README.md
+193 b/‎workflows/parallelization/README.md
+193
@@ -0,0 +1,62 @@
+# Cursor Development Logs
+
+[2024-02-15 Initial Setup]
+PROBLEM: Need to track development progress - ACTION: Created .cursorlogs file - RESULT: Successfully created tracking file
+PROBLEM: Repository structure analysis - ACTION: Listed current files and directories - RESULT: Found basic structure with building-block/, workflows/, and agents/ directories
+PROBLEM: Project dependencies - ACTION: Created requirements.txt - RESULT: Added core dependencies for LLM implementations
+PROBLEM: Building-block analysis - ACTION: Inspected building-block/augmented-llm - RESULT: Found basic implementation with augmented.py and supporting files
+PROBLEM: Workflows analysis - ACTION: Inspected workflows directory - RESULT: Found empty directories for all workflow patterns
+PROBLEM: Agents analysis - ACTION: Inspected agents directory - RESULT: Found empty directory, needs complete implementation
+
+[2024-02-15 Implementation Plan]
+
+1. Complete building-block/augmented-llm implementation
+2. Implement workflow patterns:
+   - prompt-chaining
+   - routing
+   - parallelization
+   - orchestrator-workers
+   - evaluator-optimizer
+3. Implement agent patterns:
+   - autonomous-agent
+   - domain-specific-agents
+
+[2024-02-15 Building Block Implementation]
+PROBLEM: Augmented LLM tools needed - ACTION: Created basic_tools.py with WebSearch, Calculator, and Weather tools - RESULT: Successfully implemented tool classes
+PROBLEM: Example implementation needed - ACTION: Created basic_usage.py with complete example - RESULT: Successfully demonstrated tool and memory usage
+PROBLEM: Documentation needed - ACTION: Updated augmented-llm README.md - RESULT: Comprehensive documentation of implementation and usage
+
+[2024-02-15 Workflow Implementation - Prompt Chaining]
+PROBLEM: Core prompt chaining implementation needed - ACTION: Created chain.py with ChainStep, PromptChain, and ChainBuilder classes - RESULT: Successfully implemented base functionality
+PROBLEM: Example implementation needed - ACTION: Created text_analysis.py demonstrating multi-step text analysis - RESULT: Successfully implemented three-step chain example
+PROBLEM: Documentation needed - ACTION: Created prompt-chaining README.md - RESULT: Comprehensive documentation of pattern and usage
+
+[2024-02-15 Workflow Implementation - Routing]
+PROBLEM: Core routing implementation needed - ACTION: Created router.py with Route, Router, and RouterBuilder classes - RESULT: Successfully implemented base functionality
+PROBLEM: Example implementation needed - ACTION: Created support_routing.py demonstrating support ticket routing - RESULT: Successfully implemented support routing system
+PROBLEM: Documentation needed - ACTION: Created routing README.md - RESULT: Comprehensive documentation of pattern and usage
+
+[2024-02-15 Workflow Implementation - Parallelization]
+PROBLEM: Core parallelization implementation needed - ACTION: Created parallel.py with ParallelTask, Parallelizer, and ParallelizerBuilder classes - RESULT: Successfully implemented base functionality
+PROBLEM: Example implementation needed - ACTION: Created content_moderation.py demonstrating parallel content moderation - RESULT: Successfully implemented moderation system with multiple checks
+PROBLEM: Documentation needed - ACTION: Created parallelization README.md - RESULT: Comprehensive documentation of pattern and usage
+
+[2024-02-15 Workflow Implementation - Orchestrator-Workers]
+PROBLEM: Core orchestrator implementation needed - ACTION: Created orchestrator.py with Task, Worker, Orchestrator, and OrchestratorBuilder classes - RESULT: Successfully implemented base functionality
+PROBLEM: Example implementation needed - ACTION: Created document_processing.py demonstrating document analysis system - RESULT: Successfully implemented document processing with multiple workers
+PROBLEM: Documentation needed - ACTION: Created orchestrator-workers README.md - RESULT: Comprehensive documentation of pattern and usage
+
+[2024-02-15 Workflow Implementation - Evaluator-Optimizer]
+PROBLEM: Core optimizer implementation needed - ACTION: Created optimizer.py with Candidate, Optimizer, and OptimizerBuilder classes - RESULT: Successfully implemented base functionality with multiple strategies
+PROBLEM: Example implementation needed - ACTION: Created code_optimization.py demonstrating code improvement system - RESULT: Successfully implemented code optimization with multiple strategies
+PROBLEM: Documentation needed - ACTION: Created evaluator-optimizer README.md - RESULT: Comprehensive documentation of pattern and usage
+
+[2024-02-15 Agent Implementation - Autonomous Agent]
+PROBLEM: Core agent implementation needed - ACTION: Created agent.py with Action, Plan, Tool, Memory, and AutonomousAgent classes - RESULT: Successfully implemented base functionality with planning and execution capabilities
+PROBLEM: Example implementation needed - ACTION: Created research_assistant.py demonstrating autonomous research system - RESULT: Successfully implemented research assistant with multiple tools and adaptive planning
+PROBLEM: Documentation needed - ACTION: Created autonomous-agent README.md - RESULT: Comprehensive documentation of pattern and usage
+
+[2024-02-15 Agent Implementation - Domain-Specific Agent]
+PROBLEM: Core domain-specific agent implementation needed - ACTION: Created domain-specific/agent.py with DomainConstraint, DomainKnowledge, DomainBehavior, and DomainSpecificAgent classes - RESULT: Successfully implemented base functionality with domain-specific capabilities
+PROBLEM: Example implementation needed - ACTION: Created medical_assistant.py demonstrating medical diagnosis system - RESULT: Successfully implemented medical assistant with domain knowledge, constraints, and behaviors
+PROBLEM: Documentation needed - ACTION: Created domain-specific README.md - RESULT: Comprehensive documentation of pattern and usage
@@ -0,0 +1,220 @@
+# Autonomous Agent Pattern
+
+This pattern demonstrates how to implement self-directed agents that can plan, execute, and adapt their behavior to achieve goals.
+
+## Overview
+
+The autonomous agent pattern is useful when:
+
+- Tasks require complex planning and execution
+- Goals can be broken down into discrete actions
+- Adaptation to results is needed
+- Long-running tasks need monitoring
+
+## Components
+
+### Core Classes
+
+1. **Action**
+
+   - Represents a discrete task
+   - Tracks execution status
+   - Stores results
+   - Manages parameters
+
+2. **Plan**
+
+   - Sequences of actions
+   - Goal definition
+   - Context information
+   - Execution status
+
+3. **Tool**
+
+   - Specific capabilities
+   - Parameter definitions
+   - Execution handlers
+   - Error handling
+
+4. **Memory**
+
+   - Action history
+   - Plan history
+   - Context storage
+   - State management
+
+5. **AutonomousAgent**
+   - Plan creation
+   - Action execution
+   - Progress evaluation
+   - Plan adjustment
+
+## Example Implementation
+
+The `examples/research_assistant.py` demonstrates autonomous research:
+
+1. Research Tools:
+
+   - Web Search
+   - Webpage Reading
+   - Text Summarization
+   - Fact Extraction
+   - Note Taking
+
+2. Features:
+   - Goal decomposition
+   - Adaptive planning
+   - Progress tracking
+   - Result aggregation
+
+## Usage
+
+1. Set up environment:
+
+```bash
+pip install -r requirements.txt
+```
+
+2. Configure API keys:
+
+```bash
+export ANTHROPIC_API_KEY=your_key_here
+# or
+export OPENAI_API_KEY=your_key_here
+```
+
+3. Run the example:
+
+```bash
+python -m examples.research_assistant
+```
+
+## Implementation Details
+
+### Tool Definition
+
+```python
+Tool(
+    name="tool_name",
+    description="tool description",
+    parameters={"param": "type"},
+    handler=async_handler_function
+)
+```
+
+### Agent Building
+
+```python
+agent = (
+    AgentBuilder()
+    .add_tool(...)
+    .add_tool(...)
+    .with_memory(memory)
+    .build(llm_caller)
+)
+```
+
+### Goal Execution
+
+```python
+actions = await agent.execute(
+    goal="goal description",
+    context={}
+)
+```
+
+## Agent Capabilities
+
+1. **Planning**
+
+   - Goal analysis
+   - Task decomposition
+   - Action sequencing
+   - Dependency management
+
+2. **Execution**
+
+   - Tool selection
+   - Parameter preparation
+   - Result handling
+   - Error recovery
+
+3. **Monitoring**
+
+   - Progress tracking
+   - Goal evaluation
+   - Plan adjustment
+   - Performance analysis
+
+4. **Learning**
+   - Action history
+   - Success patterns
+   - Failure analysis
+   - Strategy adaptation
+
+## Best Practices
+
+1. **Goal Design**
+
+   - Clear objectives
+   - Measurable outcomes
+   - Reasonable scope
+   - Success criteria
+
+2. **Tool Implementation**
+
+   - Focused functionality
+   - Clear interfaces
+   - Robust error handling
+   - Performance optimization
+
+3. **Memory Management**
+
+   - Relevant history
+   - Context preservation
+   - State cleanup
+   - Storage efficiency
+
+4. **Error Handling**
+   - Graceful degradation
+   - Recovery strategies
+   - Feedback loops
+   - Logging and monitoring
+
+## Extensions
+
+Consider extending this pattern with:
+
+1. **Advanced Planning**
+
+   - Multi-goal handling
+   - Priority management
+   - Resource allocation
+   - Constraint satisfaction
+
+2. **Learning Capabilities**
+
+   - Strategy optimization
+   - Pattern recognition
+   - Performance tuning
+   - Knowledge base
+
+3. **Collaboration**
+
+   - Agent communication
+   - Task delegation
+   - Resource sharing
+   - Conflict resolution
+
+4. **Safety**
+
+   - Action validation
+   - Resource limits
+   - Security checks
+   - Ethical constraints
+
+5. **Integration**
+   - External services
+   - Data sources
+   - Monitoring systems
+   - Reporting tools
@@ -0,0 +1,294 @@
+from typing import List, Dict, Any, Callable, Optional, Union, Tuple
+from abc import ABC, abstractmethod
+from pydantic import BaseModel
+import asyncio
+import json
+from enum import Enum
+
+class ActionStatus(str, Enum):
+    """Status of an action execution"""
+    PENDING = "pending"
+    IN_PROGRESS = "in_progress"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+class Action(BaseModel):
+    """An action that can be executed by the agent"""
+    id: str
+    name: str
+    description: str
+    parameters: Dict[str, Any]
+    status: ActionStatus = ActionStatus.PENDING
+    result: Optional[Dict[str, Any]] = None
+
+class Plan(BaseModel):
+    """A plan consisting of actions to achieve a goal"""
+    goal: str
+    actions: List[Action]
+    context: Dict[str, Any] = {}
+    status: ActionStatus = ActionStatus.PENDING
+
+class Tool(BaseModel):
+    """A tool that can be used by the agent"""
+    name: str
+    description: str
+    parameters: Dict[str, str]
+    handler: Callable[[Dict[str, Any]], Any]
+
+class Memory(BaseModel):
+    """Agent's memory of past actions and results"""
+    plans: List[Plan] = []
+    action_history: List[Action] = []
+    context: Dict[str, Any] = {}
+
+class AutonomousAgent:
+    """An agent that can plan and execute tasks autonomously"""
+    
+    def __init__(
+        self,
+        tools: List[Tool],
+        llm_caller: Callable[[str], str],
+        memory: Optional[Memory] = None
+    ):
+        self.tools = tools
+        self.llm_caller = llm_caller
+        self.memory = memory or Memory()
+    
+    async def _create_plan(
+        self,
+        goal: str,
+        context: Dict[str, Any]
+    ) -> Plan:
+        """Create a plan to achieve a goal"""
+        tools_desc = "\n".join([
+            f"- {tool.name}: {tool.description}\n  Parameters: {tool.parameters}"
+            for tool in self.tools
+        ])
+        
+        prompt = (
+            "Create a plan to achieve the following goal. Use available tools "
+            "and break down the goal into specific actions.\n\n"
+            f"Goal: {goal}\n\n"
+            f"Context: {json.dumps(context, indent=2)}\n\n"
+            f"Available Tools:\n{tools_desc}\n\n"
+            "Previous Actions:\n" +
+            "\n".join([
+                f"- {action.name}: {action.status.value}"
+                for action in self.memory.action_history[-5:]  # Last 5 actions
+            ]) + "\n\n"
+            "Respond with a plan in JSON format:\n"
+            "{\n"
+            '  "actions": [\n'
+            "    {\n"
+            '      "id": "action-1",\n'
+            '      "name": "tool_name",\n'
+            '      "description": "what this action will do",\n'
+            '      "parameters": {}\n'
+            "    }\n"
+            "  ]\n"
+            "}"
+        )
+        
+        try:
+            response = await self.llm_caller(prompt)
+            plan_data = json.loads(response)
+            return Plan(
+                goal=goal,
+                actions=[Action(**action) for action in plan_data["actions"]],
+                context=context
+            )
+        except Exception as e:
+            raise Exception(f"Error creating plan: {str(e)}")
+    
+    def _get_tool(self, name: str) -> Optional[Tool]:
+        """Get a tool by name"""
+        return next((t for t in self.tools if t.name == name), None)
+    
+    async def _execute_action(self, action: Action) -> Action:
+        """Execute a single action"""
+        try:
+            # Get the tool
+            tool = self._get_tool(action.name)
+            if not tool:
+                raise ValueError(f"Tool not found: {action.name}")
+            
+            # Update status
+            action.status = ActionStatus.IN_PROGRESS
+            
+            # Execute tool
+            result = await tool.handler(action.parameters)
+            
+            # Update action with result
+            action.status = ActionStatus.COMPLETED
+            action.result = result
+            
+            # Add to history
+            self.memory.action_history.append(action)
+            
+            return action
+            
+        except Exception as e:
+            action.status = ActionStatus.FAILED
+            action.result = {"error": str(e)}
+            self.memory.action_history.append(action)
+            return action
+    
+    async def _evaluate_progress(
+        self,
+        plan: Plan,
+        completed_actions: List[Action]
+    ) -> Tuple[bool, str]:
+        """Evaluate progress and decide if plan needs adjustment"""
+        prompt = (
+            "Evaluate the progress towards the goal and decide if the plan "
+            "needs adjustment.\n\n"
+            f"Goal: {plan.goal}\n\n"
+            "Completed Actions:\n" +
+            "\n".join([
+                f"- {action.name}: {json.dumps(action.result, indent=2)}"
+                for action in completed_actions
+            ]) + "\n\n"
+            "Respond in JSON format:\n"
+            "{\n"
+            '  "goal_achieved": true/false,\n'
+            '  "reasoning": "explanation"\n'
+            "}"
+        )
+        
+        try:
+            response = await self.llm_caller(prompt)
+            eval_data = json.loads(response)
+            return eval_data["goal_achieved"], eval_data["reasoning"]
+        except Exception as e:
+            raise Exception(f"Error evaluating progress: {str(e)}")
+    
+    async def _adjust_plan(
+        self,
+        plan: Plan,
+        completed_actions: List[Action],
+        evaluation_reason: str
+    ) -> Plan:
+        """Adjust the plan based on progress and results"""
+        prompt = (
+            "Adjust the plan based on completed actions and evaluation. "
+            "Create new actions to achieve the goal.\n\n"
+            f"Goal: {plan.goal}\n\n"
+            "Completed Actions:\n" +
+            "\n".join([
+                f"- {action.name}: {json.dumps(action.result, indent=2)}"
+                for action in completed_actions
+            ]) + "\n\n"
+            f"Evaluation: {evaluation_reason}\n\n"
+            "Available Tools:\n" +
+            "\n".join([
+                f"- {tool.name}: {tool.description}"
+                for tool in self.tools
+            ]) + "\n\n"
+            "Respond with adjusted plan in JSON format:\n"
+            "{\n"
+            '  "actions": [\n'
+            "    {\n"
+            '      "id": "action-1",\n'
+            '      "name": "tool_name",\n'
+            '      "description": "what this action will do",\n'
+            '      "parameters": {}\n'
+            "    }\n"
+            "  ]\n"
+            "}"
+        )
+        
+        try:
+            response = await self.llm_caller(prompt)
+            plan_data = json.loads(response)
+            return Plan(
+                goal=plan.goal,
+                actions=[Action(**action) for action in plan_data["actions"]],
+                context=plan.context
+            )
+        except Exception as e:
+            raise Exception(f"Error adjusting plan: {str(e)}")
+    
+    async def execute(
+        self,
+        goal: str,
+        context: Optional[Dict[str, Any]] = None
+    ) -> List[Action]:
+        """Execute a goal autonomously"""
+        context = context or {}
+        completed_actions = []
+        
+        try:
+            # Create initial plan
+            plan = await self._create_plan(goal, context)
+            self.memory.plans.append(plan)
+            
+            while True:
+                # Execute all actions in current plan
+                for action in plan.actions:
+                    if action.status == ActionStatus.PENDING:
+                        action = await self._execute_action(action)
+                        if action.status == ActionStatus.COMPLETED:
+                            completed_actions.append(action)
+                
+                # Evaluate progress
+                goal_achieved, reason = await self._evaluate_progress(
+                    plan,
+                    completed_actions
+                )
+                
+                if goal_achieved:
+                    break
+                
+                # Adjust plan if needed
+                plan = await self._adjust_plan(
+                    plan,
+                    completed_actions,
+                    reason
+                )
+                self.memory.plans.append(plan)
+            
+            return completed_actions
+            
+        except Exception as e:
+            raise Exception(f"Error executing goal: {str(e)}")
+
+class AgentBuilder:
+    """Helper class to build autonomous agents"""
+    
+    def __init__(self):
+        self.tools: List[Tool] = []
+        self.memory: Optional[Memory] = None
+    
+    def add_tool(
+        self,
+        name: str,
+        description: str,
+        parameters: Dict[str, str],
+        handler: Callable[[Dict[str, Any]], Any]
+    ) -> 'AgentBuilder':
+        """Add a tool to the agent"""
+        self.tools.append(
+            Tool(
+                name=name,
+                description=description,
+                parameters=parameters,
+                handler=handler
+            )
+        )
+        return self
+    
+    def with_memory(self, memory: Memory) -> 'AgentBuilder':
+        """Set agent's memory"""
+        self.memory = memory
+        return self
+    
+    def build(
+        self,
+        llm_caller: Callable[[str], str]
+    ) -> AutonomousAgent:
+        """Build the agent"""
+        return AutonomousAgent(
+            tools=self.tools,
+            llm_caller=llm_caller,
+            memory=self.memory
+        ) 
@@ -0,0 +1,175 @@
+import asyncio
+import os
+from typing import Dict, Any
+from dotenv import load_dotenv
+import anthropic
+from ..agent import AgentBuilder, Memory
+
+async def llm_call(prompt: str) -> str:
+    """Call Anthropic's Claude API"""
+    client = anthropic.Anthropic()
+    response = await client.messages.create(
+        model="claude-3-opus-20240229",
+        max_tokens=1000,
+        temperature=0.7,
+        system="You are a helpful research assistant.",
+        messages=[{"role": "user", "content": prompt}]
+    )
+    return response.content[0].text
+
+# Tool handlers for research tasks
+async def handle_web_search(params: Dict[str, Any]) -> Dict[str, Any]:
+    """Search the web for information (mock implementation)"""
+    query = params["query"]
+    return {
+        "results": [
+            {
+                "title": f"Search result 1 for: {query}",
+                "snippet": "This is a mock search result...",
+                "url": "https://example.com/1"
+            },
+            {
+                "title": f"Search result 2 for: {query}",
+                "snippet": "Another mock search result...",
+                "url": "https://example.com/2"
+            }
+        ]
+    }
+
+async def handle_read_webpage(params: Dict[str, Any]) -> Dict[str, Any]:
+    """Extract content from a webpage (mock implementation)"""
+    url = params["url"]
+    return {
+        "title": "Mock webpage title",
+        "content": f"Mock content extracted from {url}...",
+        "metadata": {
+            "author": "John Doe",
+            "date": "2024-02-15"
+        }
+    }
+
+async def handle_summarize_text(params: Dict[str, Any]) -> Dict[str, Any]:
+    """Summarize text content"""
+    text = params["text"]
+    return {
+        "summary": f"Summary of: {text}",
+        "key_points": [
+            "Key point 1",
+            "Key point 2",
+            "Key point 3"
+        ]
+    }
+
+async def handle_extract_facts(params: Dict[str, Any]) -> Dict[str, Any]:
+    """Extract factual information from text"""
+    text = params["text"]
+    return {
+        "facts": [
+            "Fact 1 extracted from text",
+            "Fact 2 extracted from text",
+            "Fact 3 extracted from text"
+        ],
+        "confidence_scores": [0.9, 0.8, 0.7]
+    }
+
+async def handle_save_notes(params: Dict[str, Any]) -> Dict[str, Any]:
+    """Save research notes (mock implementation)"""
+    notes = params["notes"]
+    return {
+        "saved": True,
+        "location": "research_notes.md",
+        "timestamp": "2024-02-15T12:00:00Z"
+    }
+
+async def main():
+    # Load environment variables
+    load_dotenv()
+    
+    # Initialize memory
+    memory = Memory(
+        context={
+            "research_topic": "artificial intelligence",
+            "focus_areas": [
+                "machine learning",
+                "neural networks",
+                "deep learning"
+            ]
+        }
+    )
+    
+    # Build the research assistant
+    assistant = (
+        AgentBuilder()
+        .add_tool(
+            name="web_search",
+            description="Search the web for information",
+            parameters={
+                "query": "search query string"
+            },
+            handler=handle_web_search
+        )
+        .add_tool(
+            name="read_webpage",
+            description="Extract content from a webpage",
+            parameters={
+                "url": "webpage URL"
+            },
+            handler=handle_read_webpage
+        )
+        .add_tool(
+            name="summarize_text",
+            description="Generate a summary of text content",
+            parameters={
+                "text": "text to summarize"
+            },
+            handler=handle_summarize_text
+        )
+        .add_tool(
+            name="extract_facts",
+            description="Extract factual information from text",
+            parameters={
+                "text": "text to analyze"
+            },
+            handler=handle_extract_facts
+        )
+        .add_tool(
+            name="save_notes",
+            description="Save research notes",
+            parameters={
+                "notes": "notes to save"
+            },
+            handler=handle_save_notes
+        )
+        .with_memory(memory)
+        .build(llm_caller=llm_call)
+    )
+    
+    # Research goals to accomplish
+    goals = [
+        "Research recent advancements in neural networks and summarize key findings",
+        "Investigate applications of deep learning in healthcare and compile examples",
+        "Analyze trends in machine learning research and identify emerging areas"
+    ]
+    
+    # Execute research tasks
+    for goal in goals:
+        print(f"\n{'-' * 50}")
+        print(f"Executing Goal: {goal}")
+        print(f"{'-' * 50}")
+        
+        try:
+            # Execute goal
+            actions = await assistant.execute(goal)
+            
+            # Print results
+            print("\nCompleted Actions:")
+            for action in actions:
+                print(f"\n{action.description}:")
+                print(f"Tool: {action.name}")
+                print(f"Result: {action.result}")
+                
+        except Exception as e:
+            print(f"Error executing goal: {str(e)}")
+
+if __name__ == "__main__":
+    asyncio.run(main()) 
@@ -0,0 +1,282 @@
+# Domain-Specific Agent Pattern
+
+This pattern demonstrates how to create specialized agents with deep domain knowledge, constraints, and behaviors for specific fields.
+
+## Overview
+
+The domain-specific agent pattern is useful when:
+
+- Tasks require specialized domain knowledge
+- Domain-specific constraints must be enforced
+- Context-aware behaviors are needed
+- Domain terminology and relationships matter
+
+## Components
+
+### Core Classes
+
+1. **DomainConstraint**
+
+   - Validation rules
+   - Domain-specific checks
+   - Error messaging
+   - Safety enforcement
+
+2. **DomainKnowledge**
+
+   - Domain facts
+   - Business rules
+   - Terminology
+   - Entity relationships
+
+3. **DomainBehavior**
+
+   - Trigger conditions
+   - Specialized actions
+   - Context awareness
+   - Adaptive responses
+
+4. **DomainMemory**
+
+   - Domain context
+   - Action history
+   - Knowledge updates
+   - State tracking
+
+5. **DomainSpecificAgent**
+   - Domain-aware planning
+   - Constraint validation
+   - Behavior triggering
+   - Knowledge application
+
+## Example Implementation
+
+The `examples/medical_assistant.py` demonstrates a medical diagnosis system:
+
+1. Medical Tools:
+
+   - Symptom analysis
+   - Medical history
+   - Drug interactions
+   - Treatment recommendations
+
+2. Domain Features:
+   - Medical knowledge base
+   - Safety constraints
+   - Clinical behaviors
+   - Patient context
+
+## Usage
+
+1. Set up environment:
+
+```bash
+pip install -r requirements.txt
+```
+
+2. Configure API keys:
+
+```bash
+export ANTHROPIC_API_KEY=your_key_here
+# or
+export OPENAI_API_KEY=your_key_here
+```
+
+3. Run the example:
+
+```bash
+python -m examples.medical_assistant
+```
+
+## Implementation Details
+
+### Domain Constraint Definition
+
+```python
+DomainConstraint(
+    name="constraint_name",
+    description="constraint description",
+    validation_fn=validation_function,
+    error_message="error details"
+)
+```
+
+### Domain Knowledge Definition
+
+```python
+domain_knowledge = {
+    "facts": {...},
+    "rules": {...},
+    "terminology": {...},
+    "relationships": {...}
+}
+```
+
+### Agent Building
+
+```python
+agent = (
+    DomainAgentBuilder(domain_name)
+    .add_tool(...)
+    .add_constraint(...)
+    .add_behavior(...)
+    .with_knowledge(...)
+    .build(llm_caller)
+)
+```
+
+## Domain Capabilities
+
+1. **Knowledge Management**
+
+   - Fact organization
+   - Rule enforcement
+   - Terminology mapping
+   - Relationship tracking
+
+2. **Constraint Handling**
+
+   - Input validation
+   - Safety checks
+   - Domain rules
+   - Error handling
+
+3. **Behavior Management**
+
+   - Context monitoring
+   - Action triggering
+   - Response adaptation
+   - Pattern recognition
+
+4. **Memory Management**
+   - Context preservation
+   - History tracking
+   - Knowledge updates
+   - State management
+
+## Best Practices
+
+1. **Domain Modeling**
+
+   - Clear boundaries
+   - Essential concepts
+   - Key relationships
+   - Core constraints
+
+2. **Knowledge Organization**
+
+   - Structured facts
+   - Clear rules
+   - Standard terminology
+   - Explicit relationships
+
+3. **Constraint Design**
+
+   - Safety first
+   - Clear validation
+   - Helpful errors
+   - Graceful handling
+
+4. **Behavior Implementation**
+   - Clear triggers
+   - Focused actions
+   - Context awareness
+   - Measurable outcomes
+
+## Extensions
+
+Consider extending this pattern with:
+
+1. **Advanced Knowledge**
+
+   - Ontologies
+   - Expert systems
+   - Learning systems
+   - Knowledge graphs
+
+2. **Complex Constraints**
+
+   - Multi-step validation
+   - Dependency checks
+   - Dynamic rules
+   - Compliance tracking
+
+3. **Adaptive Behaviors**
+
+   - Pattern learning
+   - Strategy evolution
+   - Performance optimization
+   - Context adaptation
+
+4. **Enhanced Memory**
+
+   - Long-term storage
+   - Pattern recognition
+   - Knowledge synthesis
+   - Context evolution
+
+5. **Integration**
+   - External systems
+   - Domain services
+   - Data sources
+   - Monitoring tools
+
+## Domain Examples
+
+The pattern can be applied to various domains:
+
+1. **Medical**
+
+   - Diagnosis
+   - Treatment planning
+   - Drug interactions
+   - Patient monitoring
+
+2. **Financial**
+
+   - Risk assessment
+   - Portfolio management
+   - Compliance checking
+   - Fraud detection
+
+3. **Legal**
+
+   - Document analysis
+   - Case research
+   - Compliance checking
+   - Risk assessment
+
+4. **Engineering**
+   - Design validation
+   - Safety analysis
+   - Performance optimization
+   - Quality control
+
+## Safety Considerations
+
+1. **Validation**
+
+   - Input checking
+   - Output verification
+   - Constraint enforcement
+   - Error detection
+
+2. **Monitoring**
+
+   - Action tracking
+   - Performance metrics
+   - Error patterns
+   - Usage analytics
+
+3. **Compliance**
+
+   - Domain regulations
+   - Industry standards
+   - Best practices
+   - Audit trails
+
+4. **Security**
+   - Access control
+   - Data protection
+   - Privacy measures
+   - Secure integration
@@ -0,0 +1,260 @@
+from typing import List, Dict, Any, Callable, Optional, Union, Type
+from pydantic import BaseModel, Field
+from enum import Enum
+import json
+from ..autonomous-agent.agent import AutonomousAgent, Tool, Memory, Action, Plan, ActionStatus
+
+class DomainConstraint(BaseModel):
+    """Represents a domain-specific constraint"""
+    name: str
+    description: str
+    validation_fn: Callable[[Dict[str, Any]], bool]
+    error_message: str
+
+class DomainKnowledge(BaseModel):
+    """Represents domain-specific knowledge"""
+    facts: Dict[str, Any] = {}
+    rules: Dict[str, str] = {}
+    terminology: Dict[str, str] = {}
+    relationships: Dict[str, List[str]] = {}
+
+class DomainBehavior(BaseModel):
+    """Represents domain-specific behavior"""
+    name: str
+    description: str
+    trigger_conditions: Dict[str, Any]
+    action_template: Dict[str, Any]
+
+class DomainMemory(Memory):
+    """Extended memory with domain-specific information"""
+    domain_knowledge: DomainKnowledge = Field(default_factory=DomainKnowledge)
+    domain_specific_history: List[Dict[str, Any]] = []
+
+class DomainSpecificAgent(AutonomousAgent):
+    """An agent specialized for a specific domain"""
+    
+    def __init__(
+        self,
+        domain_name: str,
+        tools: List[Tool],
+        llm_caller: Callable[[str], str],
+        constraints: List[DomainConstraint],
+        behaviors: List[DomainBehavior],
+        domain_knowledge: DomainKnowledge,
+        memory: Optional[DomainMemory] = None
+    ):
+        super().__init__(tools, llm_caller, memory or DomainMemory())
+        self.domain_name = domain_name
+        self.constraints = constraints
+        self.behaviors = behaviors
+        self.domain_knowledge = domain_knowledge
+    
+    async def _validate_action(self, action: Action) -> Tuple[bool, Optional[str]]:
+        """Validate action against domain constraints"""
+        for constraint in self.constraints:
+            try:
+                if not constraint.validation_fn(action.parameters):
+                    return False, constraint.error_message
+            except Exception as e:
+                return False, f"Error validating {constraint.name}: {str(e)}"
+        return True, None
+    
+    async def _enrich_context(self, context: Dict[str, Any]) -> Dict[str, Any]:
+        """Enrich context with domain-specific knowledge"""
+        return {
+            **context,
+            "domain_knowledge": {
+                "facts": self.domain_knowledge.facts,
+                "rules": self.domain_knowledge.rules,
+                "terminology": self.domain_knowledge.terminology,
+                "relationships": self.domain_knowledge.relationships
+            }
+        }
+    
+    async def _check_behaviors(
+        self,
+        action: Action,
+        context: Dict[str, Any]
+    ) -> Optional[Action]:
+        """Check if any domain-specific behaviors should be triggered"""
+        for behavior in self.behaviors:
+            matches = all(
+                context.get(k) == v
+                for k, v in behavior.trigger_conditions.items()
+            )
+            if matches:
+                return Action(
+                    id=f"{action.id}_behavior_{behavior.name}",
+                    name=behavior.name,
+                    description=behavior.description,
+                    parameters=behavior.action_template
+                )
+        return None
+    
+    async def _create_plan(
+        self,
+        goal: str,
+        context: Dict[str, Any]
+    ) -> Plan:
+        """Create a domain-aware plan"""
+        # Enrich context with domain knowledge
+        enriched_context = await self._enrich_context(context)
+        
+        # Add domain-specific prompt elements
+        domain_prompt = (
+            f"You are a specialized agent for the {self.domain_name} domain.\n"
+            "Consider the following domain-specific knowledge:\n"
+            f"Facts: {json.dumps(self.domain_knowledge.facts, indent=2)}\n"
+            f"Rules: {json.dumps(self.domain_knowledge.rules, indent=2)}\n"
+            f"Terminology: {json.dumps(self.domain_knowledge.terminology, indent=2)}\n"
+            "Create a plan that adheres to domain constraints and leverages domain knowledge."
+        )
+        
+        # Create plan with domain awareness
+        plan = await super()._create_plan(
+            goal=f"{domain_prompt}\n\nGoal: {goal}",
+            context=enriched_context
+        )
+        
+        return plan
+    
+    async def _execute_action(self, action: Action) -> Action:
+        """Execute action with domain-specific validation and behaviors"""
+        try:
+            # Validate against domain constraints
+            is_valid, error = await self._validate_action(action)
+            if not is_valid:
+                action.status = ActionStatus.FAILED
+                action.result = {"error": error}
+                return action
+            
+            # Check for triggered behaviors
+            behavior_action = await self._check_behaviors(
+                action,
+                self.memory.context
+            )
+            if behavior_action:
+                # Execute behavior action first
+                behavior_result = await super()._execute_action(behavior_action)
+                if behavior_result.status == ActionStatus.FAILED:
+                    return behavior_result
+            
+            # Execute main action
+            action = await super()._execute_action(action)
+            
+            # Record in domain-specific history
+            if isinstance(self.memory, DomainMemory):
+                self.memory.domain_specific_history.append({
+                    "action": action.dict(),
+                    "domain_context": self.memory.context
+                })
+            
+            return action
+            
+        except Exception as e:
+            action.status = ActionStatus.FAILED
+            action.result = {"error": str(e)}
+            return action
+
+class DomainAgentBuilder:
+    """Helper class to build domain-specific agents"""
+    
+    def __init__(self, domain_name: str):
+        self.domain_name = domain_name
+        self.tools: List[Tool] = []
+        self.constraints: List[DomainConstraint] = []
+        self.behaviors: List[DomainBehavior] = []
+        self.domain_knowledge = DomainKnowledge()
+        self.memory: Optional[DomainMemory] = None
+    
+    def add_tool(
+        self,
+        name: str,
+        description: str,
+        parameters: Dict[str, str],
+        handler: Callable[[Dict[str, Any]], Any]
+    ) -> 'DomainAgentBuilder':
+        """Add a tool to the agent"""
+        self.tools.append(
+            Tool(
+                name=name,
+                description=description,
+                parameters=parameters,
+                handler=handler
+            )
+        )
+        return self
+    
+    def add_constraint(
+        self,
+        name: str,
+        description: str,
+        validation_fn: Callable[[Dict[str, Any]], bool],
+        error_message: str
+    ) -> 'DomainAgentBuilder':
+        """Add a domain constraint"""
+        self.constraints.append(
+            DomainConstraint(
+                name=name,
+                description=description,
+                validation_fn=validation_fn,
+                error_message=error_message
+            )
+        )
+        return self
+    
+    def add_behavior(
+        self,
+        name: str,
+        description: str,
+        trigger_conditions: Dict[str, Any],
+        action_template: Dict[str, Any]
+    ) -> 'DomainAgentBuilder':
+        """Add a domain-specific behavior"""
+        self.behaviors.append(
+            DomainBehavior(
+                name=name,
+                description=description,
+                trigger_conditions=trigger_conditions,
+                action_template=action_template
+            )
+        )
+        return self
+    
+    def with_knowledge(
+        self,
+        facts: Dict[str, Any] = None,
+        rules: Dict[str, str] = None,
+        terminology: Dict[str, str] = None,
+        relationships: Dict[str, List[str]] = None
+    ) -> 'DomainAgentBuilder':
+        """Add domain knowledge"""
+        if facts:
+            self.domain_knowledge.facts.update(facts)
+        if rules:
+            self.domain_knowledge.rules.update(rules)
+        if terminology:
+            self.domain_knowledge.terminology.update(terminology)
+        if relationships:
+            self.domain_knowledge.relationships.update(relationships)
+        return self
+    
+    def with_memory(self, memory: DomainMemory) -> 'DomainAgentBuilder':
+        """Set agent's memory"""
+        self.memory = memory
+        return self
+    
+    def build(
+        self,
+        llm_caller: Callable[[str], str]
+    ) -> DomainSpecificAgent:
+        """Build the domain-specific agent"""
+        return DomainSpecificAgent(
+            domain_name=self.domain_name,
+            tools=self.tools,
+            llm_caller=llm_caller,
+            constraints=self.constraints,
+            behaviors=self.behaviors,
+            domain_knowledge=self.domain_knowledge,
+            memory=self.memory
+        ) 
@@ -0,0 +1,278 @@
+import asyncio
+import os
+from typing import Dict, Any
+from dotenv import load_dotenv
+import anthropic
+from ..agent import DomainAgentBuilder, DomainMemory, DomainKnowledge
+
+async def llm_call(prompt: str) -> str:
+    """Call Anthropic's Claude API"""
+    client = anthropic.Anthropic()
+    response = await client.messages.create(
+        model="claude-3-opus-20240229",
+        max_tokens=1000,
+        temperature=0.7,
+        system="You are a medical diagnosis assistant.",
+        messages=[{"role": "user", "content": prompt}]
+    )
+    return response.content[0].text
+
+# Domain-specific tool handlers
+async def handle_symptom_analysis(params: Dict[str, Any]) -> Dict[str, Any]:
+    """Analyze patient symptoms"""
+    symptoms = params["symptoms"]
+    return {
+        "severity": "moderate",
+        "urgency": "non-emergency",
+        "possible_conditions": [
+            "condition1",
+            "condition2",
+            "condition3"
+        ],
+        "risk_factors": [
+            "risk1",
+            "risk2"
+        ]
+    }
+
+async def handle_medical_history_check(params: Dict[str, Any]) -> Dict[str, Any]:
+    """Check patient medical history"""
+    patient_id = params["patient_id"]
+    return {
+        "previous_conditions": [
+            "condition1",
+            "condition2"
+        ],
+        "allergies": [
+            "allergy1",
+            "allergy2"
+        ],
+        "medications": [
+            "medication1",
+            "medication2"
+        ]
+    }
+
+async def handle_drug_interaction_check(params: Dict[str, Any]) -> Dict[str, Any]:
+    """Check for potential drug interactions"""
+    medications = params["medications"]
+    return {
+        "interactions": [
+            {
+                "drugs": ["drug1", "drug2"],
+                "severity": "high",
+                "recommendation": "avoid combination"
+            }
+        ],
+        "alternatives": [
+            "alt_drug1",
+            "alt_drug2"
+        ]
+    }
+
+async def handle_treatment_recommendation(params: Dict[str, Any]) -> Dict[str, Any]:
+    """Generate treatment recommendations"""
+    condition = params["condition"]
+    patient_data = params["patient_data"]
+    return {
+        "primary_treatment": "treatment1",
+        "alternatives": [
+            "treatment2",
+            "treatment3"
+        ],
+        "lifestyle_changes": [
+            "change1",
+            "change2"
+        ],
+        "follow_up": "2 weeks"
+    }
+
+# Domain-specific constraints
+def validate_patient_data(params: Dict[str, Any]) -> bool:
+    """Validate patient data completeness"""
+    required_fields = ["age", "gender", "symptoms"]
+    return all(field in params for field in required_fields)
+
+def validate_medication_safety(params: Dict[str, Any]) -> bool:
+    """Validate medication safety"""
+    if "medications" not in params:
+        return True
+    # Mock safety check
+    unsafe_medications = ["unsafe_drug1", "unsafe_drug2"]
+    return not any(med in unsafe_medications for med in params["medications"])
+
+# Medical domain knowledge
+medical_knowledge = {
+    "facts": {
+        "common_conditions": [
+            "hypertension",
+            "diabetes",
+            "asthma"
+        ],
+        "vital_signs": {
+            "normal_bp": "120/80",
+            "normal_temp": "98.6F",
+            "normal_hr": "60-100"
+        }
+    },
+    "rules": {
+        "emergency_symptoms": "Chest pain, difficulty breathing, or severe bleeding require immediate emergency care",
+        "prescription_requirements": "Controlled substances require proper documentation and authorization",
+        "follow_up_timing": "Critical conditions require follow-up within 24-48 hours"
+    },
+    "terminology": {
+        "bp": "Blood Pressure",
+        "hr": "Heart Rate",
+        "bmi": "Body Mass Index"
+    },
+    "relationships": {
+        "diabetes": ["blood_sugar", "insulin", "diet"],
+        "hypertension": ["blood_pressure", "sodium", "stress"]
+    }
+}
+
+# Domain-specific behaviors
+diagnosis_behavior = {
+    "name": "comprehensive_diagnosis",
+    "description": "Perform comprehensive diagnosis when multiple symptoms present",
+    "trigger_conditions": {
+        "symptom_count": ">3",
+        "severity": "high"
+    },
+    "action_template": {
+        "name": "detailed_analysis",
+        "parameters": {
+            "include_specialists": True,
+            "run_additional_tests": True
+        }
+    }
+}
+
+async def main():
+    # Load environment variables
+    load_dotenv()
+    
+    # Initialize domain memory
+    memory = DomainMemory(
+        context={
+            "facility_type": "primary_care",
+            "available_specialists": ["cardiology", "neurology", "endocrinology"],
+            "emergency_protocols": ["protocol1", "protocol2"]
+        }
+    )
+    
+    # Build the medical assistant
+    assistant = (
+        DomainAgentBuilder("medical_diagnosis")
+        .add_tool(
+            name="symptom_analysis",
+            description="Analyze patient symptoms and determine possible conditions",
+            parameters={
+                "symptoms": "list of symptoms"
+            },
+            handler=handle_symptom_analysis
+        )
+        .add_tool(
+            name="medical_history",
+            description="Check patient medical history",
+            parameters={
+                "patient_id": "patient identifier"
+            },
+            handler=handle_medical_history_check
+        )
+        .add_tool(
+            name="drug_interaction",
+            description="Check for potential drug interactions",
+            parameters={
+                "medications": "list of medications"
+            },
+            handler=handle_drug_interaction_check
+        )
+        .add_tool(
+            name="treatment_recommendation",
+            description="Generate treatment recommendations",
+            parameters={
+                "condition": "diagnosed condition",
+                "patient_data": "patient information"
+            },
+            handler=handle_treatment_recommendation
+        )
+        .add_constraint(
+            name="patient_data_validation",
+            description="Ensure all required patient data is provided",
+            validation_fn=validate_patient_data,
+            error_message="Missing required patient information"
+        )
+        .add_constraint(
+            name="medication_safety",
+            description="Ensure medication safety",
+            validation_fn=validate_medication_safety,
+            error_message="Unsafe medication detected"
+        )
+        .add_behavior(
+            name=diagnosis_behavior["name"],
+            description=diagnosis_behavior["description"],
+            trigger_conditions=diagnosis_behavior["trigger_conditions"],
+            action_template=diagnosis_behavior["action_template"]
+        )
+        .with_knowledge(**medical_knowledge)
+        .with_memory(memory)
+        .build(llm_caller=llm_call)
+    )
+    
+    # Example cases to diagnose
+    cases = [
+        {
+            "goal": "Diagnose patient with symptoms: fever, cough, fatigue",
+            "context": {
+                "patient_data": {
+                    "age": 45,
+                    "gender": "F",
+                    "symptoms": ["fever", "cough", "fatigue"],
+                    "vitals": {
+                        "temperature": "101.2F",
+                        "blood_pressure": "128/82",
+                        "heart_rate": 88
+                    }
+                }
+            }
+        },
+        {
+            "goal": "Recommend treatment for diagnosed hypertension",
+            "context": {
+                "patient_data": {
+                    "age": 62,
+                    "gender": "M",
+                    "condition": "hypertension",
+                    "medications": ["lisinopril"],
+                    "blood_pressure": "158/94"
+                }
+            }
+        }
+    ]
+    
+    # Process cases
+    for case in cases:
+        print(f"\n{'-' * 50}")
+        print(f"Processing Case: {case['goal']}")
+        print(f"{'-' * 50}")
+        
+        try:
+            # Execute diagnosis/treatment
+            actions = await assistant.execute(
+                goal=case["goal"],
+                context=case["context"]
+            )
+            
+            # Print results
+            print("\nCompleted Actions:")
+            for action in actions:
+                print(f"\n{action.description}:")
+                print(f"Tool: {action.name}")
+                print(f"Result: {action.result}")
+                
+        except Exception as e:
+            print(f"Error processing case: {str(e)}")
+
+if __name__ == "__main__":
+    asyncio.run(main()) 
@@ -0,0 +1,129 @@
+# Augmented LLM Pattern
+
+This directory demonstrates the Augmented LLM pattern, which enhances a base LLM with additional capabilities like tools and memory.
+
+## Components
+
+### Core Components
+
+- `augmented.py`: The main implementation of the Augmented LLM pattern
+  - `Tool`: Abstract base class for implementing tools
+  - `AugmentedLLM`: Main class that combines LLM with tools and memory
+
+### Tools
+
+Located in `tools/basic_tools.py`:
+
+- `WebSearchTool`: Mock implementation of web search
+- `CalculatorTool`: Basic calculator functionality
+- `WeatherTool`: Mock weather information retrieval
+
+### Examples
+
+Located in `examples/basic_usage.py`:
+
+- Demonstrates how to initialize and use the Augmented LLM
+- Shows tool usage and memory capabilities
+- Includes sample conversation flow
+
+## Usage
+
+1. Set up environment variables:
+
+```bash
+ANTHROPIC_API_KEY=your_key_here  # For Claude
+OPENAI_API_KEY=your_key_here     # For GPT-4
+```
+
+2. Install dependencies:
+
+```bash
+pip install -r requirements.txt
+```
+
+3. Run the example:
+
+```bash
+python -m examples.basic_usage
+```
+
+## Key Features
+
+1. **Multiple LLM Provider Support**
+
+   - Anthropic Claude
+   - OpenAI GPT-4
+   - Extensible for other providers
+
+2. **Tool Integration**
+
+   - Abstract tool interface
+   - Easy to add new tools
+   - Tool description formatting for prompts
+
+3. **Memory Support**
+   - Optional conversation history
+   - Contextual awareness in responses
+   - Memory formatting for prompts
+
+## Implementation Details
+
+### Tool Integration
+
+Tools are implemented as classes inheriting from the `Tool` base class:
+
+```python
+class Tool:
+    def __init__(self, name: str, description: str):
+        self.name = name
+        self.description = description
+
+    @abstractmethod
+    async def execute(self, **kwargs) -> str:
+        pass
+```
+
+### Memory Management
+
+Memory is implemented as a list of conversation turns:
+
+```python
+self.memory = [] if enable_memory else None
+```
+
+### Provider-Specific Processing
+
+Each provider has its own processing method:
+
+- `_process_anthropic`: For Claude
+- `_process_openai`: For GPT-4
+
+## Best Practices
+
+1. **Tool Development**
+
+   - Keep tools focused and single-purpose
+   - Provide clear descriptions
+   - Handle errors gracefully
+   - Mock external services in examples
+
+2. **Memory Usage**
+
+   - Enable only when needed
+   - Consider memory limitations
+   - Clear memory when appropriate
+
+3. **Error Handling**
+   - Graceful API error handling
+   - Tool execution error handling
+   - Clear error messages
+
+## Extensions
+
+Consider extending this pattern with:
+
+1. Tool result caching
+2. Parallel tool execution
+3. Tool chain orchestration
+4. Structured memory management
+5. Result validation
@@ -0,0 +1,39 @@
+import asyncio
+import os
+from dotenv import load_dotenv
+from ..augmented import AugmentedLLM
+from ..tools.basic_tools import WebSearchTool, CalculatorTool, WeatherTool
+
+async def main():
+    # Load environment variables
+    load_dotenv()
+    
+    # Initialize tools
+    tools = [
+        WebSearchTool(),
+        CalculatorTool(),
+        WeatherTool()
+    ]
+    
+    # Initialize augmented LLM with tools and memory
+    llm = AugmentedLLM(
+        provider="anthropic",  # or "openai"
+        tools=tools,
+        enable_memory=True
+    )
+    
+    # Example conversation
+    queries = [
+        "What's 234 * 456?",
+        "What's the weather like in London?",
+        "Can you search for information about quantum computing?",
+        "What did I ask about earlier regarding calculations?"
+    ]
+    
+    for query in queries:
+        print(f"\nUser: {query}")
+        response = await llm.process(query)
+        print(f"Assistant: {response}")
+
+if __name__ == "__main__":
+    asyncio.run(main()) 
@@ -0,0 +1,43 @@
+from typing import Dict, Any
+import aiohttp
+import json
+from ..augmented import Tool
+
+class WebSearchTool(Tool):
+    def __init__(self):
+        super().__init__(
+            name="web_search",
+            description="Search the web for information about a query"
+        )
+    
+    async def execute(self, query: str) -> str:
+        # This is a mock implementation
+        return f"Mock web search results for: {query}"
+
+class CalculatorTool(Tool):
+    def __init__(self):
+        super().__init__(
+            name="calculator",
+            description="Perform basic mathematical calculations"
+        )
+    
+    async def execute(self, expression: str) -> str:
+        try:
+            # WARNING: eval is used here for demonstration. In production,
+            # use a safer method to evaluate mathematical expressions
+            result = eval(expression, {"__builtins__": {}})
+            return f"Result: {result}"
+        except Exception as e:
+            return f"Error calculating {expression}: {str(e)}"
+
+class WeatherTool(Tool):
+    def __init__(self, api_key: str = None):
+        super().__init__(
+            name="weather",
+            description="Get current weather for a location"
+        )
+        self.api_key = api_key
+    
+    async def execute(self, location: str) -> str:
+        # This would normally use a real weather API
+        return f"Mock weather data for {location}: 22°C, Partly Cloudy" 
@@ -0,0 +1,11 @@
+anthropic>=0.7.0
+openai>=1.0.0
+google-cloud-aiplatform>=1.36.0
+azure-openai>=1.0.0
+python-dotenv>=1.0.0
+aiohttp>=3.9.0
+pytest>=7.4.0
+pydantic>=2.0.0
+tenacity>=8.2.0
+typing-extensions>=4.8.0
+rich>=13.0.0 
@@ -0,0 +1,204 @@
+# Evaluator-Optimizer Pattern
+
+This pattern demonstrates how to implement iterative improvement through evaluation and optimization feedback loops.
+
+## Overview
+
+The evaluator-optimizer pattern is useful when:
+
+- Solutions need iterative refinement
+- Quality can be evaluated systematically
+- Multiple optimization strategies may be applicable
+- Progress can be measured quantitatively
+
+## Components
+
+### Core Classes
+
+1. **Candidate**
+
+   - Represents a potential solution
+   - Tracks evaluation metrics
+   - Maintains iteration history
+   - Stores feedback data
+
+2. **Optimizer**
+
+   - Manages optimization process
+   - Implements multiple strategies
+   - Evaluates candidates
+   - Generates improvements
+
+3. **OptimizerBuilder**
+   - Configures optimization process
+   - Sets strategy parameters
+   - Defines evaluation criteria
+   - Manages optimization limits
+
+## Example Implementation
+
+The `examples/code_optimization.py` demonstrates code improvement:
+
+1. Optimization Strategies:
+
+   - Iterative Improvement
+   - Parallel Variations
+   - Tournament Selection
+
+2. Evaluation Criteria:
+
+   - Performance
+   - Readability
+   - Maintainability
+   - Error Handling
+
+3. Features:
+   - Multiple optimization strategies
+   - Weighted evaluation criteria
+   - Progress tracking
+   - Detailed feedback
+
+## Usage
+
+1. Set up environment:
+
+```bash
+pip install -r requirements.txt
+```
+
+2. Configure API keys:
+
+```bash
+export ANTHROPIC_API_KEY=your_key_here
+# or
+export OPENAI_API_KEY=your_key_here
+```
+
+3. Run the example:
+
+```bash
+python -m examples.code_optimization
+```
+
+## Implementation Details
+
+### Candidate Definition
+
+```python
+Candidate(
+    id="candidate-id",
+    content="solution content",
+    score=0.85,
+    feedback="evaluation feedback"
+)
+```
+
+### Optimizer Configuration
+
+```python
+optimizer = (
+    OptimizerBuilder()
+    .set_strategy(strategy)
+    .set_max_iterations(5)
+    .set_score_threshold(0.95)
+    .build(llm_caller)
+)
+```
+
+### Optimization
+
+```python
+best, history = await optimizer.optimize(
+    initial_solution,
+    evaluation_criteria
+)
+```
+
+## Optimization Strategies
+
+1. **Iterative**
+
+   - Sequential improvements
+   - Direct feedback incorporation
+   - Step-by-step refinement
+   - Linear progression
+
+2. **Parallel**
+
+   - Multiple variations
+   - Concurrent evaluation
+   - Best candidate selection
+   - Broader exploration
+
+3. **Tournament**
+   - Population-based
+   - Competitive selection
+   - Genetic-style evolution
+   - Diverse solutions
+
+## Best Practices
+
+1. **Evaluation Design**
+
+   - Define clear criteria
+   - Use quantitative metrics
+   - Weight importance factors
+   - Consider trade-offs
+
+2. **Strategy Selection**
+
+   - Match problem characteristics
+   - Consider resource constraints
+   - Balance exploration/exploitation
+   - Adapt to feedback
+
+3. **Termination Conditions**
+
+   - Set realistic thresholds
+   - Limit iteration count
+   - Monitor improvement rate
+   - Handle convergence
+
+4. **Result Analysis**
+   - Track improvement history
+   - Analyze feedback patterns
+   - Document optimizations
+   - Validate improvements
+
+## Extensions
+
+Consider extending this pattern with:
+
+1. **Advanced Strategies**
+
+   - Hybrid approaches
+   - Adaptive methods
+   - Multi-objective optimization
+   - Constraint handling
+
+2. **Evaluation Enhancement**
+
+   - Automated testing
+   - Performance profiling
+   - Quality metrics
+   - Validation suites
+
+3. **Learning**
+
+   - Strategy adaptation
+   - Pattern recognition
+   - Historical learning
+   - Meta-optimization
+
+4. **Visualization**
+
+   - Progress tracking
+   - Strategy comparison
+   - Result analysis
+   - Performance metrics
+
+5. **Integration**
+   - CI/CD pipelines
+   - Version control
+   - Issue tracking
+   - Documentation generation
@@ -0,0 +1,98 @@
+import asyncio
+import os
+from typing import Dict, Any
+from dotenv import load_dotenv
+import anthropic
+from ..optimizer import OptimizerBuilder, OptimizationStrategy
+
+async def llm_call(prompt: str) -> str:
+    """Call Anthropic's Claude API"""
+    client = anthropic.Anthropic()
+    response = await client.messages.create(
+        model="claude-3-opus-20240229",
+        max_tokens=1000,
+        temperature=0.7,
+        system="You are a helpful code optimization assistant.",
+        messages=[{"role": "user", "content": prompt}]
+    )
+    return response.content[0].text
+
+async def main():
+    # Load environment variables
+    load_dotenv()
+    
+    # Initial code to optimize
+    initial_code = """
+    def fibonacci(n):
+        if n <= 0:
+            return []
+        elif n == 1:
+            return [0]
+        
+        sequence = [0, 1]
+        while len(sequence) < n:
+            sequence.append(sequence[-1] + sequence[-2])
+        return sequence
+    
+    def find_fibonacci_sum(n):
+        sequence = fibonacci(n)
+        return sum(sequence)
+    """
+    
+    # Evaluation criteria
+    criteria = {
+        "performance": {
+            "description": "Code should be efficient in terms of time and space complexity",
+            "weight": 0.4
+        },
+        "readability": {
+            "description": "Code should be clear, well-documented, and follow Python best practices",
+            "weight": 0.3
+        },
+        "maintainability": {
+            "description": "Code should be easy to modify and extend",
+            "weight": 0.2
+        },
+        "error_handling": {
+            "description": "Code should handle edge cases and invalid inputs gracefully",
+            "weight": 0.1
+        }
+    }
+    
+    # Create optimizer with different strategies
+    strategies = [
+        ("Iterative Optimization", OptimizationStrategy.ITERATIVE),
+        ("Parallel Optimization", OptimizationStrategy.PARALLEL),
+        ("Tournament Optimization", OptimizationStrategy.TOURNAMENT)
+    ]
+    
+    for strategy_name, strategy in strategies:
+        print(f"\n{'-' * 50}")
+        print(f"Running {strategy_name}")
+        print(f"{'-' * 50}")
+        
+        optimizer = (
+            OptimizerBuilder()
+            .set_strategy(strategy)
+            .set_max_iterations(3)
+            .set_score_threshold(0.95)
+            .set_population_size(3)
+            .build(llm_caller=llm_call)
+        )
+        
+        try:
+            # Optimize code
+            best, history = await optimizer.optimize(initial_code, criteria)
+            
+            print(f"\nOptimization completed after {len(history)} iterations")
+            print(f"Final score: {best.score}")
+            print("\nOptimized Code:")
+            print(best.content)
+            print("\nFeedback:")
+            print(best.feedback)
+            
+        except Exception as e:
+            print(f"Error during optimization: {str(e)}")
+
+if __name__ == "__main__":
+    asyncio.run(main()) 
@@ -0,0 +1,314 @@
+from typing import List, Dict, Any, Callable, Optional, Union, Tuple
+from abc import ABC, abstractmethod
+from pydantic import BaseModel
+import asyncio
+from enum import Enum
+
+class OptimizationStrategy(str, Enum):
+    """Strategies for optimization"""
+    ITERATIVE = "iterative"
+    PARALLEL = "parallel"
+    TOURNAMENT = "tournament"
+
+class Candidate(BaseModel):
+    """A candidate solution"""
+    id: str
+    content: str
+    score: Optional[float] = None
+    feedback: Optional[str] = None
+    iteration: int = 0
+
+class EvaluationResult(BaseModel):
+    """Result of evaluating a candidate"""
+    score: float
+    feedback: str
+    improvements: List[str]
+
+class OptimizerConfig(BaseModel):
+    """Configuration for the optimizer"""
+    max_iterations: int = 5
+    score_threshold: float = 0.9
+    optimization_strategy: OptimizationStrategy = OptimizationStrategy.ITERATIVE
+    population_size: int = 3
+    tournament_size: int = 2
+
+class Optimizer:
+    """Improves solutions through evaluation and optimization"""
+    
+    def __init__(
+        self,
+        llm_caller: Callable[[str], str],
+        config: Optional[OptimizerConfig] = None
+    ):
+        self.llm_caller = llm_caller
+        self.config = config or OptimizerConfig()
+        self.candidates: List[Candidate] = []
+    
+    async def _evaluate_candidate(
+        self,
+        candidate: Candidate,
+        evaluation_criteria: Dict[str, Any]
+    ) -> EvaluationResult:
+        """Evaluate a candidate solution"""
+        prompt = (
+            "Evaluate this solution based on the given criteria. Provide:\n"
+            "1. A score between 0 and 1\n"
+            "2. Specific feedback\n"
+            "3. Suggested improvements\n\n"
+            f"Solution:\n{candidate.content}\n\n"
+            f"Criteria:\n{evaluation_criteria}\n\n"
+            "Respond in JSON format:\n"
+            "{\n"
+            '  "score": 0.8,\n'
+            '  "feedback": "Detailed feedback here",\n'
+            '  "improvements": ["improvement 1", "improvement 2"]\n'
+            "}"
+        )
+        
+        try:
+            response = await self.llm_caller(prompt)
+            return EvaluationResult.parse_raw(response)
+        except Exception as e:
+            raise Exception(f"Error evaluating candidate: {str(e)}")
+    
+    async def _optimize_candidate(
+        self,
+        candidate: Candidate,
+        evaluation_result: EvaluationResult
+    ) -> Candidate:
+        """Generate an improved version of the candidate"""
+        prompt = (
+            "Improve this solution based on the evaluation feedback and "
+            "suggested improvements. Maintain the same format and structure, "
+            "but address the identified issues.\n\n"
+            f"Original Solution:\n{candidate.content}\n\n"
+            f"Feedback:\n{evaluation_result.feedback}\n\n"
+            "Suggested Improvements:\n" +
+            "\n".join(f"- {imp}" for imp in evaluation_result.improvements) +
+            "\n\nImproved Solution:"
+        )
+        
+        try:
+            improved_content = await self.llm_caller(prompt)
+            return Candidate(
+                id=f"{candidate.id}-v{candidate.iteration + 1}",
+                content=improved_content,
+                iteration=candidate.iteration + 1
+            )
+        except Exception as e:
+            raise Exception(f"Error optimizing candidate: {str(e)}")
+    
+    async def _generate_variations(
+        self,
+        candidate: Candidate,
+        num_variations: int
+    ) -> List[Candidate]:
+        """Generate multiple variations of a candidate"""
+        prompt = (
+            f"Generate {num_variations} different variations of this solution. "
+            "Each variation should maintain the same basic structure but vary in "
+            "approach or implementation details.\n\n"
+            f"Original Solution:\n{candidate.content}\n\n"
+            "Respond with each variation separated by '---'\n"
+        )
+        
+        try:
+            response = await self.llm_caller(prompt)
+            variations = [v.strip() for v in response.split("---")]
+            return [
+                Candidate(
+                    id=f"{candidate.id}-var{i}",
+                    content=content,
+                    iteration=candidate.iteration
+                )
+                for i, content in enumerate(variations[:num_variations], 1)
+            ]
+        except Exception as e:
+            raise Exception(f"Error generating variations: {str(e)}")
+    
+    async def _run_tournament(
+        self,
+        candidates: List[Candidate],
+        evaluation_criteria: Dict[str, Any]
+    ) -> List[Candidate]:
+        """Run a tournament to select the best candidates"""
+        # Randomly pair candidates for tournament rounds
+        import random
+        tournament_pairs = []
+        available = candidates.copy()
+        while len(available) >= 2:
+            pair = random.sample(available, 2)
+            tournament_pairs.append(pair)
+            for p in pair:
+                available.remove(p)
+        
+        # If odd number, add last one to winners
+        winners = available
+        
+        # Evaluate pairs and select winners
+        for pair in tournament_pairs:
+            results = await asyncio.gather(*[
+                self._evaluate_candidate(c, evaluation_criteria)
+                for c in pair
+            ])
+            # Add winner to next round
+            winner = pair[0] if results[0].score > results[1].score else pair[1]
+            winners.append(winner)
+        
+        return winners
+    
+    async def optimize(
+        self,
+        initial_solution: str,
+        evaluation_criteria: Dict[str, Any]
+    ) -> Tuple[Candidate, List[Candidate]]:
+        """Optimize a solution through iterative improvement"""
+        # Initialize first candidate
+        current = Candidate(
+            id="solution-v0",
+            content=initial_solution,
+            iteration=0
+        )
+        
+        history = [current]
+        
+        if self.config.optimization_strategy == OptimizationStrategy.ITERATIVE:
+            # Iterative improvement
+            for i in range(self.config.max_iterations):
+                # Evaluate current candidate
+                eval_result = await self._evaluate_candidate(
+                    current,
+                    evaluation_criteria
+                )
+                
+                # Update candidate with evaluation results
+                current.score = eval_result.score
+                current.feedback = eval_result.feedback
+                
+                # Check if good enough
+                if current.score >= self.config.score_threshold:
+                    break
+                
+                # Generate improved version
+                current = await self._optimize_candidate(current, eval_result)
+                history.append(current)
+                
+        elif self.config.optimization_strategy == OptimizationStrategy.PARALLEL:
+            # Parallel variations and improvement
+            for i in range(self.config.max_iterations):
+                # Generate variations
+                variations = await self._generate_variations(
+                    current,
+                    self.config.population_size
+                )
+                
+                # Evaluate all variations
+                eval_results = await asyncio.gather(*[
+                    self._evaluate_candidate(v, evaluation_criteria)
+                    for v in variations
+                ])
+                
+                # Update variations with scores
+                for var, result in zip(variations, eval_results):
+                    var.score = result.score
+                    var.feedback = result.feedback
+                
+                # Select best variation
+                best = max(variations, key=lambda x: x.score)
+                if best.score >= self.config.score_threshold:
+                    current = best
+                    history.extend(variations)
+                    break
+                
+                # Improve best variation
+                current = await self._optimize_candidate(
+                    best,
+                    eval_results[variations.index(best)]
+                )
+                history.extend(variations + [current])
+                
+        else:  # TOURNAMENT
+            # Tournament-based optimization
+            current_population = [current]
+            
+            for i in range(self.config.max_iterations):
+                # Generate variations for tournament
+                variations = await self._generate_variations(
+                    current,
+                    self.config.population_size - 1
+                )
+                current_population.extend(variations)
+                
+                # Run tournament
+                winners = await self._run_tournament(
+                    current_population,
+                    evaluation_criteria
+                )
+                
+                # Select best winner
+                best = winners[0]
+                for w in winners[1:]:
+                    eval_result = await self._evaluate_candidate(
+                        w,
+                        evaluation_criteria
+                    )
+                    w.score = eval_result.score
+                    w.feedback = eval_result.feedback
+                    if w.score > (best.score or 0):
+                        best = w
+                
+                if best.score >= self.config.score_threshold:
+                    current = best
+                    history.extend(current_population)
+                    break
+                
+                # Use best as seed for next generation
+                current = best
+                current_population = [current]
+                history.extend(current_population)
+        
+        return current, history
+
+class OptimizerBuilder:
+    """Helper class to build optimizers"""
+    
+    def __init__(self):
+        self.config = OptimizerConfig()
+    
+    def set_max_iterations(self, max_iterations: int) -> 'OptimizerBuilder':
+        """Set maximum number of optimization iterations"""
+        self.config.max_iterations = max_iterations
+        return self
+    
+    def set_score_threshold(self, threshold: float) -> 'OptimizerBuilder':
+        """Set score threshold for early stopping"""
+        self.config.score_threshold = threshold
+        return self
+    
+    def set_strategy(
+        self,
+        strategy: OptimizationStrategy
+    ) -> 'OptimizerBuilder':
+        """Set optimization strategy"""
+        self.config.optimization_strategy = strategy
+        return self
+    
+    def set_population_size(self, size: int) -> 'OptimizerBuilder':
+        """Set population size for parallel/tournament strategies"""
+        self.config.population_size = size
+        return self
+    
+    def set_tournament_size(self, size: int) -> 'OptimizerBuilder':
+        """Set tournament size for tournament strategy"""
+        self.config.tournament_size = size
+        return self
+    
+    def build(
+        self,
+        llm_caller: Callable[[str], str]
+    ) -> Optimizer:
+        """Build the optimizer"""
+        return Optimizer(
+            llm_caller=llm_caller,
+            config=self.config
+        ) 
@@ -0,0 +1,192 @@
+# Orchestrator-Workers Pattern
+
+This pattern demonstrates how to implement dynamic task decomposition and parallel execution using an orchestrator and specialized workers.
+
+## Overview
+
+The orchestrator-workers pattern is useful when:
+
+- Complex tasks need to be broken down dynamically
+- Different subtasks require specialized handling
+- Tasks have dependencies and execution order constraints
+- Parallel execution can improve performance
+
+## Components
+
+### Core Classes
+
+1. **Task**
+
+   - Represents a unit of work
+   - Tracks status and dependencies
+   - Stores execution results
+   - Manages task metadata
+
+2. **Worker**
+
+   - Handles specific task types
+   - Executes task logic
+   - Manages concurrency
+   - Reports task results
+
+3. **Orchestrator**
+
+   - Decomposes complex tasks
+   - Manages task dependencies
+   - Delegates to workers
+   - Coordinates execution
+
+4. **OrchestratorBuilder**
+   - Configures worker pool
+   - Sets up task routing
+   - Manages concurrency limits
+   - Simplifies orchestrator creation
+
+## Example Implementation
+
+The `examples/document_processing.py` demonstrates document analysis:
+
+1. Worker Types:
+
+   - Text Extraction
+   - Language Detection
+   - Summarization
+   - Topic Extraction
+   - Sentiment Analysis
+   - Report Formatting
+
+2. Features:
+   - Dynamic task decomposition
+   - Parallel processing
+   - Dependency management
+   - Result aggregation
+
+## Usage
+
+1. Set up environment:
+
+```bash
+pip install -r requirements.txt
+```
+
+2. Configure API keys:
+
+```bash
+export ANTHROPIC_API_KEY=your_key_here
+# or
+export OPENAI_API_KEY=your_key_here
+```
+
+3. Run the example:
+
+```bash
+python -m examples.document_processing
+```
+
+## Implementation Details
+
+### Task Definition
+
+```python
+Task(
+    id="task-id",
+    type="task-type",
+    input_data={},
+    dependencies=["dep-1", "dep-2"]
+)
+```
+
+### Worker Configuration
+
+```python
+Worker(
+    name="worker-name",
+    task_types=["type-1", "type-2"],
+    handler=async_handler_function,
+    concurrency_limit=2
+)
+```
+
+### Orchestrator Building
+
+```python
+orchestrator = (
+    OrchestratorBuilder()
+    .add_worker(...)
+    .add_worker(...)
+    .build(llm_caller)
+)
+```
+
+### Execution
+
+```python
+results = await orchestrator.execute(input_data)
+```
+
+## Best Practices
+
+1. **Task Design**
+
+   - Keep tasks focused
+   - Define clear interfaces
+   - Handle errors gracefully
+   - Document dependencies
+
+2. **Worker Implementation**
+
+   - Manage resources efficiently
+   - Set appropriate concurrency
+   - Handle timeouts
+   - Validate inputs/outputs
+
+3. **Dependency Management**
+
+   - Avoid circular dependencies
+   - Handle missing dependencies
+   - Track task status
+   - Clean up resources
+
+4. **Error Handling**
+   - Handle worker failures
+   - Manage task timeouts
+   - Provide fallback options
+   - Log execution details
+
+## Extensions
+
+Consider extending this pattern with:
+
+1. **Advanced Orchestration**
+
+   - Dynamic worker allocation
+   - Load balancing
+   - Priority queues
+   - Task scheduling
+
+2. **Monitoring**
+
+   - Task progress tracking
+   - Worker performance metrics
+   - Resource utilization
+   - Error reporting
+
+3. **Resilience**
+
+   - Task retries
+   - Circuit breaking
+   - Fallback strategies
+   - State recovery
+
+4. **Optimization**
+
+   - Task batching
+   - Resource pooling
+   - Result caching
+   - Parallel execution
+
+5. **Integration**
+   - External services
+   - Message queues
+   - Storage systems
+   - Monitoring tools
@@ -0,0 +1,149 @@
+import asyncio
+import os
+from typing import Dict, Any
+from dotenv import load_dotenv
+import anthropic
+from ..orchestrator import OrchestratorBuilder
+
+async def llm_call(prompt: str) -> str:
+    """Call Anthropic's Claude API"""
+    client = anthropic.Anthropic()
+    response = await client.messages.create(
+        model="claude-3-opus-20240229",
+        max_tokens=1000,
+        temperature=0.7,
+        system="You are a helpful task decomposition assistant.",
+        messages=[{"role": "user", "content": prompt}]
+    )
+    return response.content[0].text
+
+# Worker handlers for different document processing tasks
+async def handle_text_extraction(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Extract text from document (mock implementation)"""
+    document = data["document"]
+    return {
+        "text": f"Extracted text from {document}",
+        "pages": 5,
+        "language": "en"
+    }
+
+async def handle_language_detection(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Detect language of text"""
+    text = data["text"]
+    return {
+        "language": "en",
+        "confidence": 0.95
+    }
+
+async def handle_summarization(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Summarize text content"""
+    text = data["text"]
+    return {
+        "summary": f"Summary of: {text}",
+        "length": "medium"
+    }
+
+async def handle_topic_extraction(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Extract main topics from text"""
+    text = data["text"]
+    return {
+        "topics": ["topic1", "topic2", "topic3"],
+        "confidence_scores": [0.9, 0.8, 0.7]
+    }
+
+async def handle_sentiment_analysis(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Analyze sentiment of text"""
+    text = data["text"]
+    return {
+        "sentiment": "positive",
+        "score": 0.8,
+        "aspects": {
+            "tone": "professional",
+            "emotion": "confident"
+        }
+    }
+
+async def handle_formatting(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Format the final document analysis report"""
+    return {
+        "report": {
+            "summary": data["summary"],
+            "topics": data["topics"],
+            "sentiment": data["sentiment"],
+            "language": data["language"]
+        },
+        "format": "json"
+    }
+
+async def main():
+    # Load environment variables
+    load_dotenv()
+    
+    # Build the document processor
+    processor = (
+        OrchestratorBuilder()
+        .add_worker(
+            name="extractor",
+            task_types=["text_extraction"],
+            handler=handle_text_extraction,
+            concurrency_limit=2
+        )
+        .add_worker(
+            name="language_detector",
+            task_types=["language_detection"],
+            handler=handle_language_detection,
+            concurrency_limit=1
+        )
+        .add_worker(
+            name="summarizer",
+            task_types=["summarization"],
+            handler=handle_summarization,
+            concurrency_limit=1
+        )
+        .add_worker(
+            name="topic_extractor",
+            task_types=["topic_extraction"],
+            handler=handle_topic_extraction,
+            concurrency_limit=1
+        )
+        .add_worker(
+            name="sentiment_analyzer",
+            task_types=["sentiment_analysis"],
+            handler=handle_sentiment_analysis,
+            concurrency_limit=1
+        )
+        .add_worker(
+            name="formatter",
+            task_types=["formatting"],
+            handler=handle_formatting,
+            concurrency_limit=1
+        )
+        .build(llm_caller=llm_call)
+    )
+    
+    # Example document to process
+    document = {
+        "document": "sample_document.pdf",
+        "options": {
+            "include_summary": True,
+            "include_topics": True,
+            "include_sentiment": True
+        }
+    }
+    
+    try:
+        # Process document
+        print("\nProcessing document...")
+        results = await processor.execute(document)
+        
+        # Print results
+        print("\nProcessing Results:")
+        for task_id, result in results.items():
+            print(f"\n{task_id}:")
+            print(result)
+            
+    except Exception as e:
+        print(f"Error processing document: {str(e)}")
+
+if __name__ == "__main__":
+    asyncio.run(main()) 
@@ -0,0 +1,223 @@
+from typing import List, Dict, Any, Callable, Optional, Union
+from abc import ABC, abstractmethod
+from pydantic import BaseModel
+import asyncio
+import json
+
+class Task(BaseModel):
+    """A task to be executed by a worker"""
+    id: str
+    type: str
+    input_data: Dict[str, Any]
+    dependencies: List[str] = []
+    status: str = "pending"
+    result: Optional[Dict[str, Any]] = None
+
+class Worker(BaseModel):
+    """A worker that can execute specific types of tasks"""
+    name: str
+    task_types: List[str]
+    handler: Callable[[Dict[str, Any]], Any]
+    concurrency_limit: int = 1
+
+class Orchestrator:
+    """Manages task decomposition, delegation, and execution"""
+    
+    def __init__(
+        self,
+        workers: List[Worker],
+        llm_caller: Callable[[str], str]
+    ):
+        self.workers = workers
+        self.llm_caller = llm_caller
+        self.tasks: Dict[str, Task] = {}
+        self.worker_queues: Dict[str, asyncio.Queue] = {}
+        self._validate_config()
+        self._setup_queues()
+    
+    def _validate_config(self):
+        """Validate orchestrator configuration"""
+        # Check for duplicate worker names
+        names = set()
+        for worker in self.workers:
+            if worker.name in names:
+                raise ValueError(f"Duplicate worker name: {worker.name}")
+            names.add(worker.name)
+        
+        # Check that all task types have handlers
+        task_types = set()
+        for worker in self.workers:
+            task_types.update(worker.task_types)
+    
+    def _setup_queues(self):
+        """Set up task queues for each worker"""
+        for worker in self.workers:
+            self.worker_queues[worker.name] = asyncio.Queue()
+    
+    async def _decompose_task(
+        self,
+        input_data: Dict[str, Any]
+    ) -> List[Task]:
+        """Use LLM to decompose input into subtasks"""
+        prompt = (
+            "Decompose this task into smaller subtasks. For each subtask, specify:\n"
+            "1. A unique ID\n"
+            "2. The task type (one of: " + 
+            ", ".join(self._get_all_task_types()) + ")\n"
+            "3. Input data\n"
+            "4. Dependencies (other task IDs)\n\n"
+            "Respond in JSON format:\n"
+            "{\n"
+            '  "tasks": [\n'
+            "    {\n"
+            '      "id": "task-1",\n'
+            '      "type": "task_type",\n'
+            '      "input_data": {},\n'
+            '      "dependencies": []\n'
+            "    }\n"
+            "  ]\n"
+            "}\n\n"
+            f"Task to decompose:\n{json.dumps(input_data, indent=2)}"
+        )
+        
+        try:
+            response = await self.llm_caller(prompt)
+            task_data = json.loads(response)
+            return [Task(**task) for task in task_data["tasks"]]
+        except Exception as e:
+            raise Exception(f"Error decomposing task: {str(e)}")
+    
+    def _get_all_task_types(self) -> List[str]:
+        """Get all supported task types"""
+        task_types = set()
+        for worker in self.workers:
+            task_types.update(worker.task_types)
+        return sorted(list(task_types))
+    
+    def _get_workers_for_task(self, task_type: str) -> List[Worker]:
+        """Get workers that can handle a specific task type"""
+        return [
+            worker for worker in self.workers
+            if task_type in worker.task_types
+        ]
+    
+    async def _execute_worker(self, worker: Worker):
+        """Execute tasks for a specific worker"""
+        queue = self.worker_queues[worker.name]
+        while True:
+            task = await queue.get()
+            try:
+                # Execute task
+                result = await worker.handler(task.input_data)
+                
+                # Update task status and result
+                task.status = "completed"
+                task.result = result
+                
+                # Check if any dependent tasks can now run
+                await self._check_dependencies()
+                
+            except Exception as e:
+                task.status = "failed"
+                task.result = {"error": str(e)}
+            
+            finally:
+                queue.task_done()
+    
+    async def _check_dependencies(self):
+        """Check and queue tasks whose dependencies are met"""
+        for task in self.tasks.values():
+            if task.status == "pending":
+                # Check if all dependencies are completed
+                deps_completed = all(
+                    self.tasks[dep].status == "completed"
+                    for dep in task.dependencies
+                )
+                
+                if deps_completed:
+                    # Find suitable worker and queue task
+                    workers = self._get_workers_for_task(task.type)
+                    if workers:
+                        # Simple round-robin for now
+                        worker = workers[0]
+                        await self.worker_queues[worker.name].put(task)
+                        task.status = "queued"
+    
+    async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
+        """Execute a complex task using orchestration"""
+        try:
+            # Decompose task
+            tasks = await self._decompose_task(input_data)
+            
+            # Store tasks
+            for task in tasks:
+                self.tasks[task.id] = task
+            
+            # Start worker coroutines
+            worker_tasks = []
+            for worker in self.workers:
+                for _ in range(worker.concurrency_limit):
+                    worker_tasks.append(
+                        asyncio.create_task(self._execute_worker(worker))
+                    )
+            
+            # Queue initial tasks (those with no dependencies)
+            for task in tasks:
+                if not task.dependencies:
+                    workers = self._get_workers_for_task(task.type)
+                    if workers:
+                        worker = workers[0]  # Simple round-robin
+                        await self.worker_queues[worker.name].put(task)
+                        task.status = "queued"
+            
+            # Wait for all tasks to complete
+            for queue in self.worker_queues.values():
+                await queue.join()
+            
+            # Cancel worker tasks
+            for task in worker_tasks:
+                task.cancel()
+            
+            # Combine results
+            return {
+                task.id: task.result
+                for task in self.tasks.values()
+                if task.status == "completed"
+            }
+            
+        except Exception as e:
+            raise Exception(f"Error in orchestration: {str(e)}")
+
+class OrchestratorBuilder:
+    """Helper class to build orchestrators"""
+    
+    def __init__(self):
+        self.workers: List[Worker] = []
+    
+    def add_worker(
+        self,
+        name: str,
+        task_types: List[str],
+        handler: Callable[[Dict[str, Any]], Any],
+        concurrency_limit: int = 1
+    ) -> 'OrchestratorBuilder':
+        """Add a worker to the orchestrator"""
+        self.workers.append(
+            Worker(
+                name=name,
+                task_types=task_types,
+                handler=handler,
+                concurrency_limit=concurrency_limit
+            )
+        )
+        return self
+    
+    def build(
+        self,
+        llm_caller: Callable[[str], str]
+    ) -> Orchestrator:
+        """Build the orchestrator"""
+        return Orchestrator(
+            workers=self.workers,
+            llm_caller=llm_caller
+        ) 
@@ -0,0 +1,193 @@
+# Parallelization Pattern
+
+This pattern demonstrates how to execute multiple LLM tasks in parallel and combine their results using various voting strategies.
+
+## Overview
+
+The parallelization pattern is useful when:
+
+- Multiple perspectives or analyses are needed
+- Tasks can be executed independently
+- Results need to be combined systematically
+- Performance optimization is important
+
+## Components
+
+### Core Classes
+
+1. **ParallelTask**
+
+   - Defines a task to run in parallel
+   - Includes prompt template and weight
+   - Supports context variables
+
+2. **Parallelizer**
+
+   - Manages parallel task execution
+   - Implements voting strategies
+   - Handles result combination
+
+3. **ParallelizerBuilder**
+   - Fluent interface for configuration
+   - Supports task and strategy setup
+   - Enables custom result combining
+
+## Example Implementation
+
+The `examples/content_moderation.py` demonstrates parallel content moderation:
+
+1. Parallel Checks:
+
+   - Toxicity detection
+   - Adult content detection
+   - Spam detection
+
+2. Features:
+   - Weighted voting
+   - Custom result combination
+   - Detailed explanations
+   - Fail-safe design
+
+## Usage
+
+1. Set up environment:
+
+```bash
+pip install -r requirements.txt
+```
+
+2. Configure API keys:
+
+```bash
+export ANTHROPIC_API_KEY=your_key_here
+# or
+export OPENAI_API_KEY=your_key_here
+```
+
+3. Run the example:
+
+```bash
+python -m examples.content_moderation
+```
+
+## Implementation Details
+
+### Task Definition
+
+```python
+ParallelTask(
+    name="task_name",
+    prompt_template="Your prompt with {variables}",
+    weight=1.0
+)
+```
+
+### Parallelizer Building
+
+```python
+parallelizer = (
+    ParallelizerBuilder()
+    .add_task(...)
+    .add_task(...)
+    .set_voting_strategy(strategy, custom_combiner)
+    .build(llm_caller)
+)
+```
+
+### Execution
+
+```python
+result = await parallelizer.execute(context)
+```
+
+## Voting Strategies
+
+1. **Majority Voting**
+
+   - Most common response wins
+   - Supports weighted votes
+   - Default strategy
+
+2. **Unanimous Voting**
+
+   - Requires all tasks to agree
+   - Returns all responses if no consensus
+   - Good for critical decisions
+
+3. **Weighted Averaging**
+
+   - For numeric responses
+   - Considers task weights
+   - Falls back to majority if non-numeric
+
+4. **Custom Combining**
+   - User-defined combination logic
+   - Access to all task results
+   - Maximum flexibility
+
+## Best Practices
+
+1. **Task Design**
+
+   - Keep tasks independent
+   - Use clear prompt templates
+   - Assign appropriate weights
+   - Handle edge cases
+
+2. **Performance**
+
+   - Optimize number of tasks
+   - Balance task complexity
+   - Monitor execution times
+   - Handle timeouts
+
+3. **Result Combination**
+
+   - Choose appropriate strategy
+   - Handle conflicting results
+   - Provide clear explanations
+   - Consider confidence levels
+
+4. **Error Handling**
+   - Handle task failures
+   - Implement timeouts
+   - Provide fallbacks
+   - Log issues
+
+## Extensions
+
+Consider extending this pattern with:
+
+1. **Advanced Execution**
+
+   - Dynamic task creation
+   - Conditional execution
+   - Task prioritization
+   - Resource management
+
+2. **Result Processing**
+
+   - Confidence scoring
+   - Result filtering
+   - Anomaly detection
+   - Quality metrics
+
+3. **Performance**
+
+   - Result caching
+   - Task batching
+   - Rate limiting
+   - Load balancing
+
+4. **Monitoring**
+
+   - Task timing
+   - Error rates
+   - Result distribution
+   - Resource usage
+
+5. **Integration**
+   - Multiple LLM providers
+   - External services
+   - Custom task types
+   - Result persistence