07 - 实战项目¶

import requests
import json
from typing import Dict, List

class CustomerServiceWorkflow:
    """智能客服工作流"""

    def __init__(self, api_key: str):
        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 create_app(self) -> Dict:
        """
        创建应用

        注意：应用需在Dify Web界面创建，Dify不提供通过API创建应用的接口
        创建步骤：
        1. 登录Dify控制台 -> 创建应用 -> 选择“工作流”类型
        2. 在可视化编辑器中设计工作流
        3. 发布并获取API Key
        """
        app_config = {
            "name": "智能客服系统",
            "description": "基于Dify的智能客服系统",
            "mode": "workflow",
            "icon": "🤖",
            "icon_background": "#FFEAD5"
        }
        print(f"请在Dify Web界面创建应用: {app_config['name']}")
        return {"message": "请在Dify Web界面创建应用", "config": app_config}

    def build_workflow(self) -> Dict:
        """构建完整工作流"""
        return {
            "graph": {
                "nodes": [
                    # 开始节点
                    {
                        "id": "start",
                        "type": "start",
                        "data": {
                            "title": "开始",
                            "variables": [
                                {
                                    "variable": "user_input",
                                    "label": "用户输入",
                                    "type": "paragraph",
                                    "max_length": 1000,
                                    "required": True
                                },
                                {
                                    "variable": "user_id",
                                    "label": "用户ID",
                                    "type": "text-input",
                                    "required": True
                                },
                                {
                                    "variable": "conversation_id",
                                    "label": "会话ID",
                                    "type": "text-input",
                                    "required": False
                                }
                            ]
                        }
                    },
                    # 意图识别节点
                    {
                        "id": "intent_recognition",
                        "type": "llm",
                        "data": {
                            "title": "意图识别",
                            "model": {
                                "provider": "openai",
                                "name": "gpt-4o-mini",
                                "mode": "chat"
                            },
                            "prompt_template": [
                                {
                                    "role": "system",
                                    "text": """你是一个意图识别专家。请识别用户问题的意图。

可能的意图包括：
1. product_inquiry - 产品咨询
2. order_inquiry - 订单查询
3. technical_support - 技术支持
4. refund_request - 退款申请
5. complaint - 投诉
6. greeting - 问候
7. other - 其他

只返回意图类型，不要添加其他内容。"""
                                },
                                {
                                    "role": "user",
                                    "text": "{{#start.user_input#}}"
                                }
                            ]
                        }
                    },
                    # 知识库检索节点
                    {
                        "id": "kb_retrieval",
                        "type": "knowledge-retrieval",
                        "data": {
                            "title": "知识库检索",
                            "dataset_ids": [
                                "faq_dataset_id",
                                "product_manual_dataset_id",
                                "policy_dataset_id"
                            ],
                            "retrieval_mode": "multiple",
                            "multiple_retrieval_config": {
                                "top_k": 5,
                                "score_threshold": 0.5,
                                "reranking_enable": True
                            }
                        }
                    },
                    # LLM生成回复节点
                    {
                        "id": "llm_response",
                        "type": "llm",
                        "data": {
                            "title": "生成回复",
                            "model": {
                                "provider": "openai",
                                "name": "gpt-4o",
                                "mode": "chat"
                            },
                            "prompt_template": [
                                {
                                    "role": "system",
                                    "text": """你是一个专业的客服代表，需要：
1. 礼貌友好地回答用户问题
2. 基于检索到的知识库信息回答
3. 如果知识库中没有相关信息，诚实地说明
4. 保持专业和耐心
5. 适当使用表情符号增加亲和力"""
                                },
                                {
                                    "role": "user",
                                    "text": """用户问题：{{#start.user_input#}}

用户意图：{{#intent_recognition.text#}}

检索到的相关信息：
{{#kb_retrieval.context#}}

请基于以上信息回答用户问题。"""
                                }
                            ]
                        }
                    },
                    # 情感分析节点
                    {
                        "id": "sentiment_analysis",
                        "type": "llm",
                        "data": {
                            "title": "情感分析",
                            "model": {
                                "provider": "openai",
                                "name": "gpt-4o-mini",
                                "mode": "chat"
                            },
                            "prompt_template": [
                                {
                                    "role": "system",
                                    "text": "你是一个情感分析专家。请分析用户输入的情感倾向。"
                                },
                                {
                                    "role": "user",
                                    "text": """用户输入：{{#start.user_input#}}

请分析用户的情感倾向（positive/negative/neutral）和情绪强度（1-10）。

以JSON格式返回：
{
    "sentiment": "情感倾向",
    "intensity": 情绪强度
}"""
                                }
                            ]
                        }
                    },
                    # 条件判断节点
                    {
                        "id": "check_human_needed",
                        "type": "if-else",
                        "data": {
                            "title": "是否需要人工",
                            "cases": [
                                {
                                    "case_id": "need_human",
                                    "conditions": [
                                        {
                                            "variable_selector": ["sentiment_analysis", "text"],
                                            "comparison_operator": "contains",
                                            "value": "negative"
                                        },
                                        {
                                            "variable_selector": ["intent_recognition", "text"],
                                            "comparison_operator": "is",
                                            "value": "complaint"
                                        }
                                    ],
                                    "logical_operator": "or"
                                },
                                {
                                    "case_id": "auto_reply",
                                    "conditions": [],
                                    "logical_operator": "and"
                                }
                            ]
                        }
                    },
                    # 人工转接节点
                    {
                        "id": "human_handoff",
                        "type": "code",
                        "data": {
                            "title": "人工转接",
                            "code": """
import json

def main(user_id: str, user_input: str, sentiment: str) -> dict:
    # 生成转接消息
    message = {
        "type": "human_handoff",
        "user_id": user_id,
        "user_input": user_input,
        "sentiment": sentiment,
        "timestamp": "2024-01-01T00:00:00Z",
        "priority": "high" if sentiment == "negative" else "normal"
    }

    # 这里可以发送通知给人工客服
    # 例如：发送到队列、发送邮件等

    return {
        "handoff": True,
        "message": "已为您转接人工客服，请稍候...",
        "data": message
    }
""",
                            "outputs": [
                                {"variable": "handoff", "type": "boolean"},
                                {"variable": "message", "type": "string"},
                                {"variable": "data", "type": "object"}
                            ]
                        }
                    },
                    # 数据记录节点
                    {
                        "id": "log_data",
                        "type": "code",
                        "data": {
                            "title": "记录数据",
                            "code": """
import json
from datetime import datetime

def main(user_id: str, user_input: str, response: str,
         intent: str, sentiment: str) -> dict:
    # 创建日志记录
    log_entry = {
        "user_id": user_id,
        "user_input": user_input,
        "bot_response": response,
        "intent": intent,
        "sentiment": sentiment,
        "timestamp": datetime.now().isoformat()
    }

    # 这里可以保存到数据库
    # 例如：insert into conversation_logs values (...)

    return {
        "logged": True,
        "log_entry": log_entry
    }
""",
                            "outputs": [
                                {"variable": "logged", "type": "boolean"},
                                {"variable": "log_entry", "type": "object"}
                            ]
                        }
                    },
                    # 结束节点
                    {
                        "id": "end",
                        "type": "end",
                        "data": {
                            "title": "结束",
                            "outputs": [
                                {
                                    "value_selector": ["llm_response", "text"],
                                    "variable": "auto_response"
                                },
                                {
                                    "value_selector": ["human_handoff", "message"],
                                    "variable": "human_response"
                                },
                                {
                                    "value_selector": ["log_data", "log_entry"],
                                    "variable": "log_data"
                                },
                                {
                                    "value_selector": ["intent_recognition", "text"],
                                    "variable": "intent"
                                },
                                {
                                    "value_selector": ["sentiment_analysis", "text"],
                                    "variable": "sentiment"
                                }
                            ]
                        }
                    }
                ],
                "edges": [
                    {"id": "start-intent", "source": "start", "target": "intent_recognition"},
                    {"id": "intent-kb", "source": "intent_recognition", "target": "kb_retrieval"},
                    {"id": "kb-llm", "source": "kb_retrieval", "target": "llm_response"},
                    {"id": "llm-sentiment", "source": "llm_response", "target": "sentiment_analysis"},
                    {"id": "sentiment-check", "source": "sentiment_analysis", "target": "check_human_needed"},
                    {"id": "check-human", "source": "check_human_needed", "target": "human_handoff", "sourceHandle": "need_human"},
                    {"id": "check-log", "source": "check_human_needed", "target": "log_data", "sourceHandle": "auto_reply"},
                    {"id": "human-end", "source": "human_handoff", "target": "end"},
                    {"id": "log-end", "source": "log_data", "target": "end"}
                ]
            }
        }

    def deploy_workflow(self, app_id: str, workflow: Dict) -> Dict:
        """
        部署工作流

        注意：工作流的设计和发布需在Dify Web界面完成
        发布后可通过 POST /v1/workflows/run 调用工作流
        以下展示的是工作流的数据结构，供在Web界面设计时参考
        """
        print(f"请在Dify Web界面设计并发布工作流")
        print(f"工作流包含 {len(workflow.get('graph', {}).get('nodes', []))} 个节点")
        return {"message": "请在Dify Web界面发布工作流"}

# 使用示例
if __name__ == "__main__":
    workflow = CustomerServiceWorkflow(api_key="your_api_key_here")

    # 注意：应用需在Dify Web界面创建
    app = workflow.create_app()
    print(f"应用配置: {app}")

    # 构建工作流配置（在Web界面设计时参考）
    workflow_config = workflow.build_workflow()

    # 工作流需在Web界面发布
    result = workflow.deploy_workflow("your_app_id", workflow_config)
    print(f"工作流发布提示: {result}")

class KnowledgeBaseManager:
    """知识库管理器"""

    def __init__(self, api_key: str):
        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 create_faq_dataset(self) -> Dict:
        """创建FAQ知识库"""
        url = f"{self.base_url}/datasets"

        payload = {
            "name": "FAQ知识库",
            "description": "常见问题解答",
            "permission": "only_me",
            "data_source_type": "upload_file"
        }

        try:
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}

    def create_product_manual_dataset(self) -> Dict:
        """创建产品手册知识库"""
        url = f"{self.base_url}/datasets"

        payload = {
            "name": "产品手册知识库",
            "description": "产品使用手册",
            "permission": "only_me",
            "data_source_type": "upload_file"
        }

        try:
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}

    def create_policy_dataset(self) -> Dict:
        """创建政策文档知识库"""
        url = f"{self.base_url}/datasets"

        payload = {
            "name": "政策文档知识库",
            "description": "公司政策和规定",
            "permission": "only_me",
            "data_source_type": "upload_file"
        }

        try:
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}

    def upload_document(self, dataset_id: str, file_path: str) -> Dict:
        """上传文档到知识库"""
        # 构建文件上传API地址，使用Dify的create-by-file端点
        url = f"{self.base_url}/datasets/{dataset_id}/document/create-by-file"

        try:
            with open(file_path, 'rb') as f:  # with自动管理资源，确保文件正确关闭
                # 以二进制模式读取文件，准备multipart/form-data上传
                files = {'file': f}
                # 设置索引技术为高质量模式（使用Embedding向量化）
                # process_rule设为automatic让Dify自动处理分段和清洗
                data = {
                    'indexing_technique': 'high_quality',
                    'process_rule': json.dumps({"mode": "automatic"})  # json.dumps将Python对象转为JSON字符串
                }

                # 注意：文件上传不能使用Content-Type: application/json
                # 只需传Authorization头，requests会自动设置multipart头
                response = requests.post(
                    url,
                    headers={"Authorization": f"Bearer {self.api_key}"},
                    files=files,
                    data=data
                )
                response.raise_for_status()
                return response.json()
        except Exception as e:
            return {"error": str(e)}

    def batch_upload(self, dataset_id: str, file_paths: List[str]) -> List[Dict]:
        """批量上传文档到指定知识库"""
        results = []

        # 遍历所有文件路径，逐个上传（Dify暂不支持批量上传API）
        for file_path in file_paths:
            result = self.upload_document(dataset_id, file_path)
            # 记录每个文件的上传结果，便于后续检查失败情况
            results.append({
                "file": file_path,
                "result": result
            })

        return results

# 使用示例
if __name__ == "__main__":
    manager = KnowledgeBaseManager(api_key="your_api_key_here")

    # 创建知识库
    faq_dataset = manager.create_faq_dataset()
    print(f"FAQ知识库创建成功！ID: {faq_dataset.get('id')}")

    product_dataset = manager.create_product_manual_dataset()
    print(f"产品手册知识库创建成功！ID: {product_dataset.get('id')}")

    policy_dataset = manager.create_policy_dataset()
    print(f"政策文档知识库创建成功！ID: {policy_dataset.get('id')}")

    # 上传文档
    results = manager.batch_upload(
        dataset_id=faq_dataset["id"],
        file_paths=[
            "docs/faq1.pdf",
            "docs/faq2.pdf",
            "docs/faq3.pdf"
        ]
    )

    print(f"批量上传完成，共处理 {len(results)} 个文件")

class ModelOptimizer:
    """模型优化器"""

    def __init__(self, api_key: str, app_id: str):
        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_for_customer_service(self) -> Dict:
        """
        为客服场景优化模型

        注意：模型配置需在Dify Web界面完成
        以下展示推荐的配置参数
        """
        recommended_config = {
            "provider": "openai",
            "model_name": "gpt-4o",
            "temperature": 0.7,
            "top_p": 0.9,
            "max_tokens": 2048,
            "presence_penalty": 0.1,
            "frequency_penalty": 0.1,
            "system_prompt": """你是一个专业的客服代表，需要遵循以下原则：

1. 礼貌友好：使用礼貌用语，保持友好态度
2. 专业准确：基于事实回答，确保准确性
3. 简洁明了：回答简洁，避免冗长
4. 同理心：理解用户需求，表达同理心
5. 主动性：主动提供相关信息和解决方案

请用中文回答，适当使用表情符号增加亲和力。"""
        }
        print(f"请在Dify Web界面配置模型参数: {recommended_config}")
        return recommended_config

    def test_model_performance(self, test_queries: List[str]) -> List[Dict]:
        """测试模型性能——对一组测试问题逐一调用工作流，记录响应质量和延迟"""
        # 使用工作流运行接口（blocking模式等待完整响应）
        url = f"{self.base_url}/workflows/run"

        results = []

        for query in test_queries:
            # 构建工作流输入参数，与工作流开始节点的变量定义对应
            payload = {
                "inputs": {
                    "user_input": query,
                    "user_id": "test_user",
                    "conversation_id": ""  # 空字符串表示新会话
                },
                "response_mode": "blocking",  # 同步等待，适合测试场景
                "user": "test_user"
            }

            try:
                import time
                # 记录请求开始时间，用于计算端到端响应延迟
                start_time = time.time()

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

                end_time = time.time()
                response_time = end_time - start_time

                result = response.json()

                # 收集关键性能指标：响应内容、延迟时间、Token消耗
                results.append({
                    "query": query,
                    "response": result.get("data", {}).get("outputs", {}),
                    "response_time": response_time,
                    "tokens": result.get("metadata", {}).get("usage", {}).get("total_tokens", 0)
                })

            except requests.exceptions.RequestException as e:
                results.append({
                    "query": query,
                    "error": str(e)
                })

        return results

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

    # 优化模型
    result = optimizer.optimize_for_customer_service()
    print(f"模型优化结果: {result}")

    # 测试性能
    test_queries = [
        "你们的产品有哪些？",
        "如何查询订单？",
        "我想退款",
        "产品坏了怎么办？"
    ]

    results = optimizer.test_model_performance(test_queries)

    print("\n性能测试结果:")
    for result in results:
        print(f"\n问题: {result['query']}")
        if "error" in result:
            print(f"错误: {result['error']}")
        else:
            print(f"响应时间: {result['response_time']:.2f}秒")
            print(f"Token数: {result['tokens']}")

class CustomerServiceDeployer:
    """客服系统部署器"""

    def __init__(self, api_key: str):
        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 deploy_complete_system(self, app_id: str) -> Dict:
        """
        部署完整系统——按顺序执行五步部署流程

        注意：应用需先在Dify Web界面创建，获取app_id和API Key后再调用此方法

        Args:
            app_id: 在Dify Web界面创建应用后获取的应用ID
        """
        # ===== 步骤1：获取应用配置建议 =====
        # 初始化工作流管理器，打印推荐的应用配置参数
        workflow = CustomerServiceWorkflow(self.api_key)
        app_config = workflow.create_app()
        print(f"应用配置建议: {app_config['config']}")

        # ===== 步骤2：创建三个业务知识库 =====
        # 分别创建FAQ、产品手册、政策文档知识库，实现多源知识检索
        kb_manager = KnowledgeBaseManager(self.api_key)
        faq_dataset = kb_manager.create_faq_dataset()
        product_dataset = kb_manager.create_product_manual_dataset()
        policy_dataset = kb_manager.create_policy_dataset()

        # ===== 步骤3：生成工作流配置 =====
        # 构建包含意图识别→知识库检索→LLM生成→情感分析的完整工作流
        workflow_config = workflow.build_workflow()
        # 注意：工作流需在Dify Web界面的可视化编辑器中配置
        print(f"工作流配置已生成，请在Web界面中按此配置搭建工作流")

        # ===== 步骤4：优化模型参数 =====
        # 针对客服场景调整temperature、max_tokens等参数
        optimizer = ModelOptimizer(self.api_key, app_id)
        optimizer.optimize_for_customer_service()

        # ===== 步骤5：生成API访问密钥 =====
        api_keys = self.generate_api_keys(app_id)

        return {
            "app_id": app_id,
            "faq_dataset_id": faq_dataset.get("id"),
            "product_dataset_id": product_dataset.get("id"),
            "policy_dataset_id": policy_dataset.get("id"),
            "workflow_config": workflow_config,
            "api_keys": api_keys,
            "status": "configured"
        }

    def generate_api_keys(self, app_id: str) -> List[Dict]:
        """
        生成API密钥

        注意：API密钥需在Dify Web界面管理
        进入应用 -> 访问API -> 创建/管理API密钥
        Dify不提供通过API创建密钥的接口
        """
        print(f"请在Dify Web界面为应用 {app_id} 创建API密钥")
        print("步骤：应用 -> 访问API -> API密钥 -> 创建")
        return [{"message": "请在Web界面创建API密钥"}]

    def get_deployment_info(self, app_id: str) -> Dict:
        """
        获取部署信息

        通过 /info 和 /parameters 接口获取应用基本信息
        """
        info = {}

        try:  # try/except捕获异常
            # 获取应用基本信息
            resp = requests.get(
                f"{self.base_url}/info",
                headers=self.headers
            )
            resp.raise_for_status()
            info["app_info"] = resp.json()

            # 获取应用参数
            resp = requests.get(
                f"{self.base_url}/parameters",
                headers=self.headers
            )
            resp.raise_for_status()
            info["parameters"] = resp.json()

            info["app_id"] = app_id
            info["status"] = "running"
            return info

        except requests.exceptions.RequestException as e:
            return {"error": str(e)}

# 使用示例
if __name__ == "__main__":
    deployer = CustomerServiceDeployer(api_key="your_api_key_here")

    # 注意：需先在Dify Web界面创建应用，获取app_id
    app_id = "your_app_id_here"  # 在Dify Web控制台创建后获取

    # 部署完整系统
    deployment_info = deployer.deploy_complete_system(app_id)

    print("配置完成！")
    print(f"应用ID: {deployment_info['app_id']}")
    print(f"FAQ知识库ID: {deployment_info['faq_dataset_id']}")
    print(f"产品手册知识库ID: {deployment_info['product_dataset_id']}")
    print(f"政策文档知识库ID: {deployment_info['policy_dataset_id']}")
    print(f"API密钥: {deployment_info['api_keys']}")

    # 获取部署信息
    info = deployer.get_deployment_info(deployment_info['app_id'])
    print(f"\n部署信息: {info}")