# 傳統(tǒng)LLM應(yīng)用
defprocess_query(user_input):
    response = llm.generate(user_input)
    return response

# Agent應(yīng)用
classAgentSession:
    def__init__(self):
        self.conversation_history = []
        self.task_state = {}
        self.available_tools = []
        self.execution_plan = []
    
    defprocess_task(self, user_goal):
        # 維護會話狀態(tài)，支持多輪交互
        pass

# 傳統(tǒng)方式：硬編碼工具調(diào)用
if "天氣" in user_input:
    result = weather_api.get_weather()

# Agent方式：智能工具選擇
def select_tools(task_context, available_tools):
    # 基于任務(wù)上下文動態(tài)選擇工具
    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)  # 規(guī)劃行動
        observation = self.execute_action(action)  # 執(zhí)行并觀察
        
        if self.is_task_complete(observation):
            return self.generate_final_response(observation)
        else:
            # 繼續(xù)下一輪推理
            return self.solve_task(self.update_context(task, observation))

# 單模型架構(gòu)：簡單但可能存在能力瓶頸
classSingleModelAgent:
    def__init__(self, model_name="gpt-4"):
        self.llm = load_model(model_name)
    
    defprocess(self, task):
        returnself.llm.generate(task)

# 多模型架構(gòu)：復雜但性能更優(yōu)
classMultiModelAgent:
    def__init__(self):
        self.planner_model = load_model("gpt-4")  # 規(guī)劃能力強
        self.executor_model = load_model("claude-3.5")  # 代碼執(zhí)行好
        self.critic_model = load_model("gemini-pro")  # 結(jié)果評估
    
    defprocess(self, task):
        plan = self.planner_model.generate_plan(task)
        result = self.executor_model.execute(plan)
        evaluation = self.critic_model.evaluate(result)
        returnself.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 = []
    
    defadd_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數(shù)量和重要性截斷歷史記錄
        whileself._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, 預(yù)報天數(shù)，默認1天"
        }
    
    defexecute(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)}

classExtensionManager:
    def__init__(self):
        self.extensions = {}
    
    defregister(self, name, extension):
        self.extensions[name] = extension
    
    defget_tool_descriptions(self):
        descriptions = []
        for name, ext inself.extensions.items():
            descriptions.append({
                'name': name,
                'description': ext.description,
                'parameters': ext.parameters
            })
        return descriptions
    
    defexecute_tool(self, tool_name, **kwargs):
        if tool_name inself.extensions:
            returnself.extensions[tool_name].execute(**kwargs)
        else:
            return {"error": f"Tool {tool_name} not found"}

class DatabaseFunction:
    @staticmethod
    defget_schema():
        return {
            "name": "query_database",
            "description": "查詢數(shù)據(jù)庫并返回結(jié)果",
            "parameters": {
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SQL查詢語句"},
                    "limit": {"type": "integer", "description": "返回結(jié)果數(shù)量限制"}
                },
                "required": ["sql"]
            }
        }
    
    @staticmethod
    defexecute(sql, limit=100):
        # 客戶端執(zhí)行，可以進行安全檢查
        ifnot DatabaseFunction.is_safe_query(sql):
            return {"error": "Unsafe query detected"}
        
        # 執(zhí)行查詢邏輯
        # ...
        pass
    
    @staticmethod
    defis_safe_query(sql):
        # 實現(xiàn)SQL安全檢查邏輯
        dangerous_keywords = ['DROP', 'DELETE', 'UPDATE', 'INSERT']
        returnnotany(keyword in sql.upper() for keyword in dangerous_keywords)

# Agent調(diào)用示例
classFunctionCallingAgent:
    def__init__(self):
        self.available_functions = [DatabaseFunction.get_schema()]
    
    defprocess_query(self, user_input):
        # Agent生成函數(shù)調(diào)用
        function_call = self.llm.generate_function_call(
            user_input, 
            self.available_functions
        )
        
        if function_call['name'] == 'query_database':
            # 返回函數(shù)調(diào)用給客戶端執(zhí)行
            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

classProductionRAGSystem:
    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
    
    defadd_documents(self, documents):
        """添加文檔到向量數(shù)據(jù)庫"""
        chunks = []
        for doc in documents:
            doc_chunks = self._chunk_document(doc)
            chunks.extend(doc_chunks)
        
        # 生成embedding
        embeddings = self.embedding_model.encode(chunks)
        
        # 構(gòu)建FAISS索引
        ifself.vector_index isNone:
            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 inrange(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
    
    defsearch(self, query, top_k=5):
        """檢索相關(guān)文檔"""
        ifself.vector_index isNone:
            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 inzip(scores[0], indices[0]):
            if idx < len(self.document_store):
                results.append({
                    'content': self.document_store[idx],
                    'score': float(score)
                })
        
        return results
    
    defgenerate_answer(self, query, context_docs):
        """基于檢索結(jié)果生成回答"""
        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:
        """
        
        returnself.llm.generate(prompt)

# 使用示例
classRAGAgent:
    def__init__(self):
        self.rag_system = ProductionRAGSystem()
        self.llm = load_model("gpt-4")
    
    defadd_knowledge(self, documents):
        self.rag_system.add_documents(documents)
    
    defanswer_query(self, query):
        # 檢索相關(guān)文檔
        relevant_docs = self.rag_system.search(query, top_k=3)
        
        # 生成回答
        if relevant_docs:
            returnself.rag_system.generate_answer(query, relevant_docs)
        else:
            return "抱歉，我沒有找到相關(guān)信息來回答您的問題。"

class ReActAgent:
    def__init__(self, llm, tools):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        self.max_iterations = 10
    
    defsolve_task(self, task):
        context = f"Task: {task}\n\n"
        
        for iteration inrange(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()
            
            # 執(zhí)行工具調(diào)用
            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"任務(wù)執(zhí)行超時，請簡化任務(wù)或檢查工具配置。"
    
    def_generate_thought(self, context):
        prompt = f"""
        {context}
        
        分析當前情況，思考下一步應(yīng)該怎么做。
        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 inself.tools.items()
        ])
        
        prompt = f"""
        {context}
        
        可用工具：
        {tool_descriptions}
        
        基于以上思考，選擇下一步行動。格式：
        - 使用工具：工具名稱[輸入?yún)?shù)]
        - 結(jié)束任務(wù)：Final Answer: [最終答案]
        
        Action:"""
        
        response = self.llm.generate(prompt, max_tokens=100)
        return response.strip()
    
    def_parse_action(self, action):
        # 解析 "工具名稱[輸入?yún)?shù)]" 格式
        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
    
    defsolve_task(self, task):
        try:
            returnself._solve_with_recovery(task)
        except Exception as e:
            returnf"任務(wù)執(zhí)行失敗: {str(e)}"
    
    def_solve_with_recovery(self, task):
        context = f"Task: {task}\n\n"
        errors = []
        
        for iteration inrange(10):
            try:
                # 正常執(zhí)行邏輯
                result = self._execute_iteration(context)
                if result.get('finished'):
                    return result['answer']
                context = result['updated_context']
                
            except Exception as e:
                errors.append(str(e))
                
                # 錯誤恢復策略
                iflen(errors) <= self.error_recovery_attempts:
                    recovery_prompt = f"""
                    執(zhí)行過程中遇到錯誤：{str(e)}
                    
                    歷史錯誤：{'; '.join(errors)}
                    
                    請分析錯誤原因并調(diào)整策略。繼續(xù)執(zhí)行任務(wù)：{task}
                    """
                    context += f"Error Recovery: {recovery_prompt}\n"
                    continue
                else:
                    raise e
        
        return"任務(wù)執(zhí)行超時"
    
    def_execute_iteration(self, context):
        # 具體的執(zhí)行邏輯
        pass

import asyncio
from concurrent.futures import ThreadPoolExecutor

classOptimizedAgent:
    def__init__(self):
        self.llm_pool = ThreadPoolExecutor(max_workers=3)
        self.tool_cache = {}  # 工具調(diào)用結(jié)果緩存
    
    asyncdefparallel_tool_calls(self, tool_calls):
        """并行執(zhí)行多個工具調(diào)用"""
        tasks = []
        for tool_call in tool_calls:
            ifself._can_cache(tool_call):
                cache_key = self._get_cache_key(tool_call)
                if cache_key inself.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):
        # 判斷工具調(diào)用是否可以緩存
        cacheable_tools = ['weather', 'static_data_query']
        return tool_call['name'] in cacheable_tools
    
    asyncdef_execute_tool_async(self, tool_call):
        loop = asyncio.get_event_loop()
        returnawait 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
        }
    
    defprocess_with_budget(self, task):
        ifself._check_budget():
            returnself._process_task(task)
        else:
            return"任務(wù)超出預(yù)算限制，請簡化任務(wù)或增加預(yù)算。"
    
    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
    
    defsolve_task_with_validation(self, task):
        # 主Agent執(zhí)行任務(wù)
        primary_result = self.primary_agent.solve_task(task)
        
        # 驗證器檢查結(jié)果
        validation_prompt = f"""
        任務(wù)：{task}
        執(zhí)行結(jié)果：{primary_result}
        
        請評估這個結(jié)果是否：
        1. 正確回答了問題
        2. 邏輯清晰
        3. 沒有明顯錯誤
        
        如果有問題，請指出具體問題。
        評估結(jié)果：
        """
        
        validation = self.validator_llm.generate(validation_prompt)
        
        if"有問題"in validation or"錯誤"in validation:
            # 重新執(zhí)行或修正
            correction_prompt = f"""
            原始任務(wù)：{task}
            初始結(jié)果：{primary_result}
            發(fā)現(xiàn)的問題：{validation}
            
            請基于問題反饋，重新執(zhí)行任務(wù)或修正結(jié)果。
            """
            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
defcreate_production_agent():
    # 初始化模型
    llm = ChatOpenAI(model="gpt-4-1106-preview", temperature=0)
    
    # 定義工具
    search = DuckDuckGoSearchRun()
    tools = [
        Tool(
            name="Search",
            func=search.run,
            descriptinotallow="搜索最新信息"
        ),
        Tool(
            name="Calculator",
            func=lambda x: eval(x),  # 生產(chǎn)環(huán)境需要安全的計算器
            descriptinotallow="執(zhí)行數(shù)學計算"
        )
    ]
    
    # 創(chuàng)建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技術(shù)趨勢"})

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):
        # 根據(jù)配置初始化模型
        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 inself.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):
        # 初始化記憶系統(tǒng)
        ifself.config['memory']['type'] == 'redis':
            return RedisMemory(**self.config['memory']['params'])
        elifself.config['memory']['type'] == 'postgresql':
            return PostgreSQLMemory(**self.config['memory']['params'])
        else:
            return InMemoryMemory()
    
    def_init_monitoring(self):
        # 初始化監(jiān)控系統(tǒng)
        return AgentMonitor(
            metrics_backend=self.config['monitoring']['backend'],
            alert_thresholds=self.config['monitoring']['thresholds']
        )
    
    defprocess_request(self, user_id, task):
        withself.monitor.track_execution():
            try:
                # 加載用戶會話
                session = self.memory.get_session(user_id)
                
                # 執(zhí)行任務(wù)
                result = self._execute_task(task, session)
                
                # 保存會話狀態(tài)
                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 importDict, Any

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

classAgentMonitor:
    def__init__(self):
        self.metrics_store = []
        self.alert_thresholds = {
            'max_execution_time': 30.0,  # 秒
            'max_token_usage': 5000,
            'error_rate_threshold': 0.1# 10%
        }
    
    deftrack_execution(self):
        return ExecutionTracker(self)
    
    defanalyze_performance(self, time_window_hours=24):
        recent_metrics = self._get_recent_metrics(time_window_hours)
        
        ifnot recent_metrics:
            return"沒有足夠的數(shù)據(jù)進行分析"
        
        # 計算性能指標
        avg_execution_time = sum(m.end_time - m.start_time for m in recent_metrics) / len(recent_metrics)
        success_rate = sum(1for 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}小時):
        - 平均執(zhí)行時間: {avg_execution_time:.2f}秒
        - 成功率: {success_rate:.2%}
        - 平均Token使用量: {avg_token_usage:.0f}
        - 處理任務(wù)數(shù): {len(recent_metrics)}
        """
        
        # 檢查告警
        if avg_execution_time > self.alert_thresholds['max_execution_time']:
            report += f"\n?? 執(zhí)行時間超過閾值 ({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

classExecutionTracker:
    def__init__(self, monitor):
        self.monitor = monitor
        self.start_time = None
        self.metrics = None
    
    def__enter__(self):
        self.start_time = time.time()
        returnself
    
    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},  # 需要從實際執(zhí)行中獲取
            tool_calls=0,
            success=exc_type isNone,
            error_message=str(exc_val) if exc_val elseNone
        )
        
        self.monitor.metrics_store.append(self.metrics)

class RobustToolManager:
    def__init__(self, tools, retry_cnotallow=None):
        self.tools = tools
        self.retry_config = retry_config or {
            'max_retries': 3,
            'backoff_factor': 2,
            'timeout': 30
        }
    
    defexecute_tool(self, tool_name, **kwargs):
        tool = self.tools.get(tool_name)
        ifnot tool:
            return {"error": f"Tool '{tool_name}' not found"}
        
        for attempt inrange(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
        
        deftimeout_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

classTokenManager:
    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   # 總上下文限制
        }
    
    defcount_tokens(self, text):
        returnlen(self.encoding.encode(text))
    
    deftruncate_context(self, context, max_tokens):
        """智能截斷上下文，保留重要信息"""
        current_tokens = self.count_tokens(context)
        
        if current_tokens <= max_tokens:
            return context
        
        # 分離不同部分
        parts = context.split('\n\n')
        
        # 按重要性排序（系統(tǒng)提示 > 最近對話 > 歷史對話）
        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 notin system_parts and p notin recent_parts]
        
        # 重新組合
        result = '\n\n'.join(system_parts)
        result += '\n\n' + '\n\n'.join(recent_parts)
        
        # 如果還是超限，繼續(xù)截斷
        whileself.count_tokens(result) > max_tokens and other_parts:
            iflen(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
    
    defoptimize_prompt(self, prompt, target_tokens):
        """優(yōu)化提示詞，減少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)
        
        # 如果仍然超限，使用更激進的策略
        ifself.count_tokens(optimized) > target_tokens:
            sentences = optimized.split('. ')
            # 保留前幾句和后幾句
            iflen(sentences) > 4:
                keep_sentences = sentences[:2] + sentences[-2:]
                optimized = '. '.join(keep_sentences)
        
        return optimized

import asyncio
import aioredis
from typing importOptional

classAgentRequestQueue:
    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
    
    asyncdefinit_redis(self):
        self.redis_pool = await aioredis.create_redis_pool(self.redis_url)
    
    asyncdefprocess_request(self, user_id: str, task: str, priority: int = 0):
        """處理用戶請求，支持優(yōu)先級"""
        request_id = f"{user_id}_{int(time.time())}"
        
        # 添加到隊列
        awaitself._enqueue_request(request_id, {
            'user_id': user_id,
            'task': task,
            'priority': priority,
            'timestamp': time.time()
        })
        
        # 等待處理
        returnawaitself._wait_for_result(request_id)
    
    asyncdef_enqueue_request(self, request_id: str, request_data: dict):
        # 使用Redis有序集合實現(xiàn)優(yōu)先級隊列
        score = -request_data['priority']  # 負數(shù)實現(xiàn)高優(yōu)先級在前
        awaitself.redis_pool.zadd('agent_queue', score, request_id)
        awaitself.redis_pool.hset(f'request_{request_id}', mapping=request_data)
    
    asyncdef_wait_for_result(self, request_id: str, timeout: int = 300):
        """等待處理結(jié)果"""
        for _ inrange(timeout):
            result = awaitself.redis_pool.get(f'result_{request_id}')
            if result:
                awaitself._cleanup_request(request_id)
                return json.loads(result)
            await asyncio.sleep(1)
        
        raise TimeoutError(f"Request {request_id} timeout")
    
    asyncdef_cleanup_request(self, request_id: str):
        """清理請求相關(guān)數(shù)據(jù)"""
        awaitself.redis_pool.delete(f'request_{request_id}')
        awaitself.redis_pool.delete(f'result_{request_id}')
        awaitself.redis_pool.zrem('agent_queue', request_id)

classAgentWorker:
    def__init__(self, agent, queue_manager):
        self.agent = agent
        self.queue_manager = queue_manager
        self.running = False
    
    asyncdefstart(self):
        """啟動工作進程"""
        self.running = True
        awaitself.queue_manager.init_redis()
        
        whileself.running:
            try:
                # 獲取下一個請求
                request_id = awaitself._get_next_request()
                
                if request_id:
                    asyncwithself.queue_manager.semaphore:
                        awaitself._process_request(request_id)
                else:
                    await asyncio.sleep(1)  # 沒有請求時短暫等待
                    
            except Exception as e:
                logging.error(f"Worker error: {e}")
                await asyncio.sleep(5)
    
    asyncdef_get_next_request(self) -> Optional[str]:
        """從隊列獲取下一個請求"""
        result = awaitself.queue_manager.redis_pool.zpopmin('agent_queue')
        return result[0][0].decode() if result elseNone
    
    asyncdef_process_request(self, request_id: str):
        """處理單個請求"""
        # 獲取請求數(shù)據(jù)
        request_data = awaitself.queue_manager.redis_pool.hgetall(f'request_{request_id}')
        
        ifnot request_data:
            return
        
        user_id = request_data[b'user_id'].decode()
        task = request_data[b'task'].decode()
        
        try:
            # 執(zhí)行Agent任務(wù)
            result = awaitself.agent.solve_task(task)
            
            # 保存結(jié)果
            awaitself.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:
            # 保存錯誤結(jié)果
            awaitself.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 = {}
        
        # 預(yù)定義的工作流程
        self.workflows = {
            'order_inquiry': self._handle_order_inquiry,
            'product_question': self._handle_product_question,
            'complaint': self._handle_complaint,
            'general': self._handle_general_query
        }
    
    defclassify_intent(self, user_input):
        """意圖識別"""
        classification_prompt = f"""
        分析用戶輸入，判斷屬于以下哪種類型：
        1. order_inquiry - 訂單查詢相關(guān)
        2. product_question - 產(chǎn)品咨詢
        3. complaint - 投訴建議  
        4. general - 一般咨詢
        
        用戶輸入：{user_input}
        
        返回分類結(jié)果（只返回類型名稱）：
        """
        
        intent = self.llm.predict(classification_prompt).strip()
        return intent if intent inself.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"請?zhí)峁┠挠唵翁枺襾韼湍樵冇唵螤顟B(tài)。"
        
        # 調(diào)用訂單查詢API
        order_info = self._query_order_api(order_number)
        
        if order_info:
            returnf"""
            您的訂單信息如下：
            訂單號：{order_info['order_id']}
            狀態(tài)：{order_info['status']}
            預(yù)計送達：{order_info.get('estimated_delivery', '待確定')}
            
            還有其他需要幫助的嗎？
            """
        else:
            return"抱歉，沒有找到對應(yīng)的訂單信息。請確認訂單號是否正確。"
    
    def_query_order_api(self, order_number):
        """模擬訂單API調(diào)用"""
        # 實際實現(xiàn)中這里會調(diào)用真實的訂單系統(tǒng)API
        mock_orders = {
            "12345": {
                "order_id": "12345",
                "status": "已發(fā)貨",
                "estimated_delivery": "2024-01-15"
            }
        }
        return mock_orders.get(order_number)
    
    defprocess_customer_request(self, user_id, user_input):
        """處理客服請求"""
        # 獲取會話歷史
        context = self.conversation_memory.get(user_id, [])
        
        # 意圖識別
        intent = self.classify_intent(user_input)
        
        # 根據(jù)意圖調(diào)用對應(yīng)處理流程
        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

classCodeAssistantAgent:
    def__init__(self):
        self.llm = ChatOpenAI(model="gpt-4")
        self.supported_languages = ['python', 'javascript', 'bash', 'sql']
        self.security_checker = CodeSecurityChecker()
    
    defanalyze_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
            }
    
    defgenerate_code(self, requirements, language="python"):
        """生成代碼"""
        code_prompt = f"""
        基于以下需求，生成{language}代碼：
        
        需求：{requirements}
        
        要求：
        1. 代碼要完整可運行
        2. 添加適當?shù)淖⑨?        3. 包含錯誤處理
        4. 遵循最佳實踐
        
        請只返回代碼，不要額外解釋：
        """
        
        code = self.llm.predict(code_prompt)
        
        # 清理代碼格式
        code = self._clean_code_response(code)
        
        # 安全檢查
        ifnotself.security_checker.is_safe(code, language):
            return"代碼包含潛在安全風險，請修改需求后重試。"
        
        return code
    
    defexecute_code(self, code, language="python"):
        """安全執(zhí)行代碼"""
        if language notinself.supported_languages:
            return"不支持的編程語言"
        
        # 安全檢查
        ifnotself.security_checker.is_safe(code, language):
            return"代碼包含不安全操作，無法執(zhí)行"
        
        try:
            if language == "python":
                returnself._execute_python(code)
            elif language == "bash":
                returnself._execute_bash(code)
            # ... 其他語言
        except Exception as e:
            returnf"執(zhí)行錯誤：{str(e)}"
    
    def_execute_python(self, code):
        """在沙箱環(huán)境中執(zhí)行Python代碼"""
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(code)
            temp_file = f.name
        
        try:
            # 使用subprocess執(zhí)行，限制權(quán)限
            result = subprocess.run(
                ['python', temp_file],
                capture_output=True,
                text=True,
                timeout=30,  # 30秒超時
                cwd=tempfile.gettempdir()  # 限制執(zhí)行目錄
            )
            
            if result.returncode == 0:
                returnf"執(zhí)行成功\n輸出：\n{result.stdout}"
            else:
                returnf"執(zhí)行失敗\n錯誤：\n{result.stderr}"
                
        finally:
            os.unlink(temp_file)
    
    def_clean_code_response(self, response):
        """清理LLM返回的代碼響應(yīng)"""
        # 移除代碼塊標記
        if'```'in response:
            parts = response.split('```')
            iflen(parts) >= 2:
                code = parts[1]
                # 移除語言標記
                lines = code.split('\n')
                if lines[0].strip() inself.supported_languages:
                    lines = lines[1:]
                return'\n'.join(lines).strip()
        
        return response.strip()

classCodeSecurityChecker:
    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 '
            ]
        }
    
    defis_safe(self, code, language):
        """檢查代碼是否安全"""
        if language notinself.dangerous_patterns:
            returnTrue
        
        dangerous = self.dangerous_patterns[language]
        code_lower = code.lower()
        
        for pattern in dangerous:
            if pattern.lower() in code_lower:
                returnFalse
        
        return True

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

classDataAnalysisAgent:
    def__init__(self):
        self.llm = ChatOpenAI(model="gpt-4")
        self.current_dataframes = {}  # 存儲當前會話的數(shù)據(jù)框
    
    defanalyze_data(self, user_query, data_source=None):
        """分析數(shù)據(jù)并生成報告"""
        # 加載數(shù)據(jù)
        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)
        
        # 執(zhí)行分析
        results = []
        for step in analysis_plan['steps']:
            try:
                result = self._execute_analysis_step(step)
                results.append(result)
            except Exception as e:
                results.append(f"執(zhí)行步驟 '{step}' 時出錯：{str(e)}")
        
        # 生成最終報告
        report = self._generate_report(user_query, results)
        
        return report
    
    def_generate_analysis_plan(self, user_query):
        """生成數(shù)據(jù)分析計劃"""
        available_data = list(self.current_dataframes.keys())
        
        plan_prompt = f"""
        基于用戶查詢生成數(shù)據(jù)分析計劃：
        
        用戶查詢：{user_query}
        可用數(shù)據(jù)：{available_data}
        
        返回JSON格式的分析計劃：
        {{
            "steps": [
                "描述數(shù)據(jù)基本信息",
                "執(zhí)行具體分析",
                "生成可視化圖表",
                "總結(jié)分析結(jié)果"
            ],
            "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": ["基礎(chǔ)數(shù)據(jù)分析", "生成統(tǒng)計摘要"],
                "analysis_type": "descriptive"
            }
    
    def_execute_analysis_step(self, step):
        """執(zhí)行分析步驟"""
        ifnotself.current_dataframes:
            return"沒有可用的數(shù)據(jù)進行分析"
        
        # 獲取主要數(shù)據(jù)框
        main_df_name = list(self.current_dataframes.keys())[0]
        df = self.current_dataframes[main_df_name]
        
        if"基本信息"in step:
            returnself._get_basic_info(df)
        elif"統(tǒng)計摘要"in step:
            returnself._get_statistical_summary(df)
        elif"可視化"in step:
            returnself._generate_visualizations(df)
        elif"相關(guān)性分析"in step:
            returnself._analyze_correlations(df)
        else:
            returnf"未知分析步驟：{step}"
    
    def_get_basic_info(self, df):
        """獲取數(shù)據(jù)基本信息"""
        info = {
            "行數(shù)": len(df),
            "列數(shù)": len(df.columns),
            "列名": list(df.columns),
            "數(shù)據(jù)類型": df.dtypes.to_dict(),
            "缺失值": df.isnull().sum().to_dict()
        }
        
        returnf"""
        數(shù)據(jù)基本信息：
        - 數(shù)據(jù)形狀：{info['行數(shù)']} 行 × {info['列數(shù)']} 列
        - 列名：{', '.join(info['列名'])}
        - 缺失值統(tǒng)計：{dict(filter(lambda x: x[1] > 0, info['缺失值'].items()))}
        """
    
    def_get_statistical_summary(self, df):
        """生成統(tǒng)計摘要"""
        numeric_columns = df.select_dtypes(include=['number']).columns
        
        iflen(numeric_columns) == 0:
            return"數(shù)據(jù)中沒有數(shù)值型列可以進行統(tǒng)計分析"
        
        summary = df[numeric_columns].describe()
        
        # 格式化輸出
        summary_text = "數(shù)值列統(tǒng)計摘要：\n"
        for col in summary.columns:
            summary_text += f"\n{col}:\n"
            summary_text += f"  均值: {summary.loc['mean', col]:.2f}\n"
            summary_text += f"  中位數(shù): {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
        
        iflen(numeric_columns) == 0:
            return"沒有數(shù)值數(shù)據(jù)可供可視化"
        
        # 生成分布圖
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))
        fig.suptitle('數(shù)據(jù)分布分析')
        
        for i, col inenumerate(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 inrange(len(numeric_columns), 4):
            row, col_idx = divmod(i, 2)
            axes[row, col_idx].set_visible(False)
        
        plt.tight_layout()
        
        # 將圖表轉(zhuǎn)換為base64字符串
        buffer = io.BytesIO()
        plt.savefig(buffer, format='png')
        buffer.seek(0)
        image_base64 = base64.b64encode(buffer.getvalue()).decode()
        plt.close()
        
        returnf"已生成數(shù)據(jù)分布圖表（base64編碼）：\n[圖表數(shù)據(jù): {image_base64[:50]}...]"
    
    def_generate_report(self, user_query, analysis_results):
        """生成最終分析報告"""
        report_prompt = f"""
        基于以下分析結(jié)果，生成一份專業(yè)的數(shù)據(jù)分析報告：
        
        用戶查詢：{user_query}
        
        分析結(jié)果：
        {chr(10).join(analysis_results)}
        
        請生成一份結(jié)構(gòu)清晰的分析報告，包含：
        1. 數(shù)據(jù)概況
        2. 主要發(fā)現(xiàn)
        3. 業(yè)務(wù)建議
        4. 局限性說明
        """
        
        report = self.llm.predict(report_prompt)
        return report

class AgentDebugger:
    def__init__(self, agent):
        self.agent = agent
        self.debug_logs = []
        self.performance_metrics = {}
    
    defdebug_execution(self, task, verbose=True):
        """調(diào)試Agent執(zhí)行過程"""
        self.debug_logs.clear()
        
        try:
            # 記錄開始時間
            start_time = time.time()
            
            # 執(zhí)行任務(wù)并記錄每個步驟
            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):
        """執(zhí)行任務(wù)并記錄日志"""
        self._log('TASK_START', f"開始執(zhí)行任務(wù): {task}")
        
        # 這里需要根據(jù)實際Agent實現(xiàn)來記錄執(zhí)行步驟
        # 示例：
        for step inself.agent.solve_task_steps(task):
            self._log('STEP', f"執(zhí)行步驟: {step}")
            
        result = self.agent.solve_task(task)
        self._log('TASK_END', f"任務(wù)完成: {result}")
        
        return result
    
    def_log(self, log_type, message):
        """記錄調(diào)試日志"""
        self.debug_logs.append({
            'type': log_type,
            'message': message,
            'timestamp': time.time()
        })
    
    def_print_debug_report(self):
        """打印調(diào)試報告"""
        print("=== Agent執(zhí)行調(diào)試報告 ===")
        print(f"執(zhí)行時間: {self.performance_metrics['execution_time']:.2f}秒")
        print(f"Token使用: {self.performance_metrics['total_tokens']}")
        print(f"工具調(diào)用次數(shù): {self.performance_metrics['tool_calls']}")
        print(f"錯誤次數(shù): {self.performance_metrics['error_count']}")
        
        print("\n=== 執(zhí)行日志 ===")
        for i, log inenumerate(self.debug_logs):
            timestamp = time.strftime('%H:%M:%S', time.localtime(log['timestamp']))
            print(f"[{timestamp}] {log['type']}: {log['message']}")
    
    defanalyze_performance_bottlenecks(self):
        """分析性能瓶頸"""
        bottlenecks = []
        
        ifself.performance_metrics['execution_time'] > 30:
            bottlenecks.append("執(zhí)行時間過長，考慮優(yōu)化推理鏈或并行執(zhí)行")
        
        ifself.performance_metrics['total_tokens'] > 8000:
            bottlenecks.append("Token使用量過高，考慮優(yōu)化提示詞或截斷上下文")
        
        ifself.performance_metrics['tool_calls'] > 10:
            bottlenecks.append("工具調(diào)用次數(shù)過多，檢查是否存在循環(huán)調(diào)用")
        
        return bottlenecks