# 传统LLM应用
def process_query(user_input):
    response = llm.generate(user_input)
    return response

# Agent应用
class AgentSession:
    def __init__(self):
        self.conversation_history = []
        self.task_state = {}
        self.available_tools = []
        self.execution_plan = []
    
    def process_task(self, user_goal):
        # 维护会话状态，支持多轮交互
        pass

# 传统方式：硬编码工具调用
if "天气" in user_input:
    result = weather_api.get_weather()

# Agent方式：智能工具选择
def select_tools(task_context, available_tools):
    # 基于任务上下文动态选择工具
    tool_selection_prompt = f"""
    Task: {task_context}
    Available tools: {[tool.description for tool in available_tools]}
    Select the most appropriate tools and explain why.
    """
    return llm.analyze_and_select(tool_selection_prompt)

class ReActAgent:
    def solve_task(self, task):
        thought = self.think(task)  # 思考阶段
        action = self.plan_action(thought)  # 规划行动
        observation = self.execute_action(action)  # 执行并观察
        
        if self.is_task_complete(observation):
            return self.generate_final_response(observation)
        else:
            # 继续下一轮推理
            return self.solve_task(self.update_context(task, observation))

# 单模型架构：简单但可能存在能力瓶颈
class SingleModelAgent:
    def __init__(self, model_name="gpt-4"):
        self.llm = load_model(model_name)
    
    def process(self, task):
        return self.llm.generate(task)

# 多模型架构：复杂但性能更优
class MultiModelAgent:
    def __init__(self):
        self.planner_model = load_model("gpt-4")  # 规划能力强
        self.executor_model = load_model("claude-3.5")  # 代码执行好
        self.critic_model = load_model("gemini-pro")  # 结果评估
    
    def process(self, task):
        plan = self.planner_model.generate_plan(task)
        result = self.executor_model.execute(plan)
        evaluation = self.critic_model.evaluate(result)
        return self.integrate_results(result, evaluation)

AGENT_PROMPT_TEMPLATE = """
You are an AI agent designed to complete complex tasks.
AVAILABLE TOOLS:
{tools_description}
TASK: {user_task}
CONTEXT: {conversation_history}
INSTRUCTIONS:
1. Think step-by-step about how to complete this task
2. Select appropriate tools from the available options
3. Execute actions and observe results
4. If the task is not complete, continue with the next step
5. Provide a final summary when the task is done
Begin your reasoning:
"""

class ContextManager:
    def __init__(self, max_tokens=8192):
        self.max_tokens = max_tokens
        self.conversation_history = []
    
    def add_interaction(self, user_input, agent_response):
        self.conversation_history.append({
            'user': user_input,
            'agent': agent_response,
            'timestamp': time.time()
        })
        self._truncate_if_needed()
    
    def _truncate_if_needed(self):
        # 基于token数量和重要性截断历史记录
        while self._count_tokens() > self.max_tokens:
            self.conversation_history.pop(0)

class WeatherExtension:
    def __init__(self, api_key):
        self.api_key = api_key
        self.description = "获取指定城市的天气信息"
        self.parameters = {
            "city": "string, required, 城市名称",
            "days": "int, optional, 预报天数，默认1天"
        }
    
    def execute(self, city, days=1):
        try:
            response = requests.get(
                f"https://api.weather.com/v1/forecast",
                params={"key": self.api_key, "q": city, "days": days}
            )
            return response.json()
        except Exception as e:
            return {"error": str(e)}

class ExtensionManager:
    def __init__(self):
        self.extensions = {}
    
    def register(self, name, extension):
        self.extensions[name] = extension
    
    def get_tool_descriptions(self):
        descriptions = []
        for name, ext in self.extensions.items():
            descriptions.append({
                'name': name,
                'description': ext.description,
                'parameters': ext.parameters
            })
        return descriptions
    
    def execute_tool(self, tool_name, **kwargs):
        if tool_name in self.extensions:
            return self.extensions[tool_name].execute(**kwargs)
        else:
            return {"error": f"Tool {tool_name} not found"}

class DatabaseFunction:
    @staticmethod
    def get_schema():
        return {
            "name": "query_database",
            "description": "查询数据库并返回结果",
            "parameters": {
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SQL查询语句"},
                    "limit": {"type": "integer", "description": "返回结果数量限制"}
                },
                "required": ["sql"]
            }
        }
    
    @staticmethod
    def execute(sql, limit=100):
        # 客户端执行，可以进行安全检查
        if not DatabaseFunction.is_safe_query(sql):
            return {"error": "Unsafe query detected"}
        
        # 执行查询逻辑
        # ...
        pass
    
    @staticmethod
    def is_safe_query(sql):
        # 实现SQL安全检查逻辑
        dangerous_keywords = ['DROP', 'DELETE', 'UPDATE', 'INSERT']
        return not any(keyword in sql.upper() for keyword in dangerous_keywords)

# Agent调用示例
class FunctionCallingAgent:
    def __init__(self):
        self.available_functions = [DatabaseFunction.get_schema()]
    
    def process_query(self, user_input):
        # Agent生成函数调用
        function_call = self.llm.generate_function_call(
            user_input, 
            self.available_functions
        )
        
        if function_call['name'] == 'query_database':
            # 返回函数调用给客户端执行
            return {
                "type": "function_call",
                "function": function_call['name'],
                "parameters": function_call['parameters']
            }
        
        return {"type": "text_response", "content": "..."}

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class ProductionRAGSystem:
    def __init__(self, embedding_model_name="sentence-transformers/all-MiniLM-L6-v2"):
        self.embedding_model = SentenceTransformer(embedding_model_name)
        self.vector_index = None
        self.document_store = []
        self.chunk_size = 512
        self.overlap_size = 50
    
    def add_documents(self, documents):
        """添加文档到向量数据库"""
        chunks = []
        for doc in documents:
            doc_chunks = self._chunk_document(doc)
            chunks.extend(doc_chunks)
        
        # 生成embedding
        embeddings = self.embedding_model.encode(chunks)
        
        # 构建FAISS索引
        if self.vector_index is None:
            dimension = embeddings.shape[1]
            self.vector_index = faiss.IndexFlatIP(dimension)
        
        # 标准化向量
        faiss.normalize_L2(embeddings)
        self.vector_index.add(embeddings.astype('float32'))
        self.document_store.extend(chunks)
    
    def _chunk_document(self, document):
        """文档分块策略"""
        words = document.split()
        chunks = []
        
        for i in range(0, len(words), self.chunk_size - self.overlap_size):
            chunk_words = words[i:i + self.chunk_size]
            chunk_text = ' '.join(chunk_words)
            chunks.append(chunk_text)
        
        return chunks
    
    def search(self, query, top_k=5):
        """检索相关文档"""
        if self.vector_index is None:
            return []
        
        query_embedding = self.embedding_model.encode([query])
        faiss.normalize_L2(query_embedding)
        
        scores, indices = self.vector_index.search(query_embedding.astype('float32'), top_k)
        
        results = []
        for score, idx in zip(scores[0], indices[0]):
            if idx < len(self.document_store):
                results.append({
                    'content': self.document_store[idx],
                    'score': float(score)
                })
        
        return results
    
    def generate_answer(self, query, context_docs):
        """基于检索结果生成回答"""
        context = '\n\n'.join([doc['content'] for doc in context_docs])
        
        prompt = f"""
        Based on the following context, answer the user's question.
        
        Context:
        {context}
        
        Question: {query}
        
        Answer:
        """
        
        return self.llm.generate(prompt)

# 使用示例
class RAGAgent:
    def __init__(self):
        self.rag_system = ProductionRAGSystem()
        self.llm = load_model("gpt-4")
    
    def add_knowledge(self, documents):
        self.rag_system.add_documents(documents)
    
    def answer_query(self, query):
        # 检索相关文档
        relevant_docs = self.rag_system.search(query, top_k=3)
        
        # 生成回答
        if relevant_docs:
            return self.rag_system.generate_answer(query, relevant_docs)
        else:
            return "抱歉，我没有找到相关信息来回答您的问题。"

class ReActAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        self.max_iterations = 10
    
    def solve_task(self, task):
        context = f"Task: {task}\n\n"
        
        for iteration in range(self.max_iterations):
            # Thought: 分析当前情况
            thought = self._generate_thought(context)
            context += f"Thought {iteration + 1}: {thought}\n"
            
            # Action: 决定下一步行动
            action = self._generate_action(context)
            context += f"Action {iteration + 1}: {action}\n"
            
            # 解析行动
            if action.startswith("Final Answer:"):
                return action[len("Final Answer:"):].strip()
            
            # 执行工具调用
            try:
                tool_name, tool_input = self._parse_action(action)
                observation = self.tools[tool_name].execute(tool_input)
                context += f"Observation {iteration + 1}: {observation}\n\n"
            except Exception as e:
                context += f"Observation {iteration + 1}: Error - {str(e)}\n\n"
        
        return "任务执行超时，请简化任务或检查工具配置。"
    
    def _generate_thought(self, context):
        prompt = f"""
        {context}
        
        分析当前情况，思考下一步应该怎么做。
        Thought:"""
        
        response = self.llm.generate(prompt, max_tokens=200)
        return response.strip()
    
    def _generate_action(self, context):
        tool_descriptions = '\n'.join([
            f"- {name}: {tool.description}"
            for name, tool in self.tools.items()
        ])
        
        prompt = f"""
        {context}
        
        可用工具：
        {tool_descriptions}
        
        基于以上思考，选择下一步行动。格式：
        - 使用工具：工具名称[输入参数]
        - 结束任务：Final Answer: [最终答案]
        
        Action:"""
        
        response = self.llm.generate(prompt, max_tokens=100)
        return response.strip()
    
    def _parse_action(self, action):
        # 解析 "工具名称[输入参数]" 格式
        if '[' in action and ']' in action:
            tool_name = action.split('[')[0].strip()
            tool_input = action.split('[')[1].split(']')[0]
            return tool_name, tool_input
        else:
            raise ValueError(f"无法解析行动格式: {action}")

class RobustAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.error_recovery_attempts = 3
    
    def solve_task(self, task):
        try:
            return self._solve_with_recovery(task)
        except Exception as e:
            return f"任务执行失败: {str(e)}"
    
    def _solve_with_recovery(self, task):
        context = f"Task: {task}\n\n"
        errors = []
        
        for iteration in range(10):
            try:
                # 正常执行逻辑
                result = self._execute_iteration(context)
                if result.get('finished'):
                    return result['answer']
                context = result['updated_context']
                
            except Exception as e:
                errors.append(str(e))
                
                # 错误恢复策略
                if len(errors) <= self.error_recovery_attempts:
                    recovery_prompt = f"""
                    执行过程中遇到错误：{str(e)}
                    
                    历史错误：{'; '.join(errors)}
                    
                    请分析错误原因并调整策略。继续执行任务：{task}
                    """
                    context += f"Error Recovery: {recovery_prompt}\n"
                    continue
                else:
                    raise e
        
        return "任务执行超时"
    
    def _execute_iteration(self, context):
        # 具体的执行逻辑
        pass

import asyncio
from concurrent.futures import ThreadPoolExecutor

class OptimizedAgent:
    def __init__(self):
        self.llm_pool = ThreadPoolExecutor(max_workers=3)
        self.tool_cache = {}  # 工具调用结果缓存
    
    async def parallel_tool_calls(self, tool_calls):
        """并行执行多个工具调用"""
        tasks = []
        for tool_call in tool_calls:
            if self._can_cache(tool_call):
                cache_key = self._get_cache_key(tool_call)
                if cache_key in self.tool_cache:
                    continue
            
            task = asyncio.create_task(self._execute_tool_async(tool_call))
            tasks.append(task)
        
        results = await asyncio.gather(*tasks)
        return results
    
    def _can_cache(self, tool_call):
        # 判断工具调用是否可以缓存
        cacheable_tools = ['weather', 'static_data_query']
        return tool_call['name'] in cacheable_tools
    
    async def _execute_tool_async(self, tool_call):
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(
            self.llm_pool, 
            self._execute_tool_sync, 
            tool_call
        )

class CostOptimizedAgent:
    def __init__(self):
        self.cost_tracker = {
            'input_tokens': 0,
            'output_tokens': 0,
            'tool_calls': 0
        }
        self.cost_limits = {
            'max_tokens_per_task': 10000,
            'max_tool_calls_per_task': 20
        }
    
    def process_with_budget(self, task):
        if self._check_budget():
            return self._process_task(task)
        else:
            return "任务超出预算限制，请简化任务或增加预算。"
    
    def _check_budget(self):
        return (
            self.cost_tracker['input_tokens'] + self.cost_tracker['output_tokens'] 
            < self.cost_limits['max_tokens_per_task']
            and 
            self.cost_tracker['tool_calls'] < self.cost_limits['max_tool_calls_per_task']
        )
    
    def _track_usage(self, input_tokens, output_tokens, tool_calls=0):
        self.cost_tracker['input_tokens'] += input_tokens
        self.cost_tracker['output_tokens'] += output_tokens
        self.cost_tracker['tool_calls'] += tool_calls

class QualityAssuredAgent:
    def __init__(self, primary_llm, validator_llm):
        self.primary_llm = primary_llm
        self.validator_llm = validator_llm
    
    def solve_task_with_validation(self, task):
        # 主Agent执行任务
        primary_result = self.primary_agent.solve_task(task)
        
        # 验证器检查结果
        validation_prompt = f"""
        任务：{task}
        执行结果：{primary_result}
        
        请评估这个结果是否：
        1. 正确回答了问题
        2. 逻辑清晰
        3. 没有明显错误
        
        如果有问题，请指出具体问题。
        评估结果：
        """
        
        validation = self.validator_llm.generate(validation_prompt)
        
        if "有问题" in validation or "错误" in validation:
            # 重新执行或修正
            correction_prompt = f"""
            原始任务：{task}
            初始结果：{primary_result}
            发现的问题：{validation}
            
            请基于问题反馈，重新执行任务或修正结果。
            """
            corrected_result = self.primary_llm.generate(correction_prompt)
            return corrected_result
        
        return primary_result

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain_community.tools import DuckDuckGoSearchRun

# 快速搭建Agent
def create_production_agent():
    # 初始化模型
    llm = ChatOpenAI(model="gpt-4-1106-preview", temperature=0)
    
    # 定义工具
    search = DuckDuckGoSearchRun()
    tools = [
        Tool(
            name="Search",
            func=search.run,
            description="搜索最新信息"
        ),
        Tool(
            name="Calculator",
            func=lambda x: eval(x),  # 生产环境需要安全的计算器
            description="执行数学计算"
        )
    ]
    
    # 创建Agent
    agent = create_openai_tools_agent(llm, tools, prompt_template)
    return AgentExecutor(agent=agent, tools=tools, verbose=True)

# 使用示例
agent = create_production_agent()
result = agent.invoke({"input": "帮我查找最新的AI技术趋势"})

class ProductionAgentFramework:
    def __init__(self, config):
        self.config = config
        self.llm = self._init_llm()
        self.tools = self._init_tools()
        self.memory = self._init_memory()
        self.monitor = self._init_monitoring()
    
    def _init_llm(self):
        # 根据配置初始化模型
        model_config = self.config['model']
        if model_config['provider'] == 'openai':
            return ChatOpenAI(**model_config['params'])
        elif model_config['provider'] == 'anthropic':
            return ChatAnthropic(**model_config['params'])
        # ... 其他模型
    
    def _init_tools(self):
        tools = []
        for tool_config in self.config['tools']:
            tool_class = self._get_tool_class(tool_config['type'])
            tool = tool_class(**tool_config['params'])
            tools.append(tool)
        return tools
    
    def _init_memory(self):
        # 初始化记忆系统
        if self.config['memory']['type'] == 'redis':
            return RedisMemory(**self.config['memory']['params'])
        elif self.config['memory']['type'] == 'postgresql':
            return PostgreSQLMemory(**self.config['memory']['params'])
        else:
            return InMemoryMemory()
    
    def _init_monitoring(self):
        # 初始化监控系统
        return AgentMonitor(
            metrics_backend=self.config['monitoring']['backend'],
            alert_thresholds=self.config['monitoring']['thresholds']
        )
    
    def process_request(self, user_id, task):
        with self.monitor.track_execution():
            try:
                # 加载用户会话
                session = self.memory.get_session(user_id)
                
                # 执行任务
                result = self._execute_task(task, session)
                
                # 保存会话状态
                self.memory.save_session(user_id, session)
                
                # 记录成功指标
                self.monitor.record_success(task, result)
                
                return result
                
            except Exception as e:
                # 记录错误
                self.monitor.record_error(task, e)
                raise e

import time
import logging
from dataclasses import dataclass
from typing import Dict, Any

@dataclass
class ExecutionMetrics:
    task_id: str
    start_time: float
    end_time: float
    token_usage: Dict[str, int]
    tool_calls: int
    success: bool
    error_message: str = None

class AgentMonitor:
    def __init__(self):
        self.metrics_store = []
        self.alert_thresholds = {
            'max_execution_time': 30.0,  # 秒
            'max_token_usage': 5000,
            'error_rate_threshold': 0.1  # 10%
        }
    
    def track_execution(self):
        return ExecutionTracker(self)
    
    def analyze_performance(self, time_window_hours=24):
        recent_metrics = self._get_recent_metrics(time_window_hours)
        
        if not recent_metrics:
            return "没有足够的数据进行分析"
        
        # 计算性能指标
        avg_execution_time = sum(m.end_time - m.start_time for m in recent_metrics) / len(recent_metrics)
        success_rate = sum(1 for m in recent_metrics if m.success) / len(recent_metrics)
        avg_token_usage = sum(m.token_usage.get('total', 0) for m in recent_metrics) / len(recent_metrics)
        
        # 生成报告
        report = f"""
        Agent性能报告 (最近{time_window_hours}小时):
        - 平均执行时间: {avg_execution_time:.2f}秒
        - 成功率: {success_rate:.2%}
        - 平均Token使用量: {avg_token_usage:.0f}
        - 处理任务数: {len(recent_metrics)}
        """
        
        # 检查告警
        if avg_execution_time > self.alert_thresholds['max_execution_time']:
            report += f"\n⚠️ 执行时间超过阈值 ({self.alert_thresholds['max_execution_time']}s)"
        
        if success_rate < (1 - self.alert_thresholds['error_rate_threshold']):
            report += f"\n⚠️ 错误率过高 (>{self.alert_thresholds['error_rate_threshold']:.1%})"
        
        return report

class ExecutionTracker:
    def __init__(self, monitor):
        self.monitor = monitor
        self.start_time = None
        self.metrics = None
    
    def __enter__(self):
        self.start_time = time.time()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        end_time = time.time()
        
        self.metrics = ExecutionMetrics(
            task_id=str(time.time()),
            start_time=self.start_time,
            end_time=end_time,
            token_usage={'total': 0},  # 需要从实际执行中获取
            tool_calls=0,
            success=exc_type is None,
            error_message=str(exc_val) if exc_val else None
        )
        
        self.monitor.metrics_store.append(self.metrics)

class RobustToolManager:
    def __init__(self, tools, retry_config=None):
        self.tools = tools
        self.retry_config = retry_config or {
            'max_retries': 3,
            'backoff_factor': 2,
            'timeout': 30
        }
    
    def execute_tool(self, tool_name, **kwargs):
        tool = self.tools.get(tool_name)
        if not tool:
            return {"error": f"Tool '{tool_name}' not found"}
        
        for attempt in range(self.retry_config['max_retries']):
            try:
                result = self._execute_with_timeout(tool, kwargs)
                return {"success": True, "result": result}
            
            except TimeoutError:
                if attempt == self.retry_config['max_retries'] - 1:
                    return {"error": "Tool execution timeout"}
                time.sleep(self.retry_config['backoff_factor'] ** attempt)
            
            except Exception as e:
                if attempt == self.retry_config['max_retries'] - 1:
                    return {"error": f"Tool execution failed: {str(e)}"}
                time.sleep(self.retry_config['backoff_factor'] ** attempt)
        
        return {"error": "Max retries exceeded"}
    
    def _execute_with_timeout(self, tool, kwargs):
        import signal
        
        def timeout_handler(signum, frame):
            raise TimeoutError("Tool execution timeout")
        
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(self.retry_config['timeout'])
        
        try:
            result = tool.execute(**kwargs)
            signal.alarm(0)  # 取消超时
            return result
        except Exception as e:
            signal.alarm(0)
            raise e

import tiktoken

class TokenManager:
    def __init__(self, model_name="gpt-4"):
        self.encoding = tiktoken.encoding_for_model(model_name)
        self.token_limits = {
            'input_limit': 6000,    # 输入token限制
            'output_limit': 2000,   # 输出token限制
            'context_limit': 8000   # 总上下文限制
        }
    
    def count_tokens(self, text):
        return len(self.encoding.encode(text))
    
    def truncate_context(self, context, max_tokens):
        """智能截断上下文，保留重要信息"""
        current_tokens = self.count_tokens(context)
        
        if current_tokens <= max_tokens:
            return context
        
        # 分离不同部分
        parts = context.split('\n\n')
        
        # 按重要性排序（系统提示 > 最近对话 > 历史对话）
        system_parts = [p for p in parts if 'System:' in p or 'Task:' in p]
        recent_parts = parts[-3:]  # 最近3轮对话
        other_parts = [p for p in parts if p not in system_parts and p not in recent_parts]
        
        # 重新组合
        result = '\n\n'.join(system_parts)
        result += '\n\n' + '\n\n'.join(recent_parts)
        
        # 如果还是超限，继续截断
        while self.count_tokens(result) > max_tokens and other_parts:
            if len(other_parts) > 0:
                other_parts.pop(0)
            if other_parts:
                result = '\n\n'.join(system_parts) + '\n\n' + '\n\n'.join(other_parts[-2:]) + '\n\n' + '\n\n'.join(recent_parts)
        
        return result
    
    def optimize_prompt(self, prompt, target_tokens):
        """优化提示词，减少token使用"""
        # 移除多余空格和换行
        optimized = ' '.join(prompt.split())
        
        # 简化常见短语
        replacements = {
            'Please help me': 'Help',
            'I would like to': 'I want to',
            'Could you please': 'Please',
            'Thank you very much': 'Thanks'
        }
        
        for old, new in replacements.items():
            optimized = optimized.replace(old, new)
        
        # 如果仍然超限，使用更激进的策略
        if self.count_tokens(optimized) > target_tokens:
            sentences = optimized.split('. ')
            # 保留前几句和后几句
            if len(sentences) > 4:
                keep_sentences = sentences[:2] + sentences[-2:]
                optimized = '. '.join(keep_sentences)
        
        return optimized

import asyncio
import aioredis
from typing import Optional

class AgentRequestQueue:
    def __init__(self, redis_url: str, max_concurrent: int = 5):
        self.redis_url = redis_url
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.redis_pool = None
    
    async def init_redis(self):
        self.redis_pool = await aioredis.create_redis_pool(self.redis_url)
    
    async def process_request(self, user_id: str, task: str, priority: int = 0):
        """处理用户请求，支持优先级"""
        request_id = f"{user_id}_{int(time.time())}"
        
        # 添加到队列
        await self._enqueue_request(request_id, {
            'user_id': user_id,
            'task': task,
            'priority': priority,
            'timestamp': time.time()
        })
        
        # 等待处理
        return await self._wait_for_result(request_id)
    
    async def _enqueue_request(self, request_id: str, request_data: dict):
        # 使用Redis有序集合实现优先级队列
        score = -request_data['priority']  # 负数实现高优先级在前
        await self.redis_pool.zadd('agent_queue', score, request_id)
        await self.redis_pool.hset(f'request_{request_id}', mapping=request_data)
    
    async def _wait_for_result(self, request_id: str, timeout: int = 300):
        """等待处理结果"""
        for _ in range(timeout):
            result = await self.redis_pool.get(f'result_{request_id}')
            if result:
                await self._cleanup_request(request_id)
                return json.loads(result)
            await asyncio.sleep(1)
        
        raise TimeoutError(f"Request {request_id} timeout")
    
    async def _cleanup_request(self, request_id: str):
        """清理请求相关数据"""
        await self.redis_pool.delete(f'request_{request_id}')
        await self.redis_pool.delete(f'result_{request_id}')
        await self.redis_pool.zrem('agent_queue', request_id)

class AgentWorker:
    def __init__(self, agent, queue_manager):
        self.agent = agent
        self.queue_manager = queue_manager
        self.running = False
    
    async def start(self):
        """启动工作进程"""
        self.running = True
        await self.queue_manager.init_redis()
        
        while self.running:
            try:
                # 获取下一个请求
                request_id = await self._get_next_request()
                
                if request_id:
                    async with self.queue_manager.semaphore:
                        await self._process_request(request_id)
                else:
                    await asyncio.sleep(1)  # 没有请求时短暂等待
                    
            except Exception as e:
                logging.error(f"Worker error: {e}")
                await asyncio.sleep(5)
    
    async def _get_next_request(self) -> Optional[str]:
        """从队列获取下一个请求"""
        result = await self.queue_manager.redis_pool.zpopmin('agent_queue')
        return result[0][0].decode() if result else None
    
    async def _process_request(self, request_id: str):
        """处理单个请求"""
        # 获取请求数据
        request_data = await self.queue_manager.redis_pool.hgetall(f'request_{request_id}')
        
        if not request_data:
            return
        
        user_id = request_data[b'user_id'].decode()
        task = request_data[b'task'].decode()
        
        try:
            # 执行Agent任务
            result = await self.agent.solve_task(task)
            
            # 保存结果
            await self.queue_manager.redis_pool.set(
                f'result_{request_id}',
                json.dumps({
                    'success': True,
                    'result': result,
                    'processed_at': time.time()
                }),
                expire=3600  # 1小时过期
            )
            
        except Exception as e:
            # 保存错误结果
            await self.queue_manager.redis_pool.set(
                f'result_{request_id}',
                json.dumps({
                    'success': False,
                    'error': str(e),
                    'processed_at': time.time()
                }),
                expire=3600
            )

class CustomerServiceAgent:
    def __init__(self):
        self.knowledge_base = ProductionRAGSystem()
        self.llm = ChatOpenAI(model="gpt-4")
        self.conversation_memory = {}
        
        # 预定义的工作流程
        self.workflows = {
            'order_inquiry': self._handle_order_inquiry,
            'product_question': self._handle_product_question,
            'complaint': self._handle_complaint,
            'general': self._handle_general_query
        }
    
    def classify_intent(self, user_input):
        """意图识别"""
        classification_prompt = f"""
        分析用户输入，判断属于以下哪种类型：
        1. order_inquiry - 订单查询相关
        2. product_question - 产品咨询
        3. complaint - 投诉建议   
        4. general - 一般咨询
        
        用户输入：{user_input}
        
        返回分类结果（只返回类型名称）：
        """
        
        intent = self.llm.predict(classification_prompt).strip()
        return intent if intent in self.workflows else 'general'
    
    def _handle_order_inquiry(self, user_input, context):
        """处理订单查询"""
        # 提取订单号
        order_extraction_prompt = f"""
        从用户输入中提取订单号：{user_input}
        
        如果没有订单号，返回"NEED_ORDER_NUMBER"
        如果有订单号，返回订单号
        """
        
        order_number = self.llm.predict(order_extraction_prompt).strip()
        
        if order_number == "NEED_ORDER_NUMBER":
            return "请提供您的订单号，我来帮您查询订单状态。"
        
        # 调用订单查询API
        order_info = self._query_order_api(order_number)
        
        if order_info:
            return f"""
            您的订单信息如下：
            订单号：{order_info['order_id']}
            状态：{order_info['status']}
            预计送达：{order_info.get('estimated_delivery', '待确定')}
            
            还有其他需要帮助的吗？
            """
        else:
            return "抱歉，没有找到对应的订单信息。请确认订单号是否正确。"
    
    def _query_order_api(self, order_number):
        """模拟订单API调用"""
        # 实际实现中这里会调用真实的订单系统API
        mock_orders = {
            "12345": {
                "order_id": "12345",
                "status": "已发货",
                "estimated_delivery": "2024-01-15"
            }
        }
        return mock_orders.get(order_number)
    
    def process_customer_request(self, user_id, user_input):
        """处理客服请求"""
        # 获取会话历史
        context = self.conversation_memory.get(user_id, [])
        
        # 意图识别
        intent = self.classify_intent(user_input)
        
        # 根据意图调用对应处理流程
        handler = self.workflows[intent]
        response = handler(user_input, context)
        
        # 更新会话历史
        context.append({'user': user_input, 'assistant': response})
        self.conversation_memory[user_id] = context[-10:]  # 保留最近10轮对话
        
        return response

import subprocess
import tempfile
import os

class CodeAssistantAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4")
        self.supported_languages = ['python', 'javascript', 'bash', 'sql']
        self.security_checker = CodeSecurityChecker()
    
    def analyze_code_request(self, user_input):
        """分析代码请求类型"""
        analysis_prompt = f"""
        分析用户的代码请求，返回JSON格式：
        {{
            "task_type": "write|debug|explain|optimize|review",
            "language": "python|javascript|bash|sql|other",
            "complexity": "simple|medium|complex",
            "requires_execution": true|false
        }}
        
        用户请求：{user_input}
        """
        
        try:
            response = self.llm.predict(analysis_prompt)
            return json.loads(response)
        except:
            return {
                "task_type": "write",
                "language": "python", 
                "complexity": "medium",
                "requires_execution": False
            }
    
    def generate_code(self, requirements, language="python"):
        """生成代码"""
        code_prompt = f"""
        基于以下需求，生成{language}代码：
        
        需求：{requirements}
        
        要求：
        1. 代码要完整可运行
        2. 添加适当的注释
        3. 包含错误处理
        4. 遵循最佳实践
        
        请只返回代码，不要额外解释：
        """
        
        code = self.llm.predict(code_prompt)
        
        # 清理代码格式
        code = self._clean_code_response(code)
        
        # 安全检查
        if not self.security_checker.is_safe(code, language):
            return "代码包含潜在安全风险，请修改需求后重试。"
        
        return code
    
    def execute_code(self, code, language="python"):
        """安全执行代码"""
        if language not in self.supported_languages:
            return "不支持的编程语言"
        
        # 安全检查
        if not self.security_checker.is_safe(code, language):
            return "代码包含不安全操作，无法执行"
        
        try:
            if language == "python":
                return self._execute_python(code)
            elif language == "bash":
                return self._execute_bash(code)
            # ... 其他语言
        except Exception as e:
            return f"执行错误：{str(e)}"
    
    def _execute_python(self, code):
        """在沙箱环境中执行Python代码"""
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(code)
            temp_file = f.name
        
        try:
            # 使用subprocess执行，限制权限
            result = subprocess.run(
                ['python', temp_file],
                capture_output=True,
                text=True,
                timeout=30,  # 30秒超时
                cwd=tempfile.gettempdir()  # 限制执行目录
            )
            
            if result.returncode == 0:
                return f"执行成功\n输出：\n{result.stdout}"
            else:
                return f"执行失败\n错误：\n{result.stderr}"
                
        finally:
            os.unlink(temp_file)
    
    def _clean_code_response(self, response):
        """清理LLM返回的代码响应"""
        # 移除代码块标记
        if '```' in response:
            parts = response.split('```')
            if len(parts) >= 2:
                code = parts[1]
                # 移除语言标记
                lines = code.split('\n')
                if lines[0].strip() in self.supported_languages:
                    lines = lines[1:]
                return '\n'.join(lines).strip()
        
        return response.strip()

class CodeSecurityChecker:
    def __init__(self):
        self.dangerous_patterns = {
            'python': [
                'import os', 'import sys', 'import subprocess',
                'exec(', 'eval(', 'open(', 'file(',
                '__import__', 'input(', 'raw_input('
            ],
            'bash': [
                'rm -rf', 'sudo', 'chmod', 'chown',
                '>', '>>', 'curl', 'wget', 'nc '
            ]
        }
    
    def is_safe(self, code, language):
        """检查代码是否安全"""
        if language not in self.dangerous_patterns:
            return True
        
        dangerous = self.dangerous_patterns[language]
        code_lower = code.lower()
        
        for pattern in dangerous:
            if pattern.lower() in code_lower:
                return False
        
        return True

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import io
import base64

class DataAnalysisAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4")
        self.current_dataframes = {}  # 存储当前会话的数据框
    
    def analyze_data(self, user_query, data_source=None):
        """分析数据并生成报告"""
        # 加载数据
        if data_source:
            df = self._load_data(data_source)
            df_name = f"df_{len(self.current_dataframes)}"
            self.current_dataframes[df_name] = df
        
        # 分析查询意图
        analysis_plan = self._generate_analysis_plan(user_query)
        
        # 执行分析
        results = []
        for step in analysis_plan['steps']:
            try:
                result = self._execute_analysis_step(step)
                results.append(result)
            except Exception as e:
                results.append(f"执行步骤 '{step}' 时出错：{str(e)}")
        
        # 生成最终报告
        report = self._generate_report(user_query, results)
        
        return report
    
    def _generate_analysis_plan(self, user_query):
        """生成数据分析计划"""
        available_data = list(self.current_dataframes.keys())
        
        plan_prompt = f"""
        基于用户查询生成数据分析计划：
        
        用户查询：{user_query}
        可用数据：{available_data}
        
        返回JSON格式的分析计划：
        {{
            "steps": [
                "描述数据基本信息",
                "执行具体分析",
                "生成可视化图表",
                "总结分析结果"
            ],
            "required_libraries": ["pandas", "matplotlib"],
            "analysis_type": "descriptive|exploratory|predictive"
        }}
        """
        
        try:
            plan_response = self.llm.predict(plan_prompt)
            return json.loads(plan_response)
        except:
            return {
                "steps": ["基础数据分析", "生成统计摘要"],
                "analysis_type": "descriptive"
            }
    
    def _execute_analysis_step(self, step):
        """执行分析步骤"""
        if not self.current_dataframes:
            return "没有可用的数据进行分析"
        
        # 获取主要数据框
        main_df_name = list(self.current_dataframes.keys())[0]
        df = self.current_dataframes[main_df_name]
        
        if "基本信息" in step:
            return self._get_basic_info(df)
        elif "统计摘要" in step:
            return self._get_statistical_summary(df)
        elif "可视化" in step:
            return self._generate_visualizations(df)
        elif "相关性分析" in step:
            return self._analyze_correlations(df)
        else:
            return f"未知分析步骤：{step}"
    
    def _get_basic_info(self, df):
        """获取数据基本信息"""
        info = {
            "行数": len(df),
            "列数": len(df.columns),
            "列名": list(df.columns),
            "数据类型": df.dtypes.to_dict(),
            "缺失值": df.isnull().sum().to_dict()
        }
        
        return f"""
        数据基本信息：
        - 数据形状：{info['行数']} 行 × {info['列数']} 列
        - 列名：{', '.join(info['列名'])}  
        - 缺失值统计：{dict(filter(lambda x: x[1] > 0, info['缺失值'].items()))}
        """
    
    def _get_statistical_summary(self, df):
        """生成统计摘要"""
        numeric_columns = df.select_dtypes(include=['number']).columns
        
        if len(numeric_columns) == 0:
            return "数据中没有数值型列可以进行统计分析"
        
        summary = df[numeric_columns].describe()
        
        # 格式化输出
        summary_text = "数值列统计摘要：\n"
        for col in summary.columns:
            summary_text += f"\n{col}:\n"
            summary_text += f"  均值: {summary.loc['mean', col]:.2f}\n"
            summary_text += f"  中位数: {summary.loc['50%', col]:.2f}\n"
            summary_text += f"  标准差: {summary.loc['std', col]:.2f}\n"
        
        return summary_text
    
    def _generate_visualizations(self, df):
        """生成可视化图表"""
        numeric_columns = df.select_dtypes(include=['number']).columns
        
        if len(numeric_columns) == 0:
            return "没有数值数据可供可视化"
        
        # 生成分布图
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))
        fig.suptitle('数据分布分析')
        
        for i, col in enumerate(numeric_columns[:4]):  # 最多显示4个列
            row, col_idx = divmod(i, 2)
            
            df[col].hist(ax=axes[row, col_idx], bins=20)
            axes[row, col_idx].set_title(f'{col} 分布')
            axes[row, col_idx].set_xlabel(col)
            axes[row, col_idx].set_ylabel('频次')
        
        # 隐藏空的子图
        for i in range(len(numeric_columns), 4):
            row, col_idx = divmod(i, 2)
            axes[row, col_idx].set_visible(False)
        
        plt.tight_layout()
        
        # 将图表转换为base64字符串
        buffer = io.BytesIO()
        plt.savefig(buffer, format='png')
        buffer.seek(0)
        image_base64 = base64.b64encode(buffer.getvalue()).decode()
        plt.close()
        
        return f"已生成数据分布图表（base64编码）：\n[图表数据: {image_base64[:50]}...]"
    
    def _generate_report(self, user_query, analysis_results):
        """生成最终分析报告"""
        report_prompt = f"""
        基于以下分析结果，生成一份专业的数据分析报告：
        
        用户查询：{user_query}
        
        分析结果：
        {chr(10).join(analysis_results)}
        
        请生成一份结构清晰的分析报告，包含：
        1. 数据概况
        2. 主要发现
        3. 业务建议
        4. 局限性说明
        """
        
        report = self.llm.predict(report_prompt)
        return report

class AgentDebugger:
    def __init__(self, agent):
        self.agent = agent
        self.debug_logs = []
        self.performance_metrics = {}
    
    def debug_execution(self, task, verbose=True):
        """调试Agent执行过程"""
        self.debug_logs.clear()
        
        try:
            # 记录开始时间
            start_time = time.time()
            
            # 执行任务并记录每个步骤
            result = self._execute_with_logging(task)
            
            # 记录性能指标
            execution_time = time.time() - start_time
            self.performance_metrics = {
                'execution_time': execution_time,
                'total_tokens': self._count_total_tokens(),
                'tool_calls': self._count_tool_calls(),
                'error_count': self._count_errors()
            }
            
            if verbose:
                self._print_debug_report()
            
            return result
            
        except Exception as e:
            self.debug_logs.append({
                'type': 'ERROR',
                'message': str(e),
                'timestamp': time.time()
            })
            
            if verbose:
                self._print_error_analysis()
            
            raise e
    
    def _execute_with_logging(self, task):
        """执行任务并记录日志"""
        self._log('TASK_START', f"开始执行任务: {task}")
        
        # 这里需要根据实际Agent实现来记录执行步骤
        # 示例：
        for step in self.agent.solve_task_steps(task):
            self._log('STEP', f"执行步骤: {step}")
            
        result = self.agent.solve_task(task)
        self._log('TASK_END', f"任务完成: {result}")
        
        return result
    
    def _log(self, log_type, message):
        """记录调试日志"""
        self.debug_logs.append({
            'type': log_type,
            'message': message,
            'timestamp': time.time()
        })
    
    def _print_debug_report(self):
        """打印调试报告"""
        print("=== Agent执行调试报告 ===")
        print(f"执行时间: {self.performance_metrics['execution_time']:.2f}秒")
        print(f"Token使用: {self.performance_metrics['total_tokens']}")
        print(f"工具调用次数: {self.performance_metrics['tool_calls']}")
        print(f"错误次数: {self.performance_metrics['error_count']}")
        
        print("\n=== 执行日志 ===")
        for i, log in enumerate(self.debug_logs):
            timestamp = time.strftime('%H:%M:%S', time.localtime(log['timestamp']))
            print(f"[{timestamp}] {log['type']}: {log['message']}")
    
    def analyze_performance_bottlenecks(self):
        """分析性能瓶颈"""
        bottlenecks = []
        
        if self.performance_metrics['execution_time'] > 30:
            bottlenecks.append("执行时间过长，考虑优化推理链或并行执行")
        
        if self.performance_metrics['total_tokens'] > 8000:
            bottlenecks.append("Token使用量过高，考虑优化提示词或截断上下文")
        
        if self.performance_metrics['tool_calls'] > 10:
            bottlenecks.append("工具调用次数过多，检查是否存在循环调用")
        
        return bottlenecks