Agent Frameworks

from langchain.agents import create_structured_chat_agent
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List

class CustomerInfo(BaseModel):
    """Structured customer information"""
    name: str = Field(description="Customer full name")
    email: str = Field(description="Customer email address")
    issue_category: str = Field(description="Issue category: billing, technical, or account")
    priority: str = Field(description="Priority: low, medium, high")
    summary: str = Field(description="Brief issue summary")

class StructuredAgent:
    """Agent that returns structured output"""
    
    def __init__(self):
        self.llm = ChatOpenAI(temperature=0)
        self.tools = self._setup_tools()
    
    def _setup_tools(self) -> List[Tool]:
        @tool
        def lookup_customer(email: str) -> str:
            """Look up customer by email"""
            return f"Customer found: {email}"
        
        return [lookup_customer]
    
    def process_inquiry(self, inquiry: str) -> CustomerInfo:
        """Process inquiry and return structured data"""
        
        prompt = f"""
Extract structured information from this customer inquiry:
{inquiry}

Return:
- name: Full customer name
- email: Email address
- issue_category: billing, technical, or account
- priority: low, medium, or high
- summary: Brief description
"""
        # Use structured output parser
        from langchain.output_parsers import PydanticOutputParser
        parser = PydanticOutputParser(pydantic_object=CustomerInfo)
        
        prompt_with_format = prompt + "\n\n" + parser.get_format_instructions()
        
        response = self.llm.predict(prompt_with_format)
        return parser.parse(response)

# Usage
agent = StructuredAgent()
result = agent.process_inquiry(
    "Hi, I'm Sarah Chen (sarah@example.com). My credit card was charged twice for order #123. This is urgent!"
)
print(f"Name: {result.name}")
print(f"Category: {result.issue_category}")
print(f"Priority: {result.priority}")

from langchain.agents import AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from typing import Dict, Any, List
import logging

class CustomAgentExecutor:
    """Custom agent with enhanced control"""
    
    def __init__(self, tools: List[Tool], max_iterations: int = 10):
        self.llm = ChatOpenAI(temperature=0, model="gpt-4")
        self.tools = tools
        self.max_iterations = max_iterations
        self.logger = logging.getLogger("CustomAgent")
        
        # Custom prompt template
        self.prompt = PromptTemplate.from_template("""
You are a helpful assistant with access to tools.

Available tools:
{tool_descriptions}

Use this format:
Thought: Think about what to do
Action: tool_name
Action Input: input for the tool
Observation: result from the tool

Repeat until you can provide a Final Answer.

Question: {input}

{agent_scratchpad}
""")
        
        # Create agent
        self.agent = create_react_agent(self.llm, tools, self.prompt)
        self.executor = AgentExecutor(
            agent=self.agent,
            tools=tools,
            max_iterations=max_iterations,
            verbose=True,
            handle_parsing_errors=True,
            return_intermediate_steps=True
        )
    
    def run(self, query: str) -> Dict[str, Any]:
        """Run agent with enhanced logging and error handling"""
        self.logger.info(f"Starting query: {query}")
        
        try:
            result = self.executor.invoke({"input": query})
            
            # Extract structured results
            return {
                "success": True,
                "output": result["output"],
                "steps": len(result.get("intermediate_steps", [])),
                "intermediate_steps": result.get("intermediate_steps", []),
                "error": None
            }
            
        except Exception as e:
            self.logger.error(f"Agent failed: {str(e)}")
            return {
                "success": False,
                "output": None,
                "steps": 0,
                "intermediate_steps": [],
                "error": str(e)
            }
    
    def get_tool_usage_stats(self, result: Dict[str, Any]) -> Dict[str, int]:
        """Analyze tool usage from agent run"""
        tool_counts = {}
        
        for step in result.get("intermediate_steps", []):
            action = step[0]
            tool_name = action.tool
            tool_counts[tool_name] = tool_counts.get(tool_name, 0) + 1
        
        return tool_counts

# Usage
tools = [search_tool, calculator_tool, database_tool]
executor = CustomAgentExecutor(tools, max_iterations=15)

result = executor.run("What is the weather in Boston and calculate 15% tip on $45.50?")

if result["success"]:
    print(f"Answer: {result['output']}")
    print(f"Steps: {result['steps']}")
    print(f"Tool usage: {executor.get_tool_usage_stats(result)}")
else:
    print(f"Error: {result['error']}")

from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain.agents import AgentExecutor, create_react_agent
import base64
from PIL import Image
import io

class MultiModalAgent:
    """Agent that can process text and images"""
    
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4-vision-preview")
        self.tools = self._setup_tools()
    
    def _setup_tools(self) -> List[Tool]:
        @tool
        def analyze_image(image_path: str) -> str:
            """Analyze image and describe contents"""
            with open(image_path, "rb") as f:
                image_data = base64.b64encode(f.read()).decode()
            
            # Send to GPT-4 Vision
            response = self.llm.predict(
                f"What's in this image? Data: {image_data[:100]}..."
            )
            return response
        
        @tool
        def extract_text_from_image(image_path: str) -> str:
            """Extract text from image (OCR)"""
            # Simplified - use pytesseract in production
            return "Extracted text from image"
        
        @tool
        def compare_images(image1: str, image2: str) -> str:
            """Compare two images"""
            return "Images are similar/different because..."
        
        return [analyze_image, extract_text_from_image, compare_images]
    
    def process(self, text_query: str, image_paths: List[str] = None) -> str:
        """Process text query with optional images"""
        
        # If images provided, add to context
        context = text_query
        if image_paths:
            context += f"\n\nImages provided: {', '.join(image_paths)}"
        
        agent = AgentExecutor.from_agent_and_tools(
            agent=create_react_agent(self.llm, self.tools, prompt),
            tools=self.tools,
            verbose=True
        )
        
        return agent.run(context)

# Usage
agent = MultiModalAgent()
result = agent.process(
    "What's in these product images? Are they similar?",
    image_paths=["product1.jpg", "product2.jpg"]
)
print(result)

"""
Minimal agent framework - ~200 lines
"""
from typing import List, Dict, Any, Callable, Optional
from dataclasses import dataclass
from enum import Enum
import json
import logging

# ============== Core Types ==============

class ActionType(Enum):
    THOUGHT = "thought"
    ACTION = "action"
    OBSERVATION = "observation"
    ANSWER = "answer"

@dataclass
class AgentStep:
    """Single step in agent execution"""
    type: ActionType
    content: str
    tool_used: Optional[str] = None
    tool_input: Optional[Dict] = None
    tool_output: Optional[str] = None

@dataclass
class Tool:
    """Tool definition"""
    name: str
    description: str
    func: Callable
    parameters: Dict[str, Any]

@dataclass
class AgentResult:
    """Final agent result"""
    success: bool
    answer: str
    steps: List[AgentStep]
    iterations: int
    error: Optional[str] = None

# ============== Agent Core ==============

class MinimalAgent:
    """Lightweight agent framework"""
    
    def __init__(self, llm: Callable, tools: List[Tool], max_iterations: int = 10):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.max_iterations = max_iterations
        self.logger = logging.getLogger("MinimalAgent")
    
    def run(self, query: str) -> AgentResult:
        """Execute agent loop"""
        steps = []
        
        prompt = self._build_initial_prompt(query)
        
        for iteration in range(self.max_iterations):
            # Get LLM response
            response = self.llm(prompt)
            
            # Parse response
            step = self._parse_response(response)
            steps.append(step)
            
            # Handle different step types
            if step.type == ActionType.ANSWER:
                return AgentResult(
                    success=True,
                    answer=step.content,
                    steps=steps,
                    iterations=iteration + 1
                )
            
            elif step.type == ActionType.ACTION:
                # Execute tool
                observation = self._execute_tool(step.tool_used, step.tool_input)
                steps.append(AgentStep(
                    type=ActionType.OBSERVATION,
                    content=observation,
                    tool_used=step.tool_used
                ))
                
                # Update prompt with observation
                prompt = self._update_prompt(prompt, step, observation)
        
        # Max iterations reached
        return AgentResult(
            success=False,
            answer="",
            steps=steps,
            iterations=self.max_iterations,
            error="Max iterations reached"
        )
    
    def _build_initial_prompt(self, query: str) -> str:
        """Build initial prompt with tool descriptions"""
        tool_descriptions = "\n".join([
            f"- {name}: {tool.description}"
            for name, tool in self.tools.items()
        ])
        
        return f"""
You are a helpful assistant with access to tools.

Available tools:
{tool_descriptions}

Use this format:
Thought: [your reasoning]
Action: [tool name]
Action Input: [tool input as JSON]
Observation: [tool result will be provided]

When you have enough information:
Answer: [final answer]

Question: {query}
"""
    
    def _parse_response(self, response: str) -> AgentStep:
        """Parse LLM response into AgentStep"""
        lines = response.strip().split("\n")
        
        for line in lines:
            if line.startswith("Thought:"):
                return AgentStep(
                    type=ActionType.THOUGHT,
                    content=line.replace("Thought:", "").strip()
                )
            elif line.startswith("Action:"):
                tool_name = line.replace("Action:", "").strip()
                # Look for Action Input on next lines
                tool_input = {}
                for next_line in lines[lines.index(line)+1:]:
                    if next_line.startswith("Action Input:"):
                        input_str = next_line.replace("Action Input:", "").strip()
                        try:
                            tool_input = json.loads(input_str)
                        except:
                            tool_input = {"input": input_str}
                        break
                
                return AgentStep(
                    type=ActionType.ACTION,
                    content=f"Using {tool_name}",
                    tool_used=tool_name,
                    tool_input=tool_input
                )
            elif line.startswith("Answer:"):
                return AgentStep(
                    type=ActionType.ANSWER,
                    content=line.replace("Answer:", "").strip()
                )
        
        # Default thought
        return AgentStep(type=ActionType.THOUGHT, content=response)
    
    def _execute_tool(self, tool_name: str, tool_input: Dict[str, Any]) -> str:
        """Execute tool and return observation"""
        if tool_name not in self.tools:
            return f"Error: Tool '{tool_name}' not found"
        
        tool = self.tools[tool_name]
        
        try:
            result = tool.func(**tool_input)
            return str(result)
        except Exception as e:
            return f"Error executing {tool_name}: {str(e)}"
    
    def _update_prompt(self, prompt: str, action: AgentStep, observation: str) -> str:
        """Add observation to prompt"""
        return prompt + f"\nAction: {action.tool_used}\nAction Input: {json.dumps(action.tool_input)}\nObservation: {observation}\n"

# ============== Example Usage ==============

def mock_llm(prompt: str) -> str:
    """Mock LLM for testing"""
    return "Thought: I should search\nAction: search\nAction Input: {\"query\": \"test\"}"

# Define tools
search_tool = Tool(
    name="search",
    description="Search the web",
    func=lambda query: f"Results for: {query}",
    parameters={"query": {"type": "string"}}
)

calculator_tool = Tool(
    name="calculator",
    description="Do math",
    func=lambda expression: str(eval(expression)),
    parameters={"expression": {"type": "string"}}
)

# Create and run agent
agent = MinimalAgent(
    llm=mock_llm,
    tools=[search_tool, calculator_tool],
    max_iterations=10
)

result = agent.run("What is 2+2?")
print(f"Success: {result.success}")
print(f"Answer: {result.answer}")
print(f"Steps: {result.iterations}")

using Microsoft.SemanticKernel;
using Microsoft.SemanticKernel.Connectors.OpenAI;

// Create kernel
var builder = Kernel.CreateBuilder();
builder.AddOpenAIChatCompletion("gpt-4", "your-api-key");
var kernel = builder.Build();

// Add plugins (skills)
kernel.ImportPluginFromType();
kernel.ImportPluginFromType();

// Run agent
var result = await kernel.InvokePromptAsync(
    "What is 15% of 200? Also search for AI news.",
    new OpenAIPromptExecutionSettings { ToolCallBehavior = ToolCallBehavior.AutoInvokeKernelFunctions }
);

Console.WriteLine(result);

from guidance import models, gen, select

# Load model
gpt = models.OpenAI("gpt-4")

# Structured prompt with guaranteed format
with system():
    lm = gpt + "You are a helpful assistant."

with user():
    lm += "Extract customer info from: 'Hi I'm John (john@test.com)'"

with assistant():
    lm += f"""
Name: {gen('name', stop=',')}
Email: {gen('email', regex=r'[a-z0-9@\\.]+')}
Priority: {select(['low', 'medium', 'high'], name='priority')}
"""

print(lm['name'], lm['email'], lm['priority'])

import dspy

# Configure LM
lm = dspy.OpenAI(model='gpt-4')
dspy.settings.configure(lm=lm)

# Define agent module
class ReasoningAgent(dspy.Module):
    def __init__(self):
        self.cot = dspy.ChainOfThought("question -> answer")
    
    def forward(self, question):
        return self.cot(question=question)

# Optimize with examples
agent = ReasoningAgent()

# Use teleprompter to optimize
from dspy.teleprompt import BootstrapFewShot

teleprompter = BootstrapFewShot()
optimized_agent = teleprompter.compile(
    agent,
    trainset=training_examples
)

# Optimized agent now has better prompts
result = optimized_agent(question="What is the capital of France?")
print(result.answer)

from langchain.tools import Tool
from langchain.agents import initialize_agent
from crewai import Agent, Task, Crew

# Use LangChain for individual agents' tools
lang chain_tools = [search_tool, calculator_tool, db_tool]

# Create CrewAI agents with LangChain tools
researcher = Agent(
    role="Researcher",
    goal="Gather comprehensive information",
    tools=langchain_tools,  # LangChain tools work in CrewAI
    llm=ChatOpenAI(model="gpt-4")
)

analyst = Agent(
    role="Analyst",
    goal="Analyze gathered data",
    tools=[analysis_tool],
    llm=ChatOpenAI(model="gpt-4")
)

# Orchestrate with CrewAI
crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task]
)

result = crew.kickoff()
print(result)

from typing import Any, Dict
import time
import logging
from contextlib import contextmanager

class ObservableAgentFramework:
    """Wrap any framework with observability"""
    
    def __init__(self, agent, framework_name: str):
        self.agent = agent
        self.framework_name = framework_name
        self.logger = logging.getLogger(f"Agent.{framework_name}")
        self.metrics = {
            "total_runs": 0,
            "successful_runs": 0,
            "failed_runs": 0,
            "total_duration": 0,
            "tool_calls": 0
        }
    
    @contextmanager
    def _track_run(self):
        """Track agent run metrics"""
        start = time.time()
        self.metrics["total_runs"] += 1
        
        try:
            yield
            self.metrics["successful_runs"] += 1
        except Exception as e:
            self.metrics["failed_runs"] += 1
            raise
        finally:
            duration = time.time() - start
            self.metrics["total_duration"] += duration
            self.logger.info(f"Run completed in {duration:.2f}s")
    
    def run(self, *args, **kwargs) -> Any:
        """Run agent with tracking"""
        with self._track_run():
            self.logger.info(f"Starting {self.framework_name} agent")
            result = self.agent.run(*args, **kwargs)
            return result
    
    def get_metrics(self) -> Dict[str, Any]:
        """Get performance metrics"""
        avg_duration = (
            self.metrics["total_duration"] / self.metrics["total_runs"]
            if self.metrics["total_runs"] > 0 else 0
        )
        
        success_rate = (
            self.metrics["successful_runs"] / self.metrics["total_runs"]
            if self.metrics["total_runs"] > 0 else 0
        )
        
        return {
            **self.metrics,
            "avg_duration": avg_duration,
            "success_rate": success_rate * 100
        }

# Usage with any framework
from langchain.agents import AgentExecutor

langchain_agent = AgentExecutor(...)
wrapped_agent = ObservableAgentFramework(langchain_agent, "LangChain")

result = wrapped_agent.run("Your query")
print(wrapped_agent.get_metrics())

import random
from typing import List, Dict, Any

class FrameworkABTest:
    """Test multiple frameworks on same tasks"""
    
    def __init__(self, frameworks: Dict[str, Any]):
        self.frameworks = frameworks  # {"name": agent}
        self.results = {name: [] for name in frameworks}
    
    def run_test(self, test_queries: List[str], runs_per_query: int = 3):
        """Run all frameworks on test queries"""
        for query in test_queries:
            print(f"\nTesting query: {query}")
            
            for name, agent in self.frameworks.items():
                # Run multiple times for reliability
                query_results = []
                
                for run in range(runs_per_query):
                    start = time.time()
                    try:
                        result = agent.run(query)
                        duration = time.time() - start
                        query_results.append({
                            "success": True,
                            "duration": duration,
                            "result": result
                        })
                    except Exception as e:
                        duration = time.time() - start
                        query_results.append({
                            "success": False,
                            "duration": duration,
                            "error": str(e)
                        })
                
                self.results[name].extend(query_results)
    
    def get_comparison(self) -> Dict[str, Any]:
        """Compare framework performance"""
        comparison = {}
        
        for name, results in self.results.items():
            successes = [r for r in results if r["success"]]
            success_rate = len(successes) / len(results) if results else 0
            avg_duration = sum(r["duration"] for r in results) / len(results) if results else 0
            
            comparison[name] = {
                "runs": len(results),
                "success_rate": success_rate * 100,
                "avg_duration": avg_duration,
                "fastest_run": min(r["duration"] for r in results) if results else 0,
                "slowest_run": max(r["duration"] for r in results) if results else 0
            }
        
        return comparison
    
    def print_report(self):
        """Print comparison report"""
        comparison = self.get_comparison()
        
        print("\n" + "="*70)
        print("FRAMEWORK COMPARISON REPORT")
        print("="*70)
        
        for name, stats in comparison.items():
            print(f"\n{name}:")
            print(f"  Success Rate: {stats['success_rate']:.1f}%")
            print(f"  Avg Duration: {stats['avg_duration']:.2f}s")
            print(f"  Fastest: {stats['fastest_run']:.2f}s")
            print(f"  Slowest: {stats['slowest_run']:.2f}s")

# Usage
test = FrameworkABTest({
    "LangChain": langchain_agent,
    "CrewAI": crewai_agent,
    "Custom": custom_agent
})

test_queries = [
    "What is the weather in Boston?",
    "Calculate 15% of $200",
    "Search for AI news and summarize"
]

test.run_test(test_queries, runs_per_query=5)
test.print_report()

from typing import Dict, Any, List
from enum import Enum

class FrameworkType(Enum):
    LANGCHAIN = "langchain"
    CREWAI = "crewai"
    CUSTOM = "custom"

class SmartFrameworkRouter:
    """Route queries to optimal framework"""
    
    def __init__(self):
        self.frameworks = self._initialize_frameworks()
        self.routing_history = []
    
    def _initialize_frameworks(self) -> Dict[FrameworkType, Any]:
        """Initialize all available frameworks"""
        return {
            FrameworkType.LANGCHAIN: self._create_langchain_agent(),
            FrameworkType.CREWAI: self._create_crewai_crew(),
            FrameworkType.CUSTOM: self._create_custom_agent()
        }
    
    def route_and_execute(self, query: str) -> Dict[str, Any]:
        """Analyze query and route to best framework"""
        # Analyze query
        framework = self._select_framework(query)
        
        # Execute
        start = time.time()
        try:
            result = self.frameworks[framework].run(query)
            duration = time.time() - start
            success = True
        except Exception as e:
            result = None
            duration = time.time() - start
            success = False
        
        # Record for learning
        self.routing_history.append({
            "query": query,
            "framework": framework,
            "success": success,
            "duration": duration
        })
        
        return {
            "result": result,
            "framework_used": framework.value,
            "duration": duration,
            "success": success
        }
    
    def _select_framework(self, query: str) -> FrameworkType:
        """Select best framework for query"""
        # Implement routing logic
        # - Single agent task → LangChain
        # - Multi-agent collaboration → CrewAI
        # - Simple task → Custom
        
        if "team" in query.lower() or "collaborate" in query.lower():
            return FrameworkType.CREWAI
        elif "simple" in query.lower():
            return FrameworkType.CUSTOM
        else:
            return FrameworkType.LANGCHAIN
    
    def learn_from_history(self):
        """Improve routing based on past performance"""
        # Analyze which frameworks work best for which queries
        pass

# Usage
router = SmartFrameworkRouter()
result = router.route_and_execute("Build a collaborative research report")
print(f"Used: {result['framework_used']}")
print(f"Result: {result['result']}")