from jinja2 import Template

class DynamicPromptEngine:
    def __init__(self, base_template, context_analyzer):
        """
        初始化动态提示引擎
        :param base_template: 基础Jinja2模板
        :param context_analyzer: 上下文分析器，提取关键特征
        """
        self.template = Template(base_template)
        self.context_analyzer = context_analyzer
    
    def generate_prompt(self, user_input, conversation_history):
        """
        生成定制化提示词
        :param user_input: 当前用户输入
        :param conversation_history: 完整对话历史
        :return: 优化后的提示词
        """
        # 分析上下文特征
        context_features = self.context_analyzer.analyze(
            user_input, 
            conversation_history
        )
        
        # 动态注入关键元素
        prompt = self.template.render(
            user_input=user_input,
            task_type=context_features['task_type'],
            complexity=context_features['complexity'],
            domain=context_features['domain'],
            required_steps=context_features['required_steps'],
            risk_level=context_features['risk_level']
        )
        
        # 添加安全约束
        if context_features['risk_level'] > 0.7:
            prompt += "\n⚠️ 重要：此问题涉及高风险领域，必须引用最新官方指南，避免主观判断。"
        
        return prompt

# 上下文分析器示例
class ContextAnalyzer:
    def analyze(self, user_input, history):
        # 实际项目中使用轻量级分类器
        return {
            'task_type': 'troubleshooting' if '?' in user_input else 'information',
            'complexity': min(len(user_input.split())/50, 3),
            'domain': self._detect_domain(user_input),
            'required_steps': 2 if 'how' in user_input.lower() else 1,
            'risk_level': 0.9 if 'legal' in user_input.lower() else 0.3
        }
    
    def _detect_domain(self, text):
        domains = {
            'finance': ['account', 'payment', 'transaction'],
            'health': ['symptom', 'medication', 'doctor']
        }
        for domain, keywords in domains.items():
            if any(kw in text.lower() for kw in keywords):
                return domain
        return 'general'

import chromadb
from sentence_transformers import SentenceTransformer

class MemoryBank:
    def __init__(self, collection_name="user_memory"):
        """
        初始化记忆银行
        :param collection_name: ChromaDB集合名称
        """
        self.client = chromadb.PersistentClient(path="./memory_db")
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}
        )
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        self.memory_threshold = 0.75  # 相似度阈值
    
    def store_memory(self, key, value, metadata=None, importance=0.5):
        """
        存储记忆片段
        :param key: 记忆标识
        :param value: 记忆内容
        :param metadata: 附加元数据
        :param importance: 重要性评分(0-1)
        """
        embedding = self.encoder.encode([value]).tolist()[0]
        self.collection.add(
            embeddings=[embedding],
            documents=[value],
            metadatas=[{
                "key": key,
                "importance": importance,
                **(metadata or {})
            }],
            ids=[key]
        )
    
    def retrieve_context(self, query, n_results=5):
        """
        检索相关上下文
        :param query: 当前查询
        :param n_results: 返回结果数
        :return: 相关记忆列表
        """
        query_embedding = self.encoder.encode([query]).tolist()[0]
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results,
            include=["documents", "metadatas", "distances"]
        )
        
        # 过滤低相关度结果
        relevant = []
        for i in range(len(results['distances'][0])):
            distance = results['distances'][0][i]
            # 距离越小越相关，转换为相似度
            similarity = 1 - distance
            if similarity > self.memory_threshold:
                relevant.append({
                    "content": results['documents'][0][i],
                    "metadata": results['metadatas'][0][i],
                    "similarity": similarity
                })
        
        # 按重要性和相关度排序
        relevant.sort(key=lambda x: x['metadata'].get('importin', 0.5) * x['similarity'], reverse=True)
        return relevant
    
    def update_memory(self, key, new_value, update_importance=False):
        """更新现有记忆"""
        # 实际实现需处理ChromaDB的更新逻辑
        pass

# 使用示例
if __name__ == "__main__":
    memory = MemoryBank()
    
    # 存储用户偏好
    memory.store_memory(
        "user_pref_123", 
        "用户偏好使用简洁技术术语，避免营销话术",
        metadata={"user_id": "123", "category": "preference"},
        importance=0.9
    )
    
    # 存储项目关键决策
    memory.store_memory(
        "project_x_decision_001",
        "团队决定采用微服务架构，数据库使用PostgreSQL",
        metadata={"project": "x", "date": "2023-11-15"},
        importance=0.8
    )
    
    # 检索相关上下文
    context = memory.retrieve_context("关于数据库选型的讨论")
    print(f"检索到 {len(context)} 条相关记忆:")
    for item in context:
        print(f"- {item['content']} (相关度: {item['similarity']:.2f})")

import torch
from PIL import Image
import requests
from transformers import BlipProcessor, BlipForConditionalGeneration
from pydub import AudioSegment
import speech_recognition as sr

class MultimodalProcessor:
    def __init__(self):
        """初始化多模态处理组件"""
        # 图像处理
        self.image_processor = BlipProcessor.from_pretrained("Salesforce/blip-image-captioning-base")
        self.image_model = BlipForConditionalGeneration.from_pretrained("Salesforce/blip-image-captioning-base")
        
        # 音频处理
        self.recognizer = sr.Recognizer()
    
    def process_image(self, image_path, prompt="a photo of"):
        """
        处理图像并生成描述
        :param image_path: 图像路径或URL
        :param prompt: 引导提示
        :return: 图像描述文本
        """
        # 加载图像
        if image_path.startswith('http'):
            image = Image.open(requests.get(image_path, stream=True).raw)
        else:
            image = Image.open(image_path)
        
        # 生成描述
        inputs = self.image_processor(images=image, text=prompt, return_tensors="pt")
        outputs = self.image_model.generate(**inputs)
        description = self.image_processor.decode(outputs[0], skip_special_tokens=True)
        
        return description
    
    def process_audio(self, audio_path, language="en-US"):
        """
        处理音频并转录文本
        :param audio_path: 音频文件路径
        :param language: 语言代码
        :return: 转录文本
        """
        # 转换音频格式（如果需要）
        if audio_path.endswith('.m4a'):
            sound = AudioSegment.from_file(audio_path, format="m4a")
            wav_path = audio_path.replace('.m4a', '.wav')
            sound.export(wav_path, format="wav")
            audio_path = wav_path
        
        # 音频转录
        with sr.AudioFile(audio_path) as source:
            audio = self.recognizer.record(source)
            try:
                text = self.recognizer.recognize_google(audio, language=language)
                return text
            except sr.UnknownValueError:
                return "音频无法识别"
            except sr.RequestError as e:
                return f"API错误: {e}"
    
    def process_multimodal_input(self, inputs):
        """
        处理多模态输入组合
        :param inputs: 输入字典，格式 {"text": "...", "images": [...], "audio": "..."}
        :return: 结构化上下文
        """
        context = {"text": inputs.get("text", "")}
        
        # 处理图像
        if "images" in inputs:
            image_descriptions = []
            for img in inputs["images"]:
                desc = self.process_image(img)
                image_descriptions.append(f"Image: {desc}")
            context["images"] = image_descriptions
        
        # 处理音频
        if "audio" in inputs:
            transcript = self.process_audio(inputs["audio"])
            context["audio_transcript"] = transcript
        
        return context

# 使用示例
if __name__ == "__main__":
    processor = MultimodalProcessor()
    
    # 处理图像示例
    image_desc = processor.process_image("product.jpg", "detailed description of this product")
    print(f"图像描述: {image_desc}")
    
    # 处理多模态输入
    multimodal_context = processor.process_multimodal_input({
        "text": "请分析这个产品的设计特点",
        "images": ["product_front.jpg", "product_side.jpg"],
        "audio": "customer_feedback.m4a"
    })
    
    print("\n多模态上下文:")
    print(f"文本输入: {multimodal_context['text']}")
    print(f"图像分析: {multimodal_context['images']}")
    print(f"音频转录: {multimodal_context['audio_transcript']}")

import re
from openai import OpenAI

class SelfReflectionAgent:
    def __init__(self, base_model="gpt-4-turbo", max_iterations=3):
        """
        初始化自我反思代理
        :param base_model: 基础LLM模型
        :param max_iterations: 最大反思轮次
        """
        self.client = OpenAI()
        self.base_model = base_model
        self.max_iterations = max_iterations
        self.reflection_prompt = """
        作为专业{domain}专家，请严格审查以下响应：
        [原始响应]
        {response}
        
        请按步骤执行：
        1. 事实核查：指出任何可能的事实错误或过时信息
        2. 逻辑检查：识别推理漏洞或矛盾点
        3. 风险评估：标注潜在法律/安全风险
        4. 改进建议：提供具体修改方案
        
        用JSON格式输出，包含：
        - "fact_errors": 列表
        - "logic_issues": 列表
        - "risks": 列表
        - "improvements": 列表
        """
    
    def generate_with_reflection(self, prompt, domain="general", context=""):
        """
        生成带反思的响应
        :param prompt: 用户原始提示
        :param domain: 领域类型
        :param context: 附加上下文
        :return: 最终优化响应
        """
        current_response = self._generate_initial(prompt, context)
        iteration = 0
        
        while iteration < self.max_iterations:
            reflection = self._perform_reflection(current_response, domain)
            
            # 检查是否需要改进
            if not (reflection['fact_errors'] or 
                    reflection['logic_issues'] or 
                    reflection['risks']):
                break
            
            # 生成改进版
            current_response = self._apply_improvements(
                prompt, 
                current_response, 
                reflection,
                context
            )
            iteration += 1
        
        return {
            "final_response": current_response,
            "iterations": iteration + 1,
            "reflection_summary": self._summarize_reflection(reflection)
        }
    
    def _generate_initial(self, prompt, context):
        """生成初始响应"""
        full_prompt = f"{context}\n\n用户问题：{prompt}"
        response = self.client.chat.completions.create(
            model=self.base_model,
            messages=[{"role": "user", "content": full_prompt}]
        )
        return response.choices[0].message.content
    
    def _perform_reflection(self, response, domain):
        """执行自我反思"""
        reflection_prompt = self.reflection_prompt.format(
            domain=domain,
            response=response
        )
        
        result = self.client.chat.completions.create(
            model=self.base_model,
            response_format={"type": "json_object"},
            messages=[
                {"role": "system", "content": "你是一个严谨的审查专家"},
                {"role": "user", "content": reflection_prompt}
            ]
        )
        
        try:
            return eval(result.choices[0].message.content)
        except:
            return {
                "fact_errors": [],
                "logic_issues": [],
                "risks": [],
                "improvements": ["无法生成有效反思，请重新检查"]
            }
    
    def _apply_improvements(self, original_prompt, current_response, reflection, context):
        """应用改进建议"""
        improvement_prompt = f"""
        基于以下反思结果改进原始响应：
        
        原始问题：{original_prompt}
        当前响应：{current_response}
        
        反思结果：
        - 事实错误：{', '.join(reflection['fact_errors'])}
        - 逻辑问题：{', '.join(reflection['logic_issues'])}
        - 风险点：{', '.join(reflection['risks'])}
        - 改进建议：{', '.join(reflection['improvements'])}
        
        请生成修正后的专业响应，保留有效内容，仅修正问题部分。
        """
        
        if context:
            improvement_prompt = f"{context}\n\n{improvement_prompt}"
        
        response = self.client.chat.completions.create(
            model=self.base_model,
            messages=[{"role": "user", "content": improvement_prompt}]
        )
        return response.choices[0].message.content
    
    def _summarize_reflection(self, reflection):
        """生成反思摘要"""
        summary = []
        if reflection['fact_errors']:
            summary.append(f"修正了{len(reflection['fact_errors'])}个事实错误")
        if reflection['logic_issues']:
            summary.append(f"解决了{len(reflection['logic_issues'])}个逻辑问题")
        if reflection['risks']:
            summary.append(f"消除了{len(reflection['risks'])}个风险点")
        return "; ".join(summary) if summary else "无显著问题"

# 使用示例
if __name__ == "__main__":
    agent = SelfReflectionAgent()
    
    result = agent.generate_with_reflection(
        prompt="根据2023年新法规，员工离职补偿如何计算？",
        domain="labor_law",
        context="公司位于上海市，员工工作年限5年"
    )
    
    print("最终响应:")
    print(result['final_response'])
    print(f"\n迭代次数: {result['iterations']}")
    print(f"改进摘要: {result['reflection_summary']}")

import re
from transformers import pipeline

class SafetyGuard:
    def __init__(self):
        """初始化安全防护系统"""
        # 内容安全过滤器
        self.toxicity_checker = pipeline(
            "text-classification", 
            model="unitary/toxic-bert",
            truncation=True
        )
        
        # 事实核查组件
        self.fact_checker = pipeline(
            "question-answering",
            model="valhalla/t5-base-e2e-qg",
            truncation=True
        )
        
        # 偏见检测器
        self.bias_detector = pipeline(
            "text-classification",
            model="d4data/bias-model",
            truncation=True
        )
        
        # 安全阈值配置
        self.thresholds = {
            "toxicity": 0.7,
            "bias": 0.6,
            "fact_confidence": 0.8
        }
        
        # 可信知识源
        self.knowledge_sources = {
            "medical": ["CDC", "WHO", "NIH"],
            "legal": ["Gov.cn", "最高人民法院"]
        }
    
    def screen_response(self, response, domain="general", user_profile=None):
        """
        安全筛查响应
        :param response: 待检查的LLM响应
        :param domain: 领域类型
        :param user_profile: 用户画像（含敏感信息）
        :return: 安全评估结果
        """
        results = {
            "is_safe": True,
            "issues": [],
            "suggested_fixes": []
        }
        
        # 1. 毒性内容检查
        toxicity_result = self.toxicity_checker(response)[0]
        if toxicity_result['score'] > self.thresholds['toxicity']:
            results['issues'].append({
                "type": "toxicity",
                "severity": toxicity_result['score'],
                "description": f"检测到毒性内容 ({toxicity_result['label']})"
            })
            results['is_safe'] = False
        
        # 2. 偏见检测
        if user_profile and 'gender' in user_profile:
            bias_result = self.bias_detector(f"针对{user_profile['gender']}的评价: {response}")[0]
            if bias_result['score'] > self.thresholds['bias']:
                results['issues'].append({
                    "type": "bias",
                    "severity": bias_result['score'],
                    "description": f"检测到潜在偏见 ({bias_result['label']})"
                })
                results['is_safe'] = False
        
        # 3. 事实核查（关键声明）
        claims = self._extract_key_claims(response)
        for claim in claims:
            verification = self._verify_claim(claim, domain)
            if verification['confidence'] < self.thresholds['fact_confidence']:
                results['issues'].append({
                    "type": "fact_error",
                    "severity": 1 - verification['confidence'],
                    "description": f"声明 '{claim}' 事实存疑",
                    "suggested_source": verification['source']
                })
                results['suggested_fixes'].append(
                    f"请参考{verification['source']}核实'{claim}'"
                )
                results['is_safe'] = False
        
        return results
    
    def _extract_key_claims(self, text):
        """提取关键声明（简化版）"""
        # 实际应用需更复杂的NLP处理
        sentences = re.split(r'[.!?]', text)
        claims = []
        for s in sentences:
            s = s.strip()
            if len(s) > 10 and any(word in s.lower() for word in ['是', '应该', '必须', '根据']):
                claims.append(s)
        return claims[:3]  # 限制检查数量
    
    def _verify_claim(self, claim, domain):
        """验证声明真实性"""
        # 简化版实现，实际应用需连接知识库
        if "2023" in claim and "法规" in claim:
            return {
                "confidence": 0.9,
                "source": "国务院公报2023年第15期"
            }
        elif domain == "medical" and "阿司匹林" in claim:
            return {
                "confidence": 0.6,
                "source": "需结合患者用药史判断"
            }
        return {
            "confidence": 0.95,
            "source": self.knowledge_sources.get(domain, ["通用知识库"])[0]
        }

# 使用示例
if __name__ == "__main__":
    guard = SafetyGuard()
    
    # 测试有风险的响应
    risky_response = (
        "所有头痛都应该立即服用阿司匹林。"
        "根据最新研究，女性在技术岗位上表现不如男性。"
        "这个政策自2023年1月1日起实施。"
    )
    
    result = guard.screen_response(
        response=risky_response,
        domain="medical",
        user_profile={"gender": "female"}
    )
    
    print("安全评估结果:")
    print(f"是否安全: {'✅' if result['is_safe'] else '❌'}")
    print("\n检测到的问题:")
    for issue in result['issues']:
        print(f"- [{issue['type']}] {issue['description']} (严重度: {issue['severity']:.2f})")
    
    if result['suggested_fixes']:
        print("\n改进建议:")
        for fix in result['suggested_fixes']:
            print(f"⚠️ {fix}")