实际应用案例

class ProductListingAgent:
    """商品上架 Agent"""
    
    async def process_new_products(self, supplier_data: list):
        """处理新商品上架"""
        results = []
        
        for product in supplier_data:
            # 1. 解析商品数据
            parsed = await self._parse_product_data(product)
            
            # 2. 处理商品图片
            images = await self._process_images(parsed["image_urls"])
            
            # 3. 计算价格（考虑利润率、运费、平台费用）
            price = await self._calculate_pricing(parsed)
            
            # 4. 生成商品描述
            description = await self._generate_description(parsed)
            
            # 5. 执行上架
            listing = await self._publish_to_platform({
                "name": parsed["name"],
                "price": price,
                "images": images,
                "description": description,
                "category": parsed["category"],
                "attributes": parsed["attributes"]
            })
            
            results.append({
                "product_id": product["id"],
                "listing_id": listing["id"],
                "status": "success"
            })
        
        return results
    
    async def _calculate_pricing(self, product: dict) -> float:
        """智能定价计算"""
        base_cost = product["cost"]
        
        # 考虑因素
        platform_fee = 0.05  # 平台佣金 5%
        shipping_cost = product.get("shipping", 5.0)
        target_margin = 0.20  # 目标利润率 20%
        
        # 计算售价
        price = (base_cost + shipping_cost) / (1 - platform_fee - target_margin)
        
        return round(price, 2)

class PriceMonitoringAgent:
    """竞品价格监控 Agent"""
    
    def __init__(self):
        self.agent = Agent(Config(
            name="price_monitor",
            role="价格监控助手",
            goal="实时监控竞品价格并预警"
        ))
        self.competitors = self._load_competitor_list()
    
    async def monitor_prices(self):
        """执行价格监控"""
        all_prices = {}
        
        for competitor in self.competitors:
            prices = await self._scrape_competitor(competitor)
            all_prices[competitor["name"]] = prices
        
        # 价格分析
        analysis = self._analyze_prices(all_prices)
        
        # 检测异常
        alerts = self._detect_changes(analysis)
        
        # 发送告警
        if alerts:
            await self._send_price_alerts(alerts)
        
        return analysis
    
    async def _scrape_competitor(self, competitor: dict) -> list:
        """抓取竞品价格"""
        browser = BrowserTool()
        products = []
        
        for category_url in competitor["category_urls"]:
            await browser.goto(category_url)
            
            # 滚动加载
            await browser.scroll_multiple(3)
            
            # 提取价格
            page_products = await browser.evaluate("""
                Array.from(document.querySelectorAll('.product-item'))
                    .map(item => ({
                        name: item.querySelector('.title')?.textContent?.trim(),
                        price: parseFloat(item.querySelector('.price')?.dataset?.value || 0),
                        url: item.querySelector('a')?.href,
                        in_stock: item.querySelector('.stock-status')?.textContent?.includes('In Stock')
                    }))
            """)
            
            products.extend(page_products)
        
        return products
    
    def _detect_changes(self, analysis: dict) -> list:
        """检测价格变化"""
        alerts = []
        
        for product, data in analysis["price_history"].items():
            if len(data) < 2:
                continue
            
            current = data[-1]["price"]
            previous = data[-2]["price"]
            change_pct = (current - previous) / previous * 100
            
            if abs(change_pct) > 10:  # 价格变动超过 10%
                alerts.append({
                    "product": product,
                    "change_pct": round(change_pct, 2),
                    "direction": "increase" if change_pct > 0 else "decrease",
                    "competitor": data[-1]["competitor"]
                })
        
        return alerts

class BankingCustomerServiceAgent:
    """银行客服 Agent"""
    
    def __init__(self):
        self.agent = Agent(Config(
            name="banking_cs",
            role="银行客服助手",
            goal="处理客户咨询，提供准确信息"
        ))
        self.knowledge_base = self._load_knowledge_base()
        self.escalation_keywords = ["投诉", "经理", "愤怒", "法律"]
    
    async def handle_inquiry(self, message: str, customer_id: str) -> dict:
        """处理客户咨询"""
        
        # 1. 理解问题意图
        intent = await self._understand_intent(message)
        
        # 2. 检测情绪和紧急程度
        sentiment = await self._analyze_sentiment(message)
        
        # 3. 检查是否需要人工介入
        if self._needs_human_support(message, sentiment):
            return await self._escalate_to_human(message, customer_id, sentiment)
        
        # 4. 检索知识库
        relevant_docs = self._search_knowledge_base(intent, message)
        
        # 5. 生成回复
        response = await self._generate_response(intent, relevant_docs)
        
        # 6. 记录对话
        await self._log_interaction(customer_id, intent, response)
        
        return {
            "response": response,
            "intent": intent,
            "requires_followup": self._needs_followup(intent)
        }
    
    async def _handle_account_inquiry(self, customer_id: str) -> dict:
        """处理账户查询"""
        account = await self._get_account_info(customer_id)
        
        return {
            "response": f"""您的账户信息如下：
            - 账户号码: {account['number']}
            - 账户类型: {account['type']}
            - 当前余额: ¥{account['balance']:,}
            - 上次交易: {account['last_transaction']}
            
            如需更多帮助，请告诉我。""",
            "account_info": account
        }
    
    async def _handle_transfer_inquiry(self, customer_id: str, transfer_info: dict) -> dict:
        """处理转账咨询"""
        account = await self._get_account_info(customer_id)
        
        # 验证账户余额
        if account["balance"] < transfer_info["amount"]:
            return {
                "response": "抱歉，您的账户余额不足，无法完成此转账。",
                "status": "failed"
            }
        
        # 检查转账限额
        if transfer_info["amount"] > account["daily_limit"]:
            return {
                "response": f"单笔转账限额为 ¥{account['daily_limit']:,}，请降低转账金额。",
                "status": "exceeded_limit"
            }
        
        # 执行转账
        result = await self._execute_transfer(customer_id, transfer_info)
        
        return {
            "response": f"转账成功！交易编号: {result['txn_id']}",
            "status": "success",
            "transaction": result
        }

class ComplianceCheckAgent:
    """合规检查 Agent"""
    
    async def check_transaction(self, transaction: dict) -> dict:
        """
        检查交易合规性
        
        Args:
            transaction: 交易信息
        
        Returns:
            合规检查结果
        """
        checks = []
        
        # 1. 金额检查
        amount_check = await self._check_amount_limits(transaction)
        checks.append(amount_check)
        
        # 2. 频率检查
        frequency_check = await self._check_frequency_rules(transaction)
        checks.append(frequency_check)
        
        # 3. 黑名单检查
        blacklist_check = await self._check_blacklist(transaction)
        checks.append(blacklist_check)
        
        # 4. 可疑交易检测
        suspicious_check = await self._detect_suspicious_activity(transaction)
        checks.append(suspicious_check)
        
        # 综合判断
        all_passed = all(c["passed"] for c in checks)
        high_risk = any(c["risk_level"] == "high" for c in checks)
        
        if all_passed:
            return {
                "status": "approved",
                "checks": checks,
                "message": "交易通过合规检查"
            }
        elif high_risk:
            return {
                "status": "rejected",
                "checks": checks,
                "message": "交易被拒绝：高风险"
            }
        else:
            return {
                "status": "review",
                "checks": checks,
                "message": "交易需要人工审核"
            }
    
    async def _detect_suspicious_activity(self, transaction: dict) -> dict:
        """可疑交易检测"""
        risk_score = 0
        reasons = []
        
        # 检测非常规时间交易
        if self._is_unusual_time():
            risk_score += 20
            reasons.append("非常规交易时间")
        
        # 检测大额交易
        if transaction["amount"] > 100000:
            risk_score += 30
            reasons.append("大额交易")
        
        # 检测频繁交易
        recent_count = await self._count_recent_transactions(
            transaction["account_id"],
            hours=24
        )
        if recent_count > 10:
            risk_score += 25
            reasons.append("交易频率异常")
        
        # 检测新账户大额交易
        if await self._is_new_account(transaction["account_id"], days=30):
            if transaction["amount"] > 50000:
                risk_score += 20
                reasons.append("新账户大额交易")
        
        return {
            "check": "suspicious_activity",
            "passed": risk_score < 50,
            "risk_score": risk_score,
            "risk_level": "high" if risk_score >= 50 else "medium" if risk_score >= 30 else "low",
            "reasons": reasons
        }

class ResumeScreeningAgent:
    """简历筛选 Agent"""
    
    def __init__(self):
        self.agent = Agent(Config(
            name="resume_screener",
            role="简历筛选助手",
            goal="高效筛选匹配候选人"
        ))
        self.job_requirements = {}
    
    def load_job_requirements(self, job_id: str, requirements: dict):
        """加载岗位需求"""
        self.job_requirements[job_id] = requirements
    
    async def screen_resume(self, resume: dict, job_id: str) -> dict:
        """筛选简历"""
        requirements = self.job_requirements.get(job_id)
        
        # 1. 基本条件匹配
        basic_match = self._check_basic_requirements(resume, requirements)
        
        if not basic_match["passed"]:
            return {
                "status": "rejected",
                "reason": basic_match["reason"],
                "score": 0
            }
        
        # 2. 技能匹配度评估
        skills_match = self._assess_skills_match(resume, requirements)
        
        # 3. 经验匹配度评估
        experience_match = self._assess_experience_match(resume, requirements)
        
        # 4. 文化契合度评估
        culture_match = self._assess_culture_fit(resume, requirements)
        
        # 综合评分
        total_score = (
            skills_match["score"] * 0.4 +
            experience_match["score"] * 0.4 +
            culture_match["score"] * 0.2
        )
        
        # 决策
        if total_score >= 80:
            decision = "recommend"
        elif total_score >= 60:
            decision = "interview"
        elif total_score >= 40:
            decision = "consider"
        else:
            decision = "reject"
        
        return {
            "status": decision,
            "score": total_score,
            "breakdown": {
                "skills": skills_match,
                "experience": experience_match,
                "culture": culture_match
            },
            "highlights": self._extract_highlights(resume),
            "concerns": self._extract_concerns(resume, requirements)
        }
    
    def _assess_skills_match(self, resume: dict, requirements: dict) -> dict:
        """评估技能匹配度"""
        required_skills = set(requirements.get("skills", []))
        resume_skills = set(resume.get("skills", []))
        
        matched = required_skills & resume_skills
        missing = required_skills - resume_skills
        
        match_rate = len(matched) / len(required_skills) if required_skills else 1
        
        # 额外技能加分
        extra_skills = resume_skills - required_skills
        bonus = min(len(extra_skills) * 2, 10)  # 最多加10分
        
        return {
            "score": min(match_rate * 100 + bonus, 100),
            "matched": list(matched),
            "missing": list(missing),
            "bonus_points": bonus
        }
    
    async def batch_screen(self, resumes: list, job_id: str) -> dict:
        """批量筛选简历"""
        results = []
        
        for resume in resumes:
            result = await self.screen_resume(resume, job_id)
            results.append({
                **result,
                "candidate_id": resume["id"],
                "name": resume["name"]
            })
        
        # 按分数排序
        results.sort(key=lambda x: x["score"], reverse=True)
        
        return {
            "total": len(resumes),
            "recommended": sum(1 for r in results if r["status"] == "recommend"),
            "interview": sum(1 for r in results if r["status"] == "interview"),
            "consider": sum(1 for r in results if r["status"] == "consider"),
            "rejected": sum(1 for r in results if r["status"] == "reject"),
            "candidates": results
        }

class SocialMediaAgent:
    """社交媒体运营 Agent"""
    
    def __init__(self):
        self.agent = Agent(Config(
            name="social_media",
            role="社交媒体运营助手",
            goal="自动化内容发布和互动管理"
        ))
        self.platforms = {
            "twitter": TwitterAdapter(),
            "weibo": WeiboAdapter(),
            "linkedin": LinkedInAdapter()
        }
        self.content_calendar = self._load_content_calendar()
    
    async def publish_content(self, content: dict, platforms: list) -> dict:
        """发布内容到多个平台"""
        results = {}
        
        for platform in platforms:
            if platform not in self.platforms:
                continue
            
            # 平台适配
            adapted_content = self._adapt_for_platform(content, platform)
            
            # 发布
            result = await self.platforms[platform].post(adapted_content)
            results[platform] = result
        
        return {
            "content_id": content["id"],
            "published_to": list(results.keys()),
            "results": results
        }
    
    async def manage_engagement(self, platform: str) -> dict:
        """管理社交互动"""
        adapter = self.platforms[platform]
        
        # 1. 获取提及和评论
        mentions = await adapter.get_mentions()
        comments = await adapter.get_comments()
        
        # 2. 分类互动类型
        categorized = self._categorize_interactions(mentions + comments)
        
        # 3. 自动回复（简单问题）
        auto_replies = []
        for interaction in categorized["questions"]:
            if self._can_auto_reply(interaction):
                reply = await self._generate_reply(interaction)
                await adapter.reply(interaction["id"], reply)
                auto_replies.append(interaction["id"])
        
        # 4. 标记需要人工处理
        human_required = categorized["complaints"] + categorized["complex"]
        
        return {
            "total_interactions": len(mentions) + len(comments),
            "auto_replied": len(auto_replies),
            "human_required": human_required,
            "sentiment_analysis": self._analyze_sentiment(categorized)
        }
    
    async def generate_report(self, platform: str, date_range: tuple) -> dict:
        """生成运营报告"""
        adapter = self.platforms[platform]
        
        # 收集数据
        posts = await adapter.get_posts(date_range)
        engagement = await adapter.get_engagement_stats(date_range)
        followers = await adapter.get_follower_stats(date_range)
        
        # 分析
        analysis = {
            "total_posts": len(posts),
            "total_reach": sum(p["reach"] for p in posts),
            "avg_engagement_rate": sum(p["engagement_rate"] for p in posts) / len(posts),
            "top_posts": sorted(posts, key=lambda x: x["engagement"], reverse=True)[:5],
            "follower_growth": followers["growth"],
            "best_posting_time": self._find_best_posting_time(posts)
        }
        
        # 生成建议
        suggestions = self._generate_suggestions(analysis)
        
        return {
            "platform": platform,
            "date_range": date_range,
            "analysis": analysis,
            "suggestions": suggestions
        }

class OperationsMonitoringAgent:
    """运维监控 Agent"""
    
    def __init__(self):
        self.agent = Agent(Config(
            name="ops_monitor",
            role="运维监控助手",
            goal="自动化监控和故障处理"
        ))
        self.alert_handlers = {
            "high_cpu": self._handle_high_cpu,
            "memory_pressure": self._handle_memory_pressure,
            "disk_full": self._handle_disk_full,
            "service_down": self._handle_service_down
        }
    
    async def process_alert(self, alert: dict) -> dict:
        """处理告警"""
        alert_type = alert["type"]
        
        if alert_type not in self.alert_handlers:
            return {
                "status": "unhandled",
                "alert": alert,
                "message": "未知的告警类型"
            }
        
        # 执行对应的处理逻辑
        handler = self.alert_handlers[alert_type]
        result = await handler(alert)
        
        # 记录处理历史
        await self._log_alert_handling(alert, result)
        
        return result
    
    async def _handle_high_cpu(self, alert: dict) -> dict:
        """处理高 CPU 告警"""
        server = alert["server"]
        
        # 1. 检查 CPU 使用率趋势
        trend = await self._get_cpu_trend(server, minutes=30)
        
        if trend["increasing"]:
            # 2. 识别 CPU 密集型进程
            top_processes = await self._get_top_cpu_processes(server)
            
            # 3. 检查是否有异常进程
            suspicious = await self._identify_suspicious_processes(top_processes)
            
            if suspicious:
                # 终止可疑进程
                for proc in suspicious:
                    await self._kill_process(server, proc["pid"])
                
                return {
                    "status": "resolved",
                    "action": "killed_suspicious_processes",
                    "processes": suspicious
                }
            
            # 4. 如果是正常流量增长，发送扩容建议
            return {
                "status": "escalation_required",
                "reason": "CPU使用率持续上升",
                "suggestion": "考虑扩容服务器或优化应用",
                "current_usage": alert["current_usage"],
                "trend": trend
            }
        
        # 短暂峰值，观察
        return {
            "status": "monitoring",
            "message": "CPU使用率暂时性波动，继续观察"
        }
    
    async def _handle_service_down(self, alert: dict) -> dict:
        """处理服务宕机告警"""
        service = alert["service"]
        
        # 1. 检查服务状态
        status = await self._check_service_status(service)
        
        if status["running"]:
            return {
                "status": "false_alert",
                "message": "服务已恢复正常"
            }
        
        # 2. 尝试自动重启
        restart_result = await self._restart_service(service)
        
        if restart_result["success"]:
            # 3. 验证服务恢复
            health_check = await self._wait_for_healthy(service, timeout=60)
            
            if health_check["healthy"]:
                return {
                    "status": "resolved",
                    "action": "auto_restart",
                    "message": "服务已自动重启并恢复"
                }
        
        # 4. 需要人工介入
        return {
            "status": "escalation_required",
            "severity": "critical",
            "service": service,
            "actions_tried": ["auto_restart"],
            "message": "自动恢复失败，需要运维人员介入"
        }
    
    async def run_health_check(self) -> dict:
        """系统健康检查"""
        results = {}
        
        # 检查所有监控的服务
        services = await self._get_monitored_services()
        
        for service in services:
            health = await self._check_service_health(service)
            results[service["name"]] = health
        
        # 生成健康报告
        healthy_count = sum(1 for r in results.values() if r["healthy"])
        total_count = len(results)
        
        return {
            "timestamp": datetime.now().isoformat(),
            "overall_health": "healthy" if healthy_count == total_count else "degraded",
            "services_healthy": healthy_count,
            "services_total": total_count,
            "details": results
        }