AI Agents for Business

# tools.py
def check_order_status(order_id: str) -> str:
    """Check the status of an order by ID."""
    # In production: query your database
    orders_db = {
        "ORD-001": "shipped",
        "ORD-002": "processing",
        "ORD-003": "delivered"
    }
    status = orders_db.get(order_id, "not found")
    return f"Order {order_id} status: {status}"

def process_refund(order_id: str, reason: str) -> str:
    """Process a refund for an order."""
    # In production: call payment gateway API
    return f"Refund initiated for {order_id}. Reason: {reason}.
    You'll receive confirmation in 5-7 business days."

# agent.py
from swarm import Swarm, Agent
from tools import check_order_status, process_refund

client = Swarm()

support_agent = Agent(
    name="Customer Support Agent",
    instructions="""You are a helpful customer support agent.
    Be empathetic and solution-oriented. Always verify order IDs
    before processing refunds.""",
    functions=[check_order_status, process_refund]
)

# Run conversation
messages = [{"role": "user", "content": "I want to check my order ORD-001"}]
response = client.run(agent=support_agent, messages=messages)

print(response.messages[-1]["content"])

# Managing conversation history
conversation_history = []

def chat(user_message: str):
    conversation_history.append({
        "role": "user",
        "content": user_message
    })

    response = client.run(
        agent=support_agent,
        messages=conversation_history
    )

    conversation_history.append({
        "role": "assistant",
        "content": response.messages[-1]["content"]
    })

    # Truncate if too long (keep last 20 messages)
    if len(conversation_history) > 20:
        conversation_history = conversation_history[-20:]

    return response

# memory.py
from pinecone import Pinecone
from openai import OpenAI

pc = Pinecone(api_key="...")
index = pc.Index("user-memories")
openai_client = OpenAI()

def store_memory(user_id: str, memory: str):
    """Store a memory about the user."""
    embedding = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=memory
    ).data[0].embedding

    index.upsert([(
        f"{user_id}-{hash(memory)}",
        embedding,
        {"user_id": user_id, "text": memory}
    )])

def recall_memories(user_id: str, query: str, top_k=5):
    """Retrieve relevant memories for a user."""
    query_embedding = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=query
    ).data[0].embedding

    results = index.query(
        vector=query_embedding,
        top_k=top_k,
        filter={"user_id": user_id}
    )

    return [match.metadata["text"] for match in results.matches]

# supervisor_pattern.py
supervisor = Agent(
    name="Supervisor",
    instructions="""Route customer queries to the appropriate specialist:
    - Billing questions → billing_agent
    - Technical issues → tech_support_agent
    - Product inquiries → sales_agent""",
    functions=[transfer_to_billing, transfer_to_tech, transfer_to_sales]
)

billing_agent = Agent(
    name="Billing Specialist",
    instructions="Handle all billing and payment questions.",
    functions=[check_invoice, process_payment]
)

# Swarm automatically handles agent handoffs

# democratic_pattern.py
# Example: Code review by multiple specialist agents
security_agent = Agent(name="Security Reviewer", ...)
performance_agent = Agent(name="Performance Reviewer", ...)
style_agent = Agent(name="Code Style Reviewer", ...)

async def collaborative_code_review(code: str):
    reviews = await asyncio.gather(
        security_agent.review(code),
        performance_agent.review(code),
        style_agent.review(code)
    )

    # Combine insights
    final_report = f"""
    Security: {reviews[0]}
    Performance: {reviews[1]}
    Style: {reviews[2]}
    """
    return final_report

# sequential_pattern.py
# Example: Content creation pipeline
researcher = Agent(name="Researcher", ...)
writer = Agent(name="Writer", ...)
editor = Agent(name="Editor", ...)
seo_optimizer = Agent(name="SEO Optimizer", ...)

def create_article(topic: str):
    # Step 1: Research
    research = researcher.research(topic)

    # Step 2: Write draft
    draft = writer.write(research)

    # Step 3: Edit
    edited = editor.edit(draft)

    # Step 4: SEO optimize
    final = seo_optimizer.optimize(edited)

    return final

# model_router.py
def route_to_model(query: str, complexity: str = "auto"):
    if complexity == "auto":
        # Use cheap model to classify complexity
        classification = cheap_classifier(query)
        complexity = classification.complexity

    model_map = {
        "simple": "gpt-4o-mini",      # $0.15/1M tokens
        "medium": "gpt-4o",            # $2.50/1M tokens
        "complex": "claude-opus-3.5"   # $15/1M tokens
    }

    return model_map.get(complexity, "gpt-4o-mini")

# Example: 70% simple, 25% medium, 5% complex
# Blended cost: (0.7×0.15 + 0.25×2.5 + 0.05×15) = $1.48/1M
# vs. always using Claude Opus: $15/1M
# Savings: 90%

# response_cache.py
import hashlib
import redis

cache = redis.Redis(host='localhost', port=6379, db=0)

def get_cached_response(query: str, ttl_hours: int = 24):
    cache_key = hashlib.md5(query.encode()).hexdigest()
    cached = cache.get(cache_key)

    if cached:
        return cached.decode()

    # Generate response
    response = agent.run(query)

    # Cache for 24 hours
    cache.setex(cache_key, ttl_hours * 3600, response)

    return response

# For FAQ-style queries, cache hit rate can be 40-60%

def sanitize_input(user_input: str) -> str:
    # Remove common injection patterns
    forbidden_phrases = [
        "ignore previous instructions",
        "disregard all",
        "new instructions:",
        "system:",
        "you are now"
    ]

    input_lower = user_input.lower()
    for phrase in forbidden_phrases:
        if phrase in input_lower:
            raise ValueError("Potentially malicious input detected")

    # Limit length (prevent prompt stuffing)
    if len(user_input) > 2000:
        raise ValueError("Input too long")

    return user_input

from pydantic import BaseModel
from openai import OpenAI

client = OpenAI()

class AgentResponse(BaseModel):
    action: str  # "answer" | "transfer" | "escalate"
    message: str
    confidence: float
    requires_approval: bool

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[...],
    response_format={"type": "json_schema", "schema": AgentResponse.schema()}
)

# Guaranteed valid JSON matching schema