05 - 模型配置与优化¶

import requests
from typing import Dict, List, Optional

class OpenAIModelConfig:
    """OpenAI模型配置"""

    def __init__(self, api_key):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.base_url = "https://api.dify.ai/v1"

    def configure_gpt4(self, app_id: str, config: Dict = None) -> Dict:
        """
        配置GPT-4模型

        Args:
            app_id: 应用ID
            config: 配置参数
        """
        if config is None:
            config = {
                "temperature": 0.7,
                "top_p": 0.9,
                "max_tokens": 2048,
                "presence_penalty": 0,
                "frequency_penalty": 0
            }

        return self._configure_model(app_id, "openai", "gpt-4o", config)

    def configure_gpt4o_mini(self, app_id: str, config: Dict = None) -> Dict:
        """
        配置GPT-4o-mini模型

        Args:
            app_id: 应用ID
            config: 配置参数
        """
        if config is None:
            config = {
                "temperature": 0.7,
                "top_p": 0.9,
                "max_tokens": 2048,
                "presence_penalty": 0,
                "frequency_penalty": 0
            }

        return self._configure_model(app_id, "openai", "gpt-4o-mini", config)

    def _configure_model(self, app_id: str, provider: str,
                        model_name: str, config: Dict) -> Dict:
        """
        配置模型（概念示例）

        注意：Dify不提供通过API配置应用模型的接口
        模型配置需在Dify Web界面完成：
        应用设置 -> 模型配置 -> 选择提供商和模型 -> 调整参数
        """
        model_config = {
            "provider": provider,
            "model_name": model_name,
            "completion_params": {
                "temperature": config.get("temperature", 0.7),
                "top_p": config.get("top_p", 0.9),
                "max_tokens": config.get("max_tokens", 2048),
                "presence_penalty": config.get("presence_penalty", 0),
                "frequency_penalty": config.get("frequency_penalty", 0)
            }
        }
        print(f"请在Dify Web界面配置模型: {model_config}")
        return model_config

    def get_app_parameters(self, app_id: str) -> Dict:
        """获取应用参数信息（这是真实可用的API）"""
        url = f"{self.base_url}/parameters"

        try:  # try/except捕获异常
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}

# 使用示例
if __name__ == "__main__":
    config = OpenAIModelConfig(api_key="your_api_key_here")

    # 配置GPT-4
    result = config.configure_gpt4(
        app_id="app_id_here",
        config={
            "temperature": 0.5,
            "top_p": 0.95,
            "max_tokens": 4096,
            "presence_penalty": 0.1,
            "frequency_penalty": 0.1
        }
    )
    print(f"GPT-4配置结果: {result}")

    # 配置GPT-4o-mini
    result = config.configure_gpt4o_mini(
        app_id="app_id_here",
        config={
            "temperature": 0.8,
            "top_p": 0.9,
            "max_tokens": 2048
        }
    )
    print(f"GPT-4o-mini配置结果: {result}")

class AnthropicModelConfig:
    """Anthropic模型配置"""

    def __init__(self, api_key):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.base_url = "https://api.dify.ai/v1"

    def configure_claude_opus(self, app_id: str, config: Dict = None) -> Dict:
        """配置Claude Opus 4.6"""
        if config is None:
            config = {
                "temperature": 0.7,
                "top_p": 0.9,
                "max_tokens": 8192,
                "top_k": 0
            }

        return self._configure_model(app_id, "anthropic", "claude-opus-4-6-20260215", config)

    def configure_claude_sonnet(self, app_id: str, config: Dict = None) -> Dict:
        """配置Claude Sonnet 4.6"""
        if config is None:
            config = {
                "temperature": 0.7,
                "top_p": 0.9,
                "max_tokens": 8192,
                "top_k": 0
            }

        return self._configure_model(app_id, "anthropic", "claude-sonnet-4-6-20260215", config)

    def configure_claude_haiku(self, app_id: str, config: Dict = None) -> Dict:
        """配置Claude Haiku 4.6"""
        if config is None:
            config = {
                "temperature": 0.7,
                "top_p": 0.9,
                "max_tokens": 4096,
                "top_k": 0
            }

        return self._configure_model(app_id, "anthropic", "claude-haiku-4-6-20260215", config)

    def _configure_model(self, app_id: str, provider: str,
                        model_name: str, config: Dict) -> Dict:
        """
        配置Anthropic模型（概念示例）

        注意：模型配置需在Dify Web界面完成
        """
        model_config = {
            "provider": provider,
            "model_name": model_name,
            "completion_params": {
                "temperature": config.get("temperature", 0.7),
                "top_p": config.get("top_p", 0.9),
                "max_tokens": config.get("max_tokens", 4096),
                "top_k": config.get("top_k", 0)
            }
        }
        print(f"请在Dify Web界面配置模型: {model_config}")
        return model_config

# 使用示例
if __name__ == "__main__":
    config = AnthropicModelConfig(api_key="your_api_key_here")

    # 配置Claude Opus 4.6
    result = config.configure_claude_opus(
        app_id="app_id_here",
        config={
            "temperature": 0.5,
            "top_p": 0.95,
            "max_tokens": 8192
        }
    )
    print(f"Claude Opus 4.6配置结果: {result}")

class LocalModelConfig:
    """本地模型配置"""

    def __init__(self, api_key):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.base_url = "https://api.dify.ai/v1"

    def configure_llama3(self, app_id: str, endpoint: str,
                        config: Dict = None) -> Dict:
        """
        配置Llama 3模型

        Args:
            app_id: 应用ID
            endpoint: 模型服务端点
            config: 配置参数
        """
        if config is None:
            config = {
                "temperature": 0.7,
                "top_p": 0.9,
                "max_tokens": 2048
            }

        return self._configure_local_model(
            app_id,
            "ollama/llama3.1",
            endpoint,
            config
        )

    def configure_qwen(self, app_id: str, endpoint: str,
                     config: Dict = None) -> Dict:
        """配置Qwen模型"""
        if config is None:
            config = {
                "temperature": 0.7,
                "top_p": 0.9,
                "max_tokens": 2048
            }

        return self._configure_local_model(
            app_id,
            "ollama/qwen",
            endpoint,
            config
        )

    def _configure_local_model(self, app_id: str, model_name: str,
                             endpoint: str, config: Dict) -> Dict:
        """
        配置本地模型（概念示例）

        注意：本地模型配置需在Dify Web界面完成：
        设置 -> 模型提供商 -> 添加Ollama提供商 -> 填写服务地址
        Ollama在Dify中是独立的提供商，不是通过OpenAI提供商配置
        """
        model_config = {
            "provider": "ollama",  # Ollama是独立的提供商
            "model_name": model_name,  # 例如: llama2, qwen
            "base_url": endpoint,  # 例如: http://localhost:11434
            "completion_params": {
                "temperature": config.get("temperature", 0.7),
                "top_p": config.get("top_p", 0.9),
                "max_tokens": config.get("max_tokens", 2048)
            }
        }
        print(f"请在Dify Web界面配置本地模型: {model_config}")
        return model_config

# 使用示例
if __name__ == "__main__":
    config = LocalModelConfig(api_key="your_api_key_here")

    # 配置Llama 3
    result = config.configure_llama3(
        app_id="app_id_here",
        endpoint="http://localhost:11434/v1",
        config={
            "temperature": 0.7,
            "top_p": 0.9,
            "max_tokens": 2048
        }
    )
    print(f"Llama 3配置结果: {result}")

class ModelSelector:
    """智能模型选择器"""

    def __init__(self):
        self.model_capabilities = {
            "gpt-4o": {
                "reasoning": 10,
                "coding": 10,
                "creativity": 9,
                "speed": 5,
                "cost": 1
            },
            "gpt-4o-mini": {
                "reasoning": 7,
                "coding": 8,
                "creativity": 7,
                "speed": 9,
                "cost": 9
            },
            "claude-opus-4-6": {
                "reasoning": 10,
                "coding": 10,
                "creativity": 10,
                "speed": 6,
                "cost": 2
            },
            "claude-sonnet-4-6": {
                "reasoning": 9,
                "coding": 9,
                "creativity": 9,
                "speed": 8,
                "cost": 6
            },
            "claude-haiku-4-6": {
                "reasoning": 7,
                "coding": 7,
                "creativity": 7,
                "speed": 10,
                "cost": 9
            }
        }

    def select_model(self, task_type: str,
                   priority: str = "quality") -> str:
        """
        根据任务类型和优先级选择模型

        Args:
            task_type: 任务类型 (reasoning/coding/creativity/general)
            priority: 优先级 (quality/speed/cost)
        """
        # 根据优先级调整权重
        if priority == "quality":
            weights = {
                "reasoning": 0.4,
                "coding": 0.3,
                "creativity": 0.2,
                "speed": 0.05,
                "cost": 0.05
            }
        elif priority == "speed":
            weights = {
                "reasoning": 0.2,
                "coding": 0.2,
                "creativity": 0.2,
                "speed": 0.3,
                "cost": 0.1
            }
        else:  # cost
            weights = {
                "reasoning": 0.2,
                "coding": 0.2,
                "creativity": 0.2,
                "speed": 0.1,
                "cost": 0.3
            }

        # 计算每个模型的得分
        scores = {}
        for model, capabilities in self.model_capabilities.items():
            score = 0
            for capability, weight in weights.items():
                score += capabilities.get(capability, 0) * weight
            scores[model] = score

        # 根据任务类型调整
        if task_type == "reasoning":
            scores = {k: v * 1.2 for k, v in scores.items()}
        elif task_type == "coding":
            scores = {k: v * 1.1 for k, v in scores.items()}
        elif task_type == "creativity":
            scores = {k: v * 1.1 for k, v in scores.items()}

        # 返回得分最高的模型
        return max(scores, key=scores.get)

    def get_model_recommendation(self, task_description: str,
                                constraints: Dict = None) -> Dict:
        """
        获取模型推荐

        Args:
            task_description: 任务描述
            constraints: 约束条件 (max_cost, min_speed, etc.)
        """
        # 分析任务类型
        task_type = self._analyze_task(task_description)

        # 确定优先级
        priority = constraints.get("priority", "quality") if constraints else "quality"

        # 选择模型
        model = self.select_model(task_type, priority)

        # 获取模型配置
        model_config = self.model_capabilities[model]

        return {
            "recommended_model": model,
            "task_type": task_type,
            "priority": priority,
            "model_capabilities": model_config,
            "reason": f"基于{task_type}任务和{priority}优先级选择"
        }

    def _analyze_task(self, task_description: str) -> str:
        """分析任务类型"""
        task_lower = task_description.lower()

        if any(keyword in task_lower for keyword in
               ["推理", "reasoning", "逻辑", "logic", "分析", "analysis"]):
            return "reasoning"
        elif any(keyword in task_lower for keyword in
                 ["代码", "code", "编程", "programming", "开发", "development"]):
            return "coding"
        elif any(keyword in task_lower for keyword in
                 ["创意", "creative", "写作", "writing", "生成", "generation"]):
            return "creativity"
        else:
            return "general"

# 使用示例
if __name__ == "__main__":
    selector = ModelSelector()

    # 示例1：复杂推理任务
    recommendation = selector.get_model_recommendation(
        task_description="需要解决复杂的数学推理问题",
        constraints={"priority": "quality"}
    )
    print(f"推荐模型: {recommendation['recommended_model']}")
    print(f"原因: {recommendation['reason']}")

    # 示例2：代码生成任务
    recommendation = selector.get_model_recommendation(
        task_description="生成Python代码实现快速排序",
        constraints={"priority": "speed"}
    )
    print(f"\n推荐模型: {recommendation['recommended_model']}")
    print(f"原因: {recommendation['reason']}")

    # 示例3：创意写作任务
    recommendation = selector.get_model_recommendation(
        task_description="写一个科幻小说开头",
        constraints={"priority": "quality"}
    )
    print(f"\n推荐模型: {recommendation['recommended_model']}")
    print(f"原因: {recommendation['reason']}")

class TemperatureOptimizer:
    """温度参数优化器"""

    def __init__(self, api_key, app_id):
        self.api_key = api_key
        self.app_id = app_id
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.base_url = "https://api.dify.ai/v1"

    def test_temperature(self, prompt: str, temperatures: List[float]) -> Dict:
        """
        测试不同温度参数的效果

        Args:
            prompt: 测试提示词
            temperatures: 温度参数列表
        """
        results = {}

        for temp in temperatures:
            # 配置温度
            self._set_temperature(temp)

            # 发送请求
            response = self._send_message(prompt)

            results[temp] = {
                "response": response.get("answer", ""),
                "tokens": response.get("metadata", {}).get("usage", {}).get("total_tokens", 0)
            }

        return results

    def _set_temperature(self, temperature: float):
        """
        设置温度参数

        注意：模型参数配置需在Dify Web界面完成
        此方法仅作为测试框架的概念展示
        实际测试时可在Web界面手动调整温度后再调用API
        """
        print(f"请在Dify Web界面将温度设置为: {temperature}")

    def _send_message(self, prompt: str) -> Dict:
        """发送消息"""
        url = f"{self.base_url}/chat-messages"

        payload = {
            "inputs": {},
            "query": prompt,
            "response_mode": "blocking",
            "user": "temperature_test"
        }

        response = requests.post(url, headers=self.headers, json=payload)
        return response.json()

    def recommend_temperature(self, task_type: str) -> float:
        """
        根据任务类型推荐温度参数

        Args:
            task_type: 任务类型
        """
        recommendations = {
            "code_generation": 0.2,
            "factual_qa": 0.3,
            "technical_writing": 0.4,
            "general_chat": 0.7,
            "creative_writing": 0.9,
            "brainstorming": 1.0
        }

        return recommendations.get(task_type, 0.7)

# 使用示例
if __name__ == "__main__":
    optimizer = TemperatureOptimizer(
        api_key="your_api_key_here",
        app_id="app_id_here"
    )

    # 测试不同温度
    results = optimizer.test_temperature(
        prompt="写一个Python函数计算斐波那契数列",
        temperatures=[0.2, 0.5, 0.8, 1.0]
    )

    print("温度测试结果:")
    for temp, result in results.items():
        print(f"\n温度 {temp}:")
        print(f"响应: {result['response'][:100]}...")
        print(f"Token数: {result['tokens']}")

    # 获取推荐温度
    recommended_temp = optimizer.recommend_temperature("code_generation")
    print(f"\n推荐温度: {recommended_temp}")

class TopPOptimizer:
    """Top P参数优化器"""

    def __init__(self, api_key, app_id):
        self.api_key = api_key
        self.app_id = app_id
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.base_url = "https://api.dify.ai/v1"

    def test_top_p(self, prompt: str, top_p_values: List[float]) -> Dict:
        """测试不同Top P值的效果"""
        results = {}

        for top_p in top_p_values:
            # 配置Top P
            self._set_top_p(top_p)

            # 发送请求
            response = self._send_message(prompt)

            results[top_p] = {
                "response": response.get("answer", ""),
                "tokens": response.get("metadata", {}).get("usage", {}).get("total_tokens", 0)
            }

        return results

    def _set_top_p(self, top_p: float):
        """
        设置Top P参数

        注意：模型参数配置需在Dify Web界面完成
        """
        print(f"请在Dify Web界面将Top P设置为: {top_p}")

    def _send_message(self, prompt: str) -> Dict:
        """发送消息"""
        url = f"{self.base_url}/chat-messages"

        payload = {
            "inputs": {},
            "query": prompt,
            "response_mode": "blocking",
            "user": "top_p_test"
        }

        response = requests.post(url, headers=self.headers, json=payload)
        return response.json()

# 使用示例
if __name__ == "__main__":
    optimizer = TopPOptimizer(
        api_key="your_api_key_here",
        app_id="app_id_here"
    )

    # 测试不同Top P值
    results = optimizer.test_top_p(
        prompt="请介绍人工智能的发展历程",
        top_p_values=[0.1, 0.5, 0.9, 1.0]
    )

    print("Top P测试结果:")
    for top_p, result in results.items():
        print(f"\nTop P {top_p}:")
        print(f"响应: {result['response'][:100]}...")

class PenaltyOptimizer:
    """惩罚参数优化器"""

    def __init__(self, api_key, app_id):
        self.api_key = api_key
        self.app_id = app_id
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.base_url = "https://api.dify.ai/v1"

    def test_penalties(self, prompt: str,
                     presence_penalties: List[float],
                     frequency_penalties: List[float]) -> Dict:
        """
        测试不同惩罚参数组合

        Args:
            prompt: 测试提示词
            presence_penalties: 存在惩罚列表
            frequency_penalties: 频率惩罚列表
        """
        results = {}

        for presence in presence_penalties:
            for frequency in frequency_penalties:
                # 配置惩罚参数
                self._set_penalties(presence, frequency)

                # 发送请求
                response = self._send_message(prompt)

                results[f"p{presence}_f{frequency}"] = {
                    "response": response.get("answer", ""),
                    "tokens": response.get("metadata", {}).get("usage", {}).get("total_tokens", 0)
                }

        return results

    def _set_penalties(self, presence: float, frequency: float):
        """
        设置惩罚参数

        注意：模型参数配置需在Dify Web界面完成
        """
        print(f"请在Dify Web界面设置 presence_penalty={presence}, frequency_penalty={frequency}")

    def _send_message(self, prompt: str) -> Dict:
        """发送消息"""
        url = f"{self.base_url}/chat-messages"

        payload = {
            "inputs": {},
            "query": prompt,
            "response_mode": "blocking",
            "user": "penalty_test"
        }

        response = requests.post(url, headers=self.headers, json=payload)
        return response.json()

# 使用示例
if __name__ == "__main__":
    optimizer = PenaltyOptimizer(
        api_key="your_api_key_here",
        app_id="app_id_here"
    )

    # 测试不同惩罚参数组合
    results = optimizer.test_penalties(
        prompt="写一篇关于人工智能的文章",
        presence_penalties=[0, 0.5, 1.0],
        frequency_penalties=[0, 0.5, 1.0]
    )

    print("惩罚参数测试结果:")
    for key, result in results.items():
        print(f"\n{key}:")
        print(f"响应: {result['response'][:100]}...")

class PromptOptimizer:
    """提示词优化器"""

    def __init__(self, api_key, app_id):
        self.api_key = api_key
        self.app_id = app_id
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.base_url = "https://api.dify.ai/v1"

    def optimize_prompt(self, original_prompt: str,
                      iterations: int = 3) -> Dict:
        """
        优化提示词

        Args:
            original_prompt: 原始提示词
            iterations: 迭代次数
        """
        current_prompt = original_prompt
        best_prompt = original_prompt
        best_score = 0

        for i in range(iterations):
            # 测试当前提示词
            score = self._evaluate_prompt(current_prompt)

            # 更新最佳提示词
            if score > best_score:
                best_score = score
                best_prompt = current_prompt

            # 使用AI优化提示词
            current_prompt = self._improve_prompt(current_prompt, score)

        return {
            "original_prompt": original_prompt,
            "optimized_prompt": best_prompt,
            "improvement": best_score
        }

    def _evaluate_prompt(self, prompt: str) -> float:
        """评估提示词质量"""
        # 发送测试请求
        response = self._send_message(prompt)

        # 简单评估：基于响应长度和相关性
        answer = response.get("answer", "")
        length_score = min(len(answer) / 500, 1.0)

        # 这里可以添加更复杂的评估逻辑
        return length_score

    def _improve_prompt(self, prompt: str, current_score: float) -> str:
        """改进提示词"""
        improvement_prompt = f"""
        请改进以下提示词，使其更清晰、更有效：

        当前提示词：{prompt}
        当前得分：{current_score}

        请提供改进后的提示词。
        """

        response = self._send_message(improvement_prompt)
        return response.get("answer", prompt)

    def _send_message(self, prompt: str) -> Dict:
        """发送消息"""
        url = f"{self.base_url}/chat-messages"

        payload = {
            "inputs": {},
            "query": prompt,
            "response_mode": "blocking",
            "user": "prompt_optimizer"
        }

        response = requests.post(url, headers=self.headers, json=payload)
        return response.json()

# 使用示例
if __name__ == "__main__":
    optimizer = PromptOptimizer(
        api_key="your_api_key_here",
        app_id="app_id_here"
    )

    # 优化提示词
    result = optimizer.optimize_prompt(
        original_prompt="写一个Python函数",
        iterations=3
    )

    print(f"原始提示词: {result['original_prompt']}")
    print(f"优化提示词: {result['optimized_prompt']}")
    print(f"改进程度: {result['improvement']}")

import hashlib
import json
from typing import Dict, Optional
import time

class CacheManager:
    """缓存管理器"""

    def __init__(self):
        self.cache = {}
        self.cache_stats = {
            "hits": 0,
            "misses": 0,
            "total_requests": 0
        }

    def get(self, key: str) -> Optional[Dict]:
        """获取缓存"""
        self.cache_stats["total_requests"] += 1

        if key in self.cache:
            cache_item = self.cache[key]

            # 检查是否过期
            if time.time() < cache_item["expiry"]:
                self.cache_stats["hits"] += 1
                return cache_item["data"]
            else:
                # 删除过期缓存
                del self.cache[key]

        self.cache_stats["misses"] += 1
        return None

    def set(self, key: str, data: Dict, ttl: int = 3600):
        """
        设置缓存

        Args:
            key: 缓存键
            data: 缓存数据
            ttl: 生存时间（秒）
        """
        self.cache[key] = {
            "data": data,
            "expiry": time.time() + ttl,
            "created_at": time.time()
        }

    def generate_key(self, prompt: str, model: str,
                   params: Dict) -> str:
        """
        生成缓存键

        Args:
            prompt: 提示词
            model: 模型名称
            params: 模型参数
        """
        key_string = f"{prompt}|{model}|{json.dumps(params, sort_keys=True)}"  # json.dumps将Python对象转为JSON字符串
        return hashlib.md5(key_string.encode()).hexdigest()

    def get_stats(self) -> Dict:
        """获取缓存统计"""
        hit_rate = 0
        if self.cache_stats["total_requests"] > 0:
            hit_rate = self.cache_stats["hits"] / self.cache_stats["total_requests"]

        return {
            **self.cache_stats,
            "hit_rate": hit_rate,
            "cache_size": len(self.cache)
        }

    def clear(self):
        """清空缓存"""
        self.cache.clear()
        self.cache_stats = {
            "hits": 0,
            "misses": 0,
            "total_requests": 0
        }

# 使用示例
if __name__ == "__main__":
    cache = CacheManager()

    # 生成缓存键
    key = cache.generate_key(
        prompt="写一个Python函数",
        model="gpt-4o-mini",
        params={"temperature": 0.7, "max_tokens": 2048}
    )

    # 设置缓存
    cache.set(key, {"answer": "这是一个Python函数..."}, ttl=3600)

    # 获取缓存
    cached_data = cache.get(key)
    print(f"缓存数据: {cached_data}")

    # 获取统计
    stats = cache.get_stats()
    print(f"缓存统计: {stats}")