Building Intelligent AI Agents for Workflow Automation

class IntelligentAgent:
    def __init__(self, llm, tools, memory):
        self.llm = llm              # Large Language Model
        self.tools = tools          # Available actions
        self.memory = memory        # Conversation & task history
        self.planner = Planner()    # Task decomposition
        self.executor = Executor()  # Action execution
        
    def reason(self, observation):
        """Decide what to do next based on current state"""
        context = self.memory.retrieve_relevant(observation)
        plan = self.planner.create_plan(observation, context)
        return plan
        
    def act(self, action):
        """Execute the chosen action"""
        result = self.executor.run(action, self.tools)
        self.memory.store(action, result)
        return result
        
    def perceive(self, environment):
        """Gather information from the environment"""
        return environment.get_state()

from langchain import OpenAI, LLMChain
from langchain.agents import initialize_agent, Tool, AgentType
from langchain.memory import ConversationBufferMemory
from langchain.callbacks import get_openai_callback

# Initialize the language model
llm = OpenAI(
    temperature=0.7,
    model_name="gpt-4",
    max_tokens=2000
)

# Define custom tools for the agent
def analyze_code(code_snippet):
    """Analyzes code quality and suggests improvements"""
    # Implementation here
    return {
        "quality_score": 8.5,
        "issues": ["Missing error handling", "Could use type hints"],
        "suggestions": ["Add try-catch blocks", "Implement logging"]
    }

def generate_tests(function_code):
    """Generates unit tests for given functions"""
    # Implementation here
    return """
    def test_example():
        assert example_function() == expected_result
    """

tools = [
    Tool(
        name="Code Analyzer",
        func=analyze_code,
        description="Analyzes code quality and suggests improvements. Input should be a code snippet."
    ),
    Tool(
        name="Test Generator",
        func=generate_tests,
        description="Generates unit tests for given functions. Input should be the function code."
    )
]

# Add memory for context retention
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# Create the agent with reasoning capabilities
agent = initialize_agent(
    tools, 
    llm, 
    agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
    verbose=True,
    memory=memory,
    max_iterations=5
)

# Track costs with callback
with get_openai_callback() as cb:
    result = agent.run("Analyze this Python function and generate tests for it")
    print(f"Total Cost: ${cb.total_cost:.4f}")

# Agent's internal reasoning loop
def react_loop(task):
    """
    Thought: I need to analyze the code first
    Action: Code Analyzer
    Observation: Code has quality score 8.5 but lacks error handling
    
    Thought: Based on the analysis, I should generate tests
    Action: Test Generator
    Observation: Tests generated successfully
    
    Thought: I have completed the task
    Final Answer: [Provides analysis and tests]
    """
    pass

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma, Pinecone, Weaviate
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
import pinecone

# Initialize embeddings model
embeddings = OpenAIEmbeddings(
    model="text-embedding-ada-002"
)

# Document chunking strategy
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,           # Optimal for most use cases
    chunk_overlap=200,         # Maintain context across chunks
    separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""],
    length_function=len
)

# Split documents into chunks
documents = load_documents("./knowledge_base/")
chunks = text_splitter.split_documents(documents)

# Vector Database Comparison
# Option 1: Chroma (Local, fast prototyping)
vectorstore_chroma = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# Option 2: Pinecone (Production, scalable)
pinecone.init(
    api_key="your-api-key",
    environment="us-west1-gcp"
)
vectorstore_pinecone = Pinecone.from_documents(
    documents=chunks,
    embedding=embeddings,
    index_name="knowledge-base"
)

# Option 3: Weaviate (Advanced features, hybrid search)
vectorstore_weaviate = Weaviate.from_documents(
    documents=chunks,
    embedding=embeddings,
    weaviate_url="http://localhost:8080",
    index_name="KnowledgeBase",
    text_key="content"
)

# Create RAG chain with optimized retrieval
retriever = vectorstore_pinecone.as_retriever(
    search_type="mmr",         # Maximum Marginal Relevance
    search_kwargs={
        "k": 5,                # Retrieve top 5 chunks
        "fetch_k": 20,         # Fetch 20 candidates first
        "lambda_mult": 0.7     # Diversity factor
    }
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",        # Or "map_reduce" for long contexts
    retriever=retriever,
    return_source_documents=True,
    verbose=True
)

# Query with source attribution
result = qa_chain({
    "query": "How do I implement authentication in microservices?"
})

print(f"Answer: {result['result']}")
print(f"\nSources:")
for doc in result['source_documents']:
    print(f"- {doc.metadata['source']}")

from langchain.retrievers import BM25Retriever, EnsembleRetriever

# Keyword-based retriever
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# Semantic retriever
semantic_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# Combine both with weighted scoring
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, semantic_retriever],
    weights=[0.3, 0.7]  # Favor semantic search
)

class CodeReviewAgent:
    def __init__(self):
        self.llm = OpenAI(model="gpt-4", temperature=0.2)
        self.tools = [
            self.static_analysis,
            self.security_scan,
            self.performance_check,
            self.style_validator
        ]
    
    def review_pr(self, pr_diff):
        findings = {
            "bugs": [],
            "security": [],
            "performance": [],
            "style": []
        }
        
        # Multi-stage analysis
        for tool in self.tools:
            results = tool(pr_diff)
            findings[tool.category].extend(results)
        
        # Generate comprehensive review
        review = self.llm.generate_review(findings)
        
        # Assign severity and auto-approve if safe
        if self.calculate_risk_score(findings) < 3:
            return {"status": "auto-approved", "review": review}
        
        return {"status": "needs-review", "review": review}

class DocumentationAgent:
    def on_code_change(self, changed_files):
        for file in changed_files:
            # Extract function signatures and docstrings
            functions = self.parse_code(file)
            
            # Update API documentation
            self.update_api_docs(functions)
            
            # Generate examples
            examples = self.generate_usage_examples(functions)
            
            # Create changelog entries
            self.update_changelog(file, functions)
            
            # Update README if public API changed
            if file.is_public_api:
                self.update_readme(functions)

class DevOpsAgent:
    def monitor_and_respond(self):
        while True:
            metrics = self.collect_metrics()
            
            # Anomaly detection
            if self.detect_anomaly(metrics):
                # Predict failure type
                failure_type = self.predict_failure(metrics)
                
                # Determine remediation
                action = self.plan_remediation(failure_type)
                
                # Execute or escalate
                if action.confidence > 0.85:
                    self.execute_fix(action)
                    self.notify_team("Auto-resolved", action)
                else:
                    self.create_incident(failure_type, action)

class PersonalizationAgent:
    def recommend(self, user_context):
        # Analyze user behavior
        preferences = self.analyze_history(user_context['user_id'])
        
        # Real-time context
        session_intent = self.detect_intent(user_context['session'])
        
        # Retrieve relevant products via RAG
        products = self.search_catalog(preferences, session_intent)
        
        # Generate personalized descriptions
        for product in products:
            product['headline'] = self.generate_headline(
                product, user_context
            )
        
        return products

# Phase 1: Simple automation
agent = SimpleAgent(task="Format code on commit")

# Phase 2: Add decision-making
agent.add_capability("Decide if tests are needed")

# Phase 3: Full autonomy
agent.enable_autonomous_mode()

class AgentMetrics:
    def track_execution(self, agent_run):
        metrics = {
            "accuracy": self.measure_accuracy(agent_run),
            "latency": agent_run.execution_time,
            "cost": agent_run.api_costs,
            "user_satisfaction": self.get_feedback(),
            "tokens_used": agent_run.total_tokens
        }
        
        self.log_to_dashboard(metrics)
        
        # Alert on degradation
        if metrics['accuracy'] < self.threshold:
            self.alert_team("Agent accuracy below threshold")

class SafetyWrapper:
    def __init__(self, agent, critical_actions):
        self.agent = agent
        self.critical_actions = critical_actions
        
    def execute(self, task):
        action = self.agent.plan(task)
        
        # Require approval for critical actions
        if action.type in self.critical_actions:
            approval = self.request_human_approval(action)
            if not approval:
                return "Action rejected by human"
        
        return self.agent.execute(action)

# Token usage optimization
def optimize_prompt(original_prompt):
    # Remove redundant context
    optimized = remove_redundancy(original_prompt)
    
    # Use smaller model for simple tasks
    if task_complexity(optimized) < 5:
        return {"prompt": optimized, "model": "gpt-3.5-turbo"}
    
    return {"prompt": optimized, "model": "gpt-4"}

# Cost tracking
monthly_budget = 5000  # USD
if current_spend > monthly_budget * 0.8:
    switch_to_rate_limited_mode()

// n8n workflow node example
{
  "nodes": [
    {
      "name": "Trigger",
      "type": "n8n-nodes-base.webhook",
      "parameters": {
        "path": "new-support-ticket"
      }
    },
    {
      "name": "OpenAI Agent",
      "type": "n8n-nodes-base.openAi",
      "parameters": {
        "operation": "complete",
        "model": "gpt-4",
        "prompt": "={{$json.ticketContent}}",
        "maxTokens": 500
      }
    },
    {
      "name": "Decision",
      "type": "n8n-nodes-base.if",
      "parameters": {
        "conditions": {
          "string": [
            {
              "value1": "={{$json.sentiment}}",
              "operation": "equals",
              "value2": "urgent"
            }
          ]
        }
      }
    },
    {
      "name": "Auto-Respond",
      "type": "n8n-nodes-base.sendEmail",
      "parameters": {
        "toEmail": "={{$json.customerEmail}}",
        "subject": "Re: Your Support Ticket",
        "text": "={{$json.aiResponse}}"
      }
    }
  ]
}

# Solution: Implement sliding window
class ContextManager:
    def __init__(self, max_tokens=8000):
        self.max_tokens = max_tokens
        self.history = []
    
    def add_message(self, message):
        self.history.append(message)
        
        # Trim if exceeding limit
        while self.count_tokens(self.history) > self.max_tokens:
            # Keep system message and recent history
            self.history = [self.history[0]] + self.history[2:]

# Solution: Semantic compression
def compress_context(conversation_history):
    # Summarize older messages
    old_messages = conversation_history[:-10]
    summary = llm.summarize(old_messages)
    
    # Keep recent messages verbatim
    recent = conversation_history[-10:]
    
    return [summary] + recent

import time
from functools import wraps

def retry_with_backoff(max_retries=5):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except RateLimitError:
                    wait_time = (2 ** attempt) + random.random()
                    time.sleep(wait_time)
            raise Exception("Max retries exceeded")
        return wrapper
    return decorator

@retry_with_backoff()
def call_llm_api(prompt):
    return llm.complete(prompt)

# Solution: Fact-checking layer
class FactChecker:
    def verify_response(self, response, sources):
        # Extract claims
        claims = self.extract_claims(response)
        
        # Verify against sources
        verified = []
        for claim in claims:
            if self.can_verify(claim, sources):
                verified.append(claim)
            else:
                # Remove or flag unverified claims
                response = response.replace(claim, "[UNVERIFIED]")
        
        return response

import pytest
from unittest.mock import Mock

def test_agent_tool_selection():
    """Test that agent selects correct tool"""
    agent = create_test_agent()
    
    result = agent.run("Analyze this code for bugs")
    
    assert "Code Analyzer" in result.tools_used
    assert result.confidence > 0.7

def test_agent_error_handling():
    """Test agent handles tool failures gracefully"""
    agent = create_test_agent()
    agent.tools['analyzer'].fail()
    
    result = agent.run("Analyze code")
    
    assert not result.crashed
    assert "fallback" in result.strategy_used

class AgentEvaluator:
    def evaluate_accuracy(self, test_cases):
        results = []
        for case in test_cases:
            agent_answer = self.agent.run(case.question)
            score = self.compare_answers(
                agent_answer, 
                case.expected_answer
            )
            results.append(score)
        
        return {
            "average_accuracy": np.mean(results),
            "p95_accuracy": np.percentile(results, 95),
            "failures": [r for r in results if r < 0.5]
        }

from prometheus_client import Counter, Histogram

# Define metrics
agent_requests = Counter('agent_requests_total', 'Total requests')
agent_latency = Histogram('agent_latency_seconds', 'Request latency')
agent_errors = Counter('agent_errors_total', 'Total errors')

@agent_latency.time()
def run_agent(task):
    agent_requests.inc()
    try:
        return agent.execute(task)
    except Exception as e:
        agent_errors.inc()
        raise

class AgentTeam:
    def __init__(self):
        self.researcher = ResearchAgent()
        self.coder = CodingAgent()
        self.reviewer = ReviewAgent()
        self.deployer = DeploymentAgent()
    
    def build_feature(self, requirements):
        # Research phase
        context = self.researcher.gather_info(requirements)
        
        # Implementation phase
        code = self.coder.write_code(context)
        
        # Review phase
        review = self.reviewer.check_quality(code)
        if not review.approved:
            code = self.coder.fix_issues(review.issues)
        
        # Deployment phase
        self.deployer.deploy(code)

class SelfImprovingAgent:
    def optimize_self(self):
        # Analyze performance logs
        bottlenecks = self.analyze_performance()
        
        # Generate improved version
        optimized_code = self.llm.improve_code(
            self.source_code, 
            bottlenecks
        )
        
        # Test in sandbox
        if self.test_improvements(optimized_code):
            self.update_self(optimized_code)

class PredictiveAgent:
    def anticipate_needs(self, user_behavior):
        # Predict next action
        next_action = self.ml_model.predict(user_behavior)
        
        # Prepare resources proactively
        if next_action.probability > 0.8:
            self.pre_cache_data(next_action)
            self.pre_warm_services(next_action)