소매 및 전자상거래 분야의 AI 에이전트

import json
from datetime import datetime, timedelta
from litellm import completion
import os
import weave; weave.init('retail-email-agent')


os.environ["OPENAI_API_KEY"] = "your api key"


# Mock email database
MOCK_EMAILS = [
    {
        "id": "1",
        "from": "customer@example.com",
        "subject": "App keeps crashing",
        "body": "The mobile app has been crashing constantly for the past week. This is frustrating as I rely on it for work.",
        "timestamp": (datetime.now() - timedelta(hours=2)).isoformat(),
        "urgency": None,
        "category": None
    },
    {
        "id": "2",
        "from": "support@partner.com",
        "subject": "Urgent: Service integration issue",
        "body": "We're experiencing problems with the API integration. Several customers are affected. Need immediate assistance.",
        "timestamp": (datetime.now() - timedelta(hours=1)).isoformat(),
        "urgency": None,
        "category": None
    },
    {
        "id": "3",
        "from": "user123@example.com",
        "subject": "Billing overcharge",
        "body": "I was charged twice for my monthly subscription. Please refund the extra charge. Your billing system needs better validation.",
        "timestamp": (datetime.now() - timedelta(minutes=30)).isoformat(),
        "urgency": None,
        "category": None
    },
    {
        "id": "4",
        "from": "enterprise@bigcorp.com",
        "subject": "Missing enterprise features",
        "body": "We need better team management features and role-based access control. Currently managing large teams is very manual.",
        "timestamp": (datetime.now() - timedelta(minutes=45)).isoformat(),
        "urgency": None,
        "category": None
    }
]

# Urgency levels mapped to descriptors
URGENCY_LEVELS = {
    2: "HIGH",    # Critical issues, system down, multiple users affected
    1: "MEDIUM",  # Important but not critical
    0: "LOW"      # Regular requests, no immediate impact
}

# Categories mapped to integers
CATEGORIES = {
    0: "Technical Issues",      # App crashes, bugs
    1: "Performance",          # Speed, reliability
    2: "Billing/Pricing",      # Payment problems
    3: "User Experience",      # UI/UX issues
    4: "Feature Requests",     # New features
    5: "Security/Privacy",     # Security concerns
    6: "Integration",          # API issues
    7: "Customer Service",     # Support general
    8: "Documentation",        # Help docs
    9: "Enterprise"           # Large customer needs
}

class EmailSystem:
    def __init__(self, model_id="openai/gpt-4o"):
        self.emails = MOCK_EMAILS.copy()
        self.model_id = model_id
    
    def _run_inference(self, prompt):
        """Run inference using the specified model"""
        try:
            response = completion(
                model=self.model_id,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.1
            )
            return response["choices"][0]["message"]["content"].strip()
        except Exception as e:
            print(f"Inference error: {str(e)}")
            return "7"  # Default to Customer Service
    
    def get_category(self, email):
        """Get category as integer 0-9"""
        categories = "\n".join(f"{k}: {v}" for k, v in CATEGORIES.items())
        
        prompt = f"""Categorize this email into exactly ONE category by responding with ONLY its number (0-9):

Categories:
{categories}

Email subject: {email['subject']}
Email body: {email['body']}

Respond with ONLY a single number 0-9:"""
        
        try:
            response = self._run_inference(prompt)
            category = int(response)
            if category not in CATEGORIES:
                return 7  # Default to Customer Service
            return category
        except:
            return 7
    @weave.op
    def get_urgency(self, email):
        """Get urgency as integer 0-2"""
        prompt = f"""Rate this email's urgency with ONLY ONE number:
2: HIGH - Emergency, system down, multiple users affected
1: MEDIUM - Important issue but not critical
0: LOW - Regular request, no immediate impact

Email subject: {email['subject']}
Email body: {email['body']}

Respond with ONLY the number (2, 1, or 0):"""
        
        try:
            response = self._run_inference(prompt)
            urgency = int(response)
            if urgency not in [0, 1, 2]:
                return 0
            return urgency
        except:
            return 0

    def process_emails(self):
        """Process all emails and return summary"""
        urgency_map = {2: "HIGH", 1: "MEDIUM", 0: "LOW"}
        results = []
        
        for email in self.emails:
            # Get category
            category_num = self.get_category(email)
            category_name = CATEGORIES[category_num]
            
            # Get urgency
            urgency_num = self.get_urgency(email)
            urgency_name = urgency_map[urgency_num]
            
            results.append({
                "subject": email["subject"],
                "category_num": category_num,
                "category": category_name,
                "urgency_num": urgency_num,
                "urgency": urgency_name
            })
            
        return results

class UrgencyRater:
    def __init__(self, model_id="openai/gpt-4o"):
        self.model_id = model_id
    
    def _run_inference(self, prompt):
        """Run inference using the specified model"""
        try:
            response = completion(
                model=self.model_id,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.1
            )
            return response["choices"][0]["message"]["content"].strip()
        except Exception as e:
            print(f"Inference error: {str(e)}")
            return "0"  # Default to LOW urgency
    
    @weave.op
    def rate_urgency(self, email):
        """Rate email urgency from 0-2"""
        prompt = f"""Analyze this email's urgency and respond with ONLY ONE number:

2: HIGH URGENCY - Critical issues:
- System down/major outage
- Multiple users/customers affected
- Security incidents
- Significant revenue impact
- Words like "urgent", "emergency", "immediate"

1: MEDIUM URGENCY - Important issues:
- Single user blocked
- Bug affecting functionality
- Billing problems
- Integration issues
- Performance problems

0: LOW URGENCY - Regular requests:
- Feature requests
- Documentation
- General questions
- Non-blocking issues
- Future planning

Email subject: {email['subject']}
Email body: {email['body']}

Respond with ONLY the number (2, 1, or 0):"""
        
        try:
            response = self._run_inference(prompt)
            urgency = int(response)
            if urgency not in URGENCY_LEVELS:
                return 0
            return urgency
        except:
            return 0
    
    def batch_rate_emails(self, emails):
        """Rate urgency for a batch of emails"""
        results = []
        for email in emails:
            urgency_num = self.rate_urgency(email)
            results.append({
                "subject": email["subject"],
                "urgency_num": urgency_num,
                "urgency": URGENCY_LEVELS[urgency_num],
                "timestamp": email["timestamp"]
            })
        
        # Sort by urgency (high to low) and then by timestamp
        return sorted(results, 
                     key=lambda x: (-x["urgency_num"], x["timestamp"]))

if __name__ == "__main__":
    # Test both systems
    print("=== Category System Test ===")
    category_system = EmailSystem()
    category_results = category_system.process_emails()
    for r in category_results:
        print(f"\nSubject: {r['subject']}")
        print(f"Category: {r['category_num']} ({r['category']})")
        print(f"Urgency: {r['urgency_num']} ({r['urgency']})")
    
    print("\n=== Urgency Rater Test ===")
    urgency_system = UrgencyRater()
    urgency_results = urgency_system.batch_rate_emails(MOCK_EMAILS)
    for r in urgency_results:
        print(f"\nSubject: {r['subject']}")
        print(f"Urgency: {r['urgency_num']} ({r['urgency']})")
        print(f"Timestamp: {r['timestamp']}")

import os
import pandas as pd
import pickle
from datasets import load_dataset
from langchain_openai import OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.documents import Document

# Set up OpenAI API Key (Replace with your actual API key)
os.environ["OPENAI_API_KEY"] = "your api key"

# Load dataset from Hugging Face
dataset = load_dataset('DBQ/Burberry.Product.prices.United.States')
df = pd.DataFrame(dataset['train'])

# Drop missing values
df = df.dropna(subset=["title", "imageurl", "category2_code", "category3_code", "product_code"])

# Convert product descriptions into embeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
documents = [
    Document(
        page_content=f"{row['title']} - Category: {row['category2_code']} > {row['category3_code']}",
        metadata={
            "title": row["title"],
            "imageurl": row["imageurl"],
            "price": row["price"],
            "category": row["category2_code"],
            "product_code": str(row["product_code"])
        }
    ) for _, row in df.iterrows()
]

# Store only document data (not the full vector store)
with open("documents.pkl", "wb") as f:
    pickle.dump(documents, f)

# Save metadata separately
with open("metadata.pkl", "wb") as f:
    pickle.dump(df, f)

print("Documents and metadata saved successfully! You can now run the Flask app.")


from flask import Flask, request, render_template_string
import pickle
import os
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
import random
from langchain.chat_models import ChatOpenAI
import weave; weave.init("retail-agent")

# Initialize LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# Set up OpenAI API Key
os.environ["OPENAI_API_KEY"] = "your api key"


app = Flask(__name__)

# Load stored documents (precomputed embeddings)
with open("documents.pkl", "rb") as f:
    documents = pickle.load(f)

# Recreate the vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = InMemoryVectorStore(embeddings)
_ = vector_store.add_documents(documents)

# Load metadata (original product details)
with open("metadata.pkl", "rb") as f:
    df = pickle.load(f)

# Track viewed products
viewed_products = []

# Home page template
home_template = """
<!DOCTYPE html>
<html>
<head>
    <title>Product Categories</title>
    <style> body { font-family: Arial, sans-serif; } </style>
</head>
<body>
    <h1>Product Categories</h1>
    <ul>
        {% for category in categories %}
            <li><a href="{{ url_for('subcategory_page', category=category) }}">{{ category }}</a></li>
        {% endfor %}
    </ul>
</body>
</html>
"""

# Subcategory page template
subcategory_template = """
<!DOCTYPE html>
<html>
<head>
    <title>{{ category }} Subcategories</title>
    <style> body { font-family: Arial, sans-serif; } </style>
</head>
<body>
    <h1>{{ category }} Subcategories</h1>
    <ul>
        {% for subcategory in subcategories %}
            <li><a href="{{ url_for('product_page', category=category, subcategory=subcategory) }}">{{ subcategory }}</a></li>
        {% endfor %}
    </ul>
</body>
</html>
"""

# Product listing page template
product_template = """
<!DOCTYPE html>
<html>
<head>
    <title>{{ subcategory }} Products</title>
    <style>
        body { font-family: Arial, sans-serif; }
        .product { display: flex; align-items: center; margin-bottom: 20px; }
        .product img { width: 100px; margin-right: 10px; }
    </style>
</head>
<body>
    <h1>{{ subcategory }} Products</h1>
    <ul>
        {% for product in products %}
            <li class="product">
                <a href="{{ url_for('product_detail', category=category, subcategory=subcategory, product_id=product['product_code']) }}">
                    <img src="{{ product['imageurl'] }}" alt="{{ product['title'] }}">
                </a>
                <div>
                    <p><strong>{{ product['title'] }}</strong></p>
                    <p>Price: ${{ product['price'] }}</p>
                </div>
            </li>
        {% endfor %}
    </ul>
</body>
</html>
"""
product_detail_template = """
<!DOCTYPE html>
<html>
<head>
    <title>{{ product['title'] }}</title>
    <style>
        body { 
            font-family: Arial, sans-serif;
            margin: 0;
            padding: 20px;
        }
        .page-container {
            display: flex;
            gap: 30px;
            max-width: 1400px;
            margin: 0 auto;
        }
        .product-container {
            flex: 0 0 600px;
            position: sticky;
            top: 20px;
            align-self: flex-start;
        }
        .product-container img {
            width: 100%;
            max-width: 500px;
            height: auto;
        }
        .recommendations {
            flex: 1;
            min-width: 0;
        }
        .recommendations-grid {
            display: grid;
            grid-template-columns: repeat(auto-fill, minmax(200px, 1fr));
            gap: 20px;
            margin-top: 20px;
        }
        .recommendation-item {
            text-align: center;
        }
        .recommendation-item img {
            width: 100%;
            height: 200px;
            object-fit: contain;
        }
        .recommendation-item p {
            margin: 5px 0;
            font-size: 14px;
        }
        .back-link {
            display: block;
            margin-top: 20px;
        }
        h1 {
            font-size: 24px;
            margin-bottom: 20px;
        }
        h2 {
            font-size: 20px;
            margin-bottom: 15px;
        }
    </style>
</head>
<body>
    <div class="page-container">
        <div class="product-container">
            <h1>{{ product['title'] }}</h1>
            <img src="{{ product['imageurl'] }}" alt="{{ product['title'] }}">
            <p><strong>Category:</strong> {{ product['category2_code'] }} - {{ product['category3_code'] }}</p>
            <p><strong>Price:</strong> ${{ product['price'] }}</p>
            <a href="{{ url_for('product_page', category=category, subcategory=subcategory) }}" class="back-link">Back to Products</a>
        </div>

        <div class="recommendations">
            <h2>Recommended Products</h2>
            <div class="recommendations-grid">
                {% for rec in recommendations %}
                    <div class="recommendation-item">
                        <a href="{{ url_for('product_detail', category=rec['category'], subcategory=rec['subcategory'], product_id=rec['product_code']) }}">
                            <img src="{{ rec['imageurl'] }}" alt="{{ rec['title'] }}">
                        </a>
                        <p><strong>{{ rec['title'] }}</strong></p>
                        <p>${{ rec['price'] }}</p>
                    </div>
                {% endfor %}
            </div>
        </div>
    </div>
</body>
</html>
"""
@app.route("/")
def home():
    """Display the main categories."""
    categories = df["category2_code"].dropna().unique()
    return render_template_string(home_template, categories=categories)

@app.route("/category/<category>")
def subcategory_page(category):
    """Display subcategories under a main category."""
    subcategories = df[df["category2_code"] == category]["category3_code"].dropna().unique()
    return render_template_string(subcategory_template, category=category, subcategories=subcategories)

@app.route("/category/<category>/<subcategory>")
def product_page(category, subcategory):
    """Display products under a specific subcategory."""
    products = df[(df["category2_code"] == category) & (df["category3_code"] == subcategory)][["title", "imageurl", "price", "product_code"]].head(20).to_dict(orient="records")
    return render_template_string(product_template, category=category, subcategory=subcategory, products=products)



@app.route("/category/<category>/<subcategory>/<product_id>")
def product_detail(category, subcategory, product_id):
    """Display product details with recommendations from two separate LLM queries."""
    product = df[df["product_code"].astype(str) == product_id].iloc[0].to_dict()

    # Add product to viewed list
    if product not in viewed_products:
        viewed_products.append(product)

    # Generate two separate recommendation queries
    search_query_all_products = generate_recommendation_query(all_products=True)
    search_query_current_product = generate_recommendation_query(current_product=product)

    recommended_products = []

    # Fetch general recommendations from all viewed products
    if search_query_all_products:
        search_results_all = vector_store.similarity_search(search_query_all_products, k=50)
        sampled_results_all = random.sample(search_results_all, min(10, len(search_results_all)))

        recommended_products.extend([
            {
                "title": doc.metadata.get("title", "Unknown Title"),
                "imageurl": doc.metadata.get("imageurl", ""),
                "price": doc.metadata.get("price", "N/A"),
                "category": doc.metadata.get("category2_code", "Unknown"),
                "subcategory": doc.metadata.get("category3_code", "Unknown"),
                "product_code": doc.metadata.get("product_code", "Unknown")
            }
            for doc in sampled_results_all
        ])

    # Fetch 5 recommendations similar to the currently viewed product
    if search_query_current_product:
        search_results_current = vector_store.similarity_search(search_query_current_product, k=5)
        recommended_products.extend([
            {
                "title": doc.metadata.get("title", "Unknown Title"),
                "imageurl": doc.metadata.get("imageurl", ""),
                "price": doc.metadata.get("price", "N/A"),
                "category": doc.metadata.get("category2_code", "Unknown"),
                "subcategory": doc.metadata.get("category3_code", "Unknown"),
                "product_code": doc.metadata.get("product_code", "Unknown")
            }
            for doc in search_results_current
        ])

    return render_template_string(product_detail_template, category=category, subcategory=subcategory, product=product, recommendations=recommended_products)

@weave.op
def generate_recommendation_query(all_products=False, current_product=None):
    """Generate search queries for product recommendations using an LLM."""
    
    if all_products and viewed_products:
        sampled_products = random.sample(viewed_products, min(3, len(viewed_products)))
        product_descriptions = [
            f"Product: {p.get('title', '')}, Category: {p.get('category2_code', '')}, Subcategory: {p.get('category3_code', '')}"
            for p in sampled_products
        ]
        prompt = f"""
        The user has viewed the following products:
        {', '.join(product_descriptions)}

        Based on these products, generate a concise search query to find similar items with related styles, names, or categories.
        You can suggest other product categories that the user might be interested in.
        RESPOND ONLY with the query.
        """
    elif current_product:
        prompt = f"""
        The user is currently viewing:
        Product: {current_product.get('title', '')}, Category: {current_product.get('category2_code', '')}, Subcategory: {current_product.get('category3_code', '')}

        Generate a concise search query to find similar products with related styles, names, or categories.
        RESPOND ONLY with the query.
        """
    else:
        return None

    response = llm.invoke(prompt)
    return response.content.strip()


if __name__ == "__main__":
    app.run(debug=True)