Dify开发实践指南：从入门到精通的完整开发手册

# 1. 克隆项目
git clone https://github.com/langgenius/dify.git
cd dify

# 2. 启动开发环境
cp .env.example .env
docker-compose -f docker-compose.dev.yaml up -d

# 3. 安装依赖
# 后端依赖
cd api
uv sync

# 前端依赖
cd ../web
pnpm install

# 4. 启动开发服务器
# 后端
./dev/start-api

# 前端
cd web
pnpm dev

# api/.env 配置示例
# 数据库配置
DATABASE_URL=postgresql://postgres:password@localhost:5432/dify
REDIS_URL=redis://localhost:6379

# 模型配置
OPENAI_API_KEY=your_openai_api_key
ANTHROPIC_API_KEY=your_anthropic_api_key

# 存储配置
S3_ENDPOINT=http://localhost:9000
S3_ACCESS_KEY=minioadmin
S3_SECRET_KEY=minioadmin

# 开发配置
DEBUG=True
LOG_LEVEL=DEBUG
FLASK_ENV=development

# api/app_factory.py
from flask import Flask
from configs import dify_config
from extensions import ext_database, ext_redis, ext_celery

def create_app() -> Flask:
    """创建Flask应用实例"""
    
    app = Flask(__name__)
    
    # 加载配置
    app.config.from_object(dify_config)
    
    # 初始化扩展
    initialize_extensions(app)
    
    # 注册蓝图
    register_blueprints(app)
    
    # 注册错误处理器
    register_error_handlers(app)
    
    return app

def initialize_extensions(app: Flask):
    """初始化Flask扩展"""
    ext_database.init_app(app)
    ext_redis.init_app(app)
    ext_celery.init_app(app)

def register_blueprints(app: Flask):
    """注册蓝图"""
    from controllers.console import bp as console_bp
    from controllers.service_api import bp as service_api_bp
    from controllers.web import bp as web_bp
    
    app.register_blueprint(console_bp, url_prefix='/console/api')
    app.register_blueprint(service_api_bp, url_prefix='/v1')
    app.register_blueprint(web_bp, url_prefix='/api')

# controllers/service_api/completion.py
from flask import request
from flask_restx import Resource, reqparse
from libs.login import validate_app_token
from services.completion_service import CompletionService

class CompletionApi(Resource):
    """文本补全API"""
    
    @validate_app_token
    def post(self, app_model, end_user):
        """
        创建文本补全
        
        Args:
            app_model: 应用模型（装饰器注入）
            end_user: 终端用户（装饰器注入）
        """
        
        # 1. 参数解析
        parser = reqparse.RequestParser()
        parser.add_argument('inputs', type=dict, required=True)
        parser.add_argument('query', type=str, required=True)
        parser.add_argument('response_mode', type=str, choices=['blocking', 'streaming'])
        parser.add_argument('user', type=str, required=True)
        
        args = parser.parse_args()
        
        # 2. 业务逻辑调用
        try:
            result = CompletionService.generate(
                app_model=app_model,
                user=end_user,
                args=args,
                invoke_from=InvokeFrom.SERVICE_API,
                streaming=args['response_mode'] == 'streaming'
            )
            
            return result
            
        except Exception as e:
            logger.exception(f"Completion generation failed: {e}")
            raise InternalServerError()

# services/completion_service.py
from typing import Generator, Union
from models.model import App, EndUser
from core.app.apps.completion.app_generator import CompletionAppGenerator

class CompletionService:
    """文本补全服务"""
    
    @classmethod
    def generate(
        cls,
        app_model: App,
        user: Union[Account, EndUser],
        args: dict,
        invoke_from: InvokeFrom,
        streaming: bool = True
    ) -> Generator:
        """
        生成文本补全
        
        Args:
            app_model: 应用模型
            user: 用户实例
            args: 生成参数
            invoke_from: 调用来源
            streaming: 是否流式输出
            
        Returns:
            生成结果流
        """
        
        # 1. 参数验证
        cls._validate_args(args)
        
        # 2. 权限检查
        cls._check_permissions(app_model, user)
        
        # 3. 创建生成器
        generator = CompletionAppGenerator()
        
        # 4. 执行生成
        try:
            result_stream = generator.generate(
                app_model=app_model,
                user=user,
                args=args,
                invoke_from=invoke_from,
                streaming=streaming
            )
            
            # 5. 结果处理
            for result in result_stream:
                yield cls._process_result(result)
                
        except Exception as e:
            logger.exception(f"Generation failed: {e}")
            raise CompletionGenerationError(str(e))
    
    @classmethod
    def _validate_args(cls, args: dict):
        """验证参数"""
        required_fields = ['inputs', 'query', 'user']
        for field in required_fields:
            if field not in args:
                raise ValueError(f"Missing required field: {field}")
    
    @classmethod
    def _check_permissions(cls, app_model: App, user: Union[Account, EndUser]):
        """检查权限"""
        if app_model.status != 'normal':
            raise AppUnavailableError()
        
        # 其他权限检查逻辑

// web/app/layout.tsx
import type { Metadata } from 'next'
import { Inter } from 'next/font/google'
import './globals.css'
import { Providers } from './providers'

const inter = Inter({ subsets: ['latin'] })

export const metadata: Metadata = {
  title: 'Dify',
  description: 'The next generation of AI application development platform',
}

export default function RootLayout({
  children,
}: {
  children: React.ReactNode
}) {
  return (
    <html lang="en">
      <body className={inter.className}>
        <Providers>
          {children}
        </Providers>
      </body>
    </html>
  )
}

// web/app/components/chat/chat-input.tsx
'use client'

import { useState, useRef } from 'react'
import { useTranslation } from 'react-i18next'
import { PaperAirplaneIcon } from '@heroicons/react/24/solid'
import Button from '@/app/components/base/button'
import Textarea from '@/app/components/base/textarea'

interface ChatInputProps {
  onSend: (message: string) => void
  disabled?: boolean
  placeholder?: string
}

export default function ChatInput({
  onSend,
  disabled = false,
  placeholder
}: ChatInputProps) {
  const { t } = useTranslation()
  const [message, setMessage] = useState('')
  const textareaRef = useRef<HTMLTextAreaElement>(null)

  const handleSend = () => {
    if (!message.trim() || disabled) return
    
    onSend(message.trim())
    setMessage('')
    
    // 重置文本框高度
    if (textareaRef.current) {
      textareaRef.current.style.height = 'auto'
    }
  }

  const handleKeyDown = (e: React.KeyboardEvent) => {
    if (e.key === 'Enter' && !e.shiftKey) {
      e.preventDefault()
      handleSend()
    }
  }

  return (
    <div className="flex items-end space-x-2 p-4 border-t border-gray-200">
      <div className="flex-1">
        <Textarea
          ref={textareaRef}
          value={message}
          onChange={setMessage}
          onKeyDown={handleKeyDown}
          placeholder={placeholder || t('chat.input.placeholder')}
          disabled={disabled}
          autoSize
          maxRows={4}
          className="resize-none"
        />
      </div>
      <Button
        type="primary"
        onClick={handleSend}
        disabled={disabled || !message.trim()}
        className="flex-shrink-0"
      >
        <PaperAirplaneIcon className="w-4 h-4" />
      </Button>
    </div>
  )
}

// web/context/app-context.tsx
'use client'

import { createContext, useContext, useReducer, ReactNode } from 'react'

interface AppState {
  user: User | null
  currentApp: App | null
  loading: boolean
  error: string | null
}

type AppAction =
  | { type: 'SET_USER'; payload: User | null }
  | { type: 'SET_CURRENT_APP'; payload: App | null }
  | { type: 'SET_LOADING'; payload: boolean }
  | { type: 'SET_ERROR'; payload: string | null }

const initialState: AppState = {
  user: null,
  currentApp: null,
  loading: false,
  error: null
}

function appReducer(state: AppState, action: AppAction): AppState {
  switch (action.type) {
    case 'SET_USER':
      return { ...state, user: action.payload }
    case 'SET_CURRENT_APP':
      return { ...state, currentApp: action.payload }
    case 'SET_LOADING':
      return { ...state, loading: action.payload }
    case 'SET_ERROR':
      return { ...state, error: action.payload }
    default:
      return state
  }
}

const AppContext = createContext<{
  state: AppState
  dispatch: React.Dispatch<AppAction>
} | null>(null)

export function AppProvider({ children }: { children: ReactNode }) {
  const [state, dispatch] = useReducer(appReducer, initialState)

  return (
    <AppContext.Provider value={{ state, dispatch }}>
      {children}
    </AppContext.Provider>
  )
}

export function useAppContext() {
  const context = useContext(AppContext)
  if (!context) {
    throw new Error('useAppContext must be used within AppProvider')
  }
  return context
}

# 智能客服应用配置
class CustomerServiceApp:
    """智能客服应用最佳实践"""
    
    def __init__(self):
        self.app_config = {
            "mode": "chat",
            "model_config": {
                "provider": "openai",
                "model": "gpt-4",
                "parameters": {
                    "temperature": 0.1,  # 降低随机性，提高一致性
                    "max_tokens": 1000,
                    "top_p": 0.9,
                    "presence_penalty": 0.1,
                    "frequency_penalty": 0.1
                }
            },
            "prompt_template": {
                "system_message": """你是一个专业的客服助手，请遵循以下原则：
1. 保持礼貌和专业的语调
2. 准确理解用户问题并提供有用的解答
3. 如果不确定答案，诚实说明并提供替代方案
4. 优先使用知识库中的信息回答问题
5. 对于复杂问题，引导用户联系人工客服

当前时间：{{#sys.datetime#}}
用户信息：{{#sys.user_name#}}""",
                "user_input_form": [
                    {
                        "variable": "query",
                        "label": "用户问题",
                        "type": "text-input",
                        "required": True,
                        "max_length": 500
                    }
                ]
            },
            "dataset_configs": {
                "retrieval_model": "vector",
                "top_k": 5,
                "score_threshold": 0.7,
                "reranking_enable": True,
                "reranking_model": {
                    "provider": "cohere",
                    "model": "rerank-multilingual-v2.0"
                }
            },
            "conversation_variables": [
                {
                    "variable": "user_level",
                    "name": "用户等级",
                    "description": "VIP/普通用户标识"
                },
                {
                    "variable": "issue_category",
                    "name": "问题类别",
                    "description": "技术/账单/产品问题分类"
                }
            ]
        }
    
    def optimize_for_performance(self):
        """性能优化配置"""
        return {
            # 启用流式响应
            "stream": True,
            
            # 配置缓存策略
            "cache_config": {
                "enabled": True,
                "ttl": 3600,  # 1小时缓存
                "cache_key_strategy": "semantic_hash"
            },
            
            # 并发控制
            "concurrency_config": {
                "max_concurrent_requests": 10,
                "queue_timeout": 30
            },
            
            # 监控配置
            "monitoring": {
                "enable_metrics": True,
                "log_level": "INFO",
                "trace_requests": True
            }
        }

// web/app/chat/page.tsx
'use client'

import { useState, useEffect, useRef } from 'react'
import { useTranslation } from 'react-i18next'
import ChatInput from '@/app/components/chat/chat-input'
import MessageList from '@/app/components/chat/message-list'
import { chatAPI } from '@/service/chat'
import type { Message } from '@/types/chat'

export default function ChatPage() {
  const { t } = useTranslation()
  const [messages, setMessages] = useState<Message[]>([])
  const [loading, setLoading] = useState(false)
  const [conversationId, setConversationId] = useState<string>()
  const abortControllerRef = useRef<AbortController>()

  const handleSendMessage = async (content: string) => {
    if (loading) return

    // 添加用户消息
    const userMessage: Message = {
      id: Date.now().toString(),
      role: 'user',
      content,
      timestamp: new Date()
    }
    setMessages(prev => [...prev, userMessage])
    setLoading(true)

    // 创建助手消息占位符
    const assistantMessage: Message = {
      id: (Date.now() + 1).toString(),
      role: 'assistant',
      content: '',
      timestamp: new Date(),
      streaming: true
    }
    setMessages(prev => [...prev, assistantMessage])

    try {
      // 创建新的AbortController
      abortControllerRef.current = new AbortController()

      // 调用聊天API
      const response = await chatAPI.sendMessage({
        query: content,
        conversation_id: conversationId,
        inputs: {},
        response_mode: 'streaming',
        user: 'user-123'
      }, {
        signal: abortControllerRef.current.signal,
        onMessage: (chunk) => {
          // 更新助手消息内容
          setMessages(prev => prev.map(msg => 
            msg.id === assistantMessage.id 
              ? { ...msg, content: msg.content + chunk.answer }
              : msg
          ))
        },
        onEnd: (data) => {
          // 完成流式响应
          setMessages(prev => prev.map(msg => 
            msg.id === assistantMessage.id 
              ? { ...msg, streaming: false }
              : msg
          ))
          
          // 设置对话ID
          if (data.conversation_id) {
            setConversationId(data.conversation_id)
          }
        },
        onError: (error) => {
          console.error('Chat error:', error)
          // 显示错误消息
          setMessages(prev => prev.map(msg => 
            msg.id === assistantMessage.id 
              ? { ...msg, content: t('chat.error.general'), error: true, streaming: false }
              : msg
          ))
        }
      })

    } catch (error) {
      console.error('Send message error:', error)
    } finally {
      setLoading(false)
    }
  }

  const handleStopGeneration = () => {
    if (abortControllerRef.current) {
      abortControllerRef.current.abort()
      setLoading(false)
    }
  }

  return (
    <div className="flex flex-col h-screen bg-gray-50">
      {/* 头部 */}
      <div className="flex-shrink-0 bg-white border-b border-gray-200 px-6 py-4">
        <h1 className="text-xl font-semibold text-gray-900">
          {t('chat.title')}
        </h1>
      </div>

      {/* 消息列表 */}
      <div className="flex-1 overflow-hidden">
        <MessageList 
          messages={messages}
          loading={loading}
          onStop={handleStopGeneration}
        />
      </div>

      {/* 输入框 */}
      <div className="flex-shrink-0 bg-white">
        <ChatInput
          onSend={handleSendMessage}
          disabled={loading}
          placeholder={t('chat.input.placeholder')}
        />
      </div>
    </div>
  )
}

# 工作流节点开发示例
class CustomBusinessNode(BaseNode):
    """自定义业务节点"""
    
    _node_data_cls = CustomBusinessNodeData
    _node_type = NodeType.CUSTOM_BUSINESS
    
    def _run(self, variable_pool: VariablePool) -> NodeRunResult:
        """
        执行自定义业务逻辑
        
        Args:
            variable_pool: 变量池
            
        Returns:
            节点执行结果
        """
        
        # 1. 获取输入参数
        inputs = self.node_data.inputs
        business_type = variable_pool.get(inputs['business_type'])
        user_data = variable_pool.get(inputs['user_data'])
        
        # 2. 执行业务逻辑
        try:
            result = self._execute_business_logic(business_type, user_data)
            
            # 3. 返回结果
            return NodeRunResult(
                status=WorkflowNodeExecutionStatus.SUCCEEDED,
                outputs={
                    'result': result,
                    'status': 'success',
                    'timestamp': datetime.now().isoformat()
                }
            )
            
        except Exception as e:
            logger.exception(f"Business node execution failed: {e}")
            return NodeRunResult(
                status=WorkflowNodeExecutionStatus.FAILED,
                error=str(e)
            )
    
    def _execute_business_logic(self, business_type: str, user_data: dict) -> dict:
        """执行具体业务逻辑"""
        
        if business_type == 'order_processing':
            return self._process_order(user_data)
        elif business_type == 'user_verification':
            return self._verify_user(user_data)
        elif business_type == 'payment_handling':
            return self._handle_payment(user_data)
        else:
            raise ValueError(f"Unsupported business type: {business_type}")
    
    def _process_order(self, user_data: dict) -> dict:
        """处理订单"""
        # 订单处理逻辑
        return {
            'order_id': f"ORD-{int(time.time())}",
            'status': 'processed',
            'amount': user_data.get('amount', 0)
        }
    
    @classmethod
    def get_default_config(cls, filters: Optional[dict] = None) -> dict:
        """获取默认配置"""
        return {
            'inputs': {
                'business_type': {
                    'type': 'string',
                    'required': True,
                    'options': ['order_processing', 'user_verification', 'payment_handling']
                },
                'user_data': {
                    'type': 'object',
                    'required': True
                }
            },
            'outputs': {
                'result': {
                    'type': 'object',
                    'description': '业务处理结果'
                },
                'status': {
                    'type': 'string',
                    'description': '处理状态'
                }
            }
        }

class MultiToolAgent:
    """多工具集成智能体"""
    
    def __init__(self):
        self.tools = {
            'web_search': WebSearchTool(),
            'calculator': CalculatorTool(),
            'database_query': DatabaseQueryTool(),
            'email_sender': EmailSenderTool(),
            'file_processor': FileProcessorTool()
        }
        
        self.agent_config = {
            'strategy': 'function_calling',
            'max_iterations': 10,
            'temperature': 0.1,
            'model': 'gpt-4'
        }
    
    def create_agent_prompt(self) -> str:
        """创建Agent提示模板"""
        
        tools_description = []
        for tool_name, tool in self.tools.items():
            tools_description.append(f"""
{tool_name}:
- 描述: {tool.description}
- 参数: {tool.get_parameters_schema()}
- 使用场景: {tool.use_cases}
            """)
        
        return f"""
你是一个功能强大的AI助手，可以使用多种工具来帮助用户解决问题。

可用工具:
{chr(10).join(tools_description)}

使用原则:
1. 仔细分析用户需求，选择最合适的工具
2. 如果需要多个步骤，请逐步执行
3. 确保工具调用的参数正确
4. 基于工具结果提供准确的回答
5. 如果遇到错误，尝试其他方法或工具

请根据用户的问题，智能选择和使用工具。
        """
    
    def process_query(self, query: str, context: dict = None) -> dict:
        """处理用户查询"""
        
        # 1. 分析查询意图
        intent = self._analyze_intent(query)
        
        # 2. 选择合适的工具策略
        strategy = self._select_strategy(intent, context)
        
        # 3. 执行Agent推理
        result = self._execute_agent(query, strategy, context)
        
        return {
            'query': query,
            'intent': intent,
            'strategy': strategy,
            'result': result,
            'tools_used': result.get('tools_used', []),
            'execution_time': result.get('execution_time', 0)
        }
    
    def _analyze_intent(self, query: str) -> dict:
        """分析查询意图"""
        
        # 使用简单的关键词匹配或更复杂的意图识别
        intents = {
            'search': ['搜索', '查找', '找', 'search', 'find'],
            'calculate': ['计算', '算', 'calculate', 'compute'],
            'database': ['查询', '数据', 'query', 'data'],
            'email': ['发送', '邮件', 'email', 'send'],
            'file': ['文件', '处理', 'file', 'process']
        }
        
        detected_intents = []
        for intent, keywords in intents.items():
            if any(keyword in query.lower() for keyword in keywords):
                detected_intents.append(intent)
        
        return {
            'primary_intent': detected_intents[0] if detected_intents else 'general',
            'all_intents': detected_intents,
            'confidence': len(detected_intents) / len(intents)
        }
    
    def _select_strategy(self, intent: dict, context: dict) -> dict:
        """选择执行策略"""
        
        primary_intent = intent['primary_intent']
        
        strategies = {
            'search': {
                'preferred_tools': ['web_search'],
                'fallback_tools': ['database_query'],
                'max_iterations': 3
            },
            'calculate': {
                'preferred_tools': ['calculator'],
                'fallback_tools': [],
                'max_iterations': 2
            },
            'database': {
                'preferred_tools': ['database_query'],
                'fallback_tools': ['web_search'],
                'max_iterations': 5
            },
            'general': {
                'preferred_tools': list(self.tools.keys()),
                'fallback_tools': [],
                'max_iterations': 10
            }
        }
        
        return strategies.get(primary_intent, strategies['general'])

# 数据库查询优化示例
class OptimizedDataService:
    """优化的数据服务"""
    
    @staticmethod
    def get_conversations_with_pagination(
        user_id: str,
        page: int = 1,
        per_page: int = 20,
        app_id: str = None
    ) -> dict:
        """
        分页查询对话列表（优化版）
        """
        
        # 1. 构建基础查询
        query = db.session.query(Conversation).filter(
            Conversation.from_end_user_id == user_id
        )
        
        # 2. 添加应用过滤
        if app_id:
            query = query.filter(Conversation.app_id == app_id)
        
        # 3. 添加索引优化的排序
        query = query.order_by(Conversation.updated_at.desc())
        
        # 4. 执行分页查询
        pagination = query.paginate(
            page=page,
            per_page=per_page,
            error_out=False
        )
        
        # 5. 预加载关联数据（避免N+1查询）
        conversations = pagination.items
        if conversations:
            # 批量加载消息统计
            conversation_ids = [c.id for c in conversations]
            message_counts = db.session.query(
                Message.conversation_id,
                func.count(Message.id).label('count')
            ).filter(
                Message.conversation_id.in_(conversation_ids)
            ).group_by(Message.conversation_id).all()
            
            # 构建计数映射
            count_map = {item[0]: item[1] for item in message_counts}
            
            # 设置消息计数
            for conversation in conversations:
                conversation.message_count = count_map.get(conversation.id, 0)
        
        return {
            'conversations': [c.to_dict() for c in conversations],
            'pagination': {
                'page': page,
                'per_page': per_page,
                'total': pagination.total,
                'pages': pagination.pages,
                'has_next': pagination.has_next,
                'has_prev': pagination.has_prev
            }
        }
    
    @staticmethod
    @cached(ttl=300)  # 5分钟缓存
    def get_app_statistics(app_id: str) -> dict:
        """
        获取应用统计信息（带缓存）
        """
        
        # 使用单个查询获取所有统计信息
        stats = db.session.query(
            func.count(distinct(Conversation.id)).label('total_conversations'),
            func.count(distinct(Message.id)).label('total_messages'),
            func.count(distinct(Conversation.from_end_user_id)).label('unique_users'),
            func.avg(Message.provider_response_latency).label('avg_response_time')
        ).select_from(Conversation).join(
            Message, Conversation.id == Message.conversation_id
        ).filter(
            Conversation.app_id == app_id
        ).first()
        
        return {
            'total_conversations': stats.total_conversations or 0,
            'total_messages': stats.total_messages or 0,
            'unique_users': stats.unique_users or 0,
            'avg_response_time': float(stats.avg_response_time or 0)
        }

-- 数据库索引优化
-- 1. 复合索引优化查询
CREATE INDEX CONCURRENTLY idx_conversations_user_app_updated 
ON conversations (from_end_user_id, app_id, updated_at DESC);

-- 2. 部分索引优化存储
CREATE INDEX CONCURRENTLY idx_messages_active_conversations 
ON messages (conversation_id, created_at) 
WHERE status = 'normal';

-- 3. 表达式索引优化搜索
CREATE INDEX CONCURRENTLY idx_messages_content_search 
ON messages USING gin(to_tsvector('english', content));

-- 4. 唯一索引保证数据一致性
CREATE UNIQUE INDEX CONCURRENTLY idx_api_tokens_unique 
ON api_tokens (token) WHERE type = 'app';

class MultiLevelCache:
    """多级缓存实现"""
    
    def __init__(self):
        # L1: 本地内存缓存
        self.local_cache = {}
        self.local_cache_ttl = {}
        
        # L2: Redis缓存
        self.redis_client = redis.from_url(dify_config.REDIS_URL)
        
        # L3: 数据库查询缓存
        self.query_cache = {}
    
    def get(self, key: str, default=None):
        """多级缓存获取"""
        
        # 1. 尝试L1缓存
        if key in self.local_cache:
            if self._is_local_cache_valid(key):
                return self.local_cache[key]
            else:
                # 清理过期的本地缓存
                del self.local_cache[key]
                del self.local_cache_ttl[key]
        
        # 2. 尝试L2缓存
        try:
            value = self.redis_client.get(key)
            if value:
                # 反序列化
                data = json.loads(value)
                # 更新L1缓存
                self._set_local_cache(key, data, ttl=60)
                return data
        except Exception as e:
            logger.warning(f"Redis cache get failed: {e}")
        
        # 3. 返回默认值
        return default
    
    def set(self, key: str, value, ttl: int = 3600):
        """多级缓存设置"""
        
        # 1. 设置L1缓存
        self._set_local_cache(key, value, ttl=min(ttl, 300))  # 本地缓存最多5分钟
        
        # 2. 设置L2缓存
        try:
            serialized_value = json.dumps(value, cls=DateTimeEncoder)
            self.redis_client.setex(key, ttl, serialized_value)
        except Exception as e:
            logger.warning(f"Redis cache set failed: {e}")
    
    def delete(self, key: str):
        """多级缓存删除"""
        
        # 1. 删除L1缓存
        if key in self.local_cache:
            del self.local_cache[key]
            del self.local_cache_ttl[key]
        
        # 2. 删除L2缓存
        try:
            self.redis_client.delete(key)
        except Exception as e:
            logger.warning(f"Redis cache delete failed: {e}")
    
    def _set_local_cache(self, key: str, value, ttl: int):
        """设置本地缓存"""
        self.local_cache[key] = value
        self.local_cache_ttl[key] = time.time() + ttl
    
    def _is_local_cache_valid(self, key: str) -> bool:
        """检查本地缓存是否有效"""
        return (key in self.local_cache_ttl and 
                time.time() < self.local_cache_ttl[key])

# 缓存装饰器
def cached(ttl: int = 3600, key_prefix: str = ""):
    """缓存装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存键
            cache_key = f"{key_prefix}:{func.__name__}:{hash(str(args) + str(kwargs))}"
            
            # 尝试从缓存获取
            cached_result = cache.get(cache_key)
            if cached_result is not None:
                return cached_result
            
            # 执行函数
            result = func(*args, **kwargs)
            
            # 设置缓存
            cache.set(cache_key, result, ttl)
            
            return result
        return wrapper
    return decorator

# tasks/optimization.py
from celery import Celery
from celery.signals import task_prerun, task_postrun
import time

# Celery配置优化
celery_app = Celery('dify')
celery_app.conf.update(
    # 任务路由优化
    task_routes={
        'tasks.heavy_computation.*': {'queue': 'heavy'},
        'tasks.io_intensive.*': {'queue': 'io'},
        'tasks.quick_tasks.*': {'queue': 'quick'},
    },
    
    # 并发优化
    worker_concurrency=4,
    worker_prefetch_multiplier=1,
    
    # 结果后端优化
    result_backend='redis://localhost:6379/1',
    result_expires=3600,
    
    # 任务序列化优化
    task_serializer='json',
    result_serializer='json',
    accept_content=['json'],
    
    # 任务压缩
    task_compression='gzip',
    result_compression='gzip',
)

@celery_app.task(bind=True, max_retries=3)
def process_document_async(self, document_id: str, processing_config: dict):
    """
    异步文档处理任务（优化版）
    """
    
    try:
        # 1. 获取文档信息
        document = DocumentService.get_document(document_id)
        if not document:
            raise ValueError(f"Document {document_id} not found")
        
        # 2. 更新任务状态
        self.update_state(
            state='PROCESSING',
            meta={'progress': 0, 'stage': 'initializing'}
        )
        
        # 3. 分阶段处理
        stages = [
            ('parsing', DocumentParser.parse, 20),
            ('splitting', TextSplitter.split, 40),
            ('embedding', EmbeddingService.embed, 80),
            ('indexing', VectorStore.index, 100)
        ]
        
        results = {}
        for stage_name, stage_func, progress in stages:
            self.update_state(
                state='PROCESSING',
                meta={'progress': progress, 'stage': stage_name}
            )
            
            # 执行阶段处理
            stage_result = stage_func(document, processing_config)
            results[stage_name] = stage_result
            
            # 检查任务是否被取消
            if self.is_aborted():
                raise Ignore()
        
        # 4. 完成处理
        self.update_state(
            state='SUCCESS',
            meta={'progress': 100, 'stage': 'completed', 'results': results}
        )
        
        return results
        
    except Exception as exc:
        # 重试逻辑
        if self.request.retries < self.max_retries:
            # 指数退避重试
            countdown = 2 ** self.request.retries
            raise self.retry(exc=exc, countdown=countdown)
        else:
            # 最终失败
            self.update_state(
                state='FAILURE',
                meta={'error': str(exc), 'stage': 'failed'}
            )
            raise exc

# 任务监控
@task_prerun.connect
def task_prerun_handler(sender=None, task_id=None, task=None, args=None, kwargs=None, **kwds):
    """任务执行前钩子"""
    logger.info(f"Task {task.name}[{task_id}] started")

@task_postrun.connect
def task_postrun_handler(sender=None, task_id=None, task=None, args=None, kwargs=None, 
                        retval=None, state=None, **kwds):
    """任务执行后钩子"""
    logger.info(f"Task {task.name}[{task_id}] finished with state {state}")

# libs/security.py
import jwt
import hashlib
from functools import wraps
from flask import request, current_app
from werkzeug.security import check_password_hash

class SecurityManager:
    """安全管理器"""
    
    @staticmethod
    def generate_api_key(user_id: str, app_id: str) -> str:
        """生成API密钥"""
        
        # 使用安全随机数生成器
        import secrets
        random_part = secrets.token_urlsafe(32)
        
        # 组合用户和应用信息
        payload = f"{user_id}:{app_id}:{random_part}"
        
        # 生成哈希
        api_key = hashlib.sha256(payload.encode()).hexdigest()
        
        return f"dify-{api_key[:32]}"
    
    @staticmethod
    def validate_api_key(api_key: str) -> dict:
        """验证API密钥"""
        
        if not api_key or not api_key.startswith('dify-'):
            raise InvalidAPIKeyError("Invalid API key format")
        
        # 从数据库查询API密钥
        token_record = db.session.query(ApiToken).filter(
            ApiToken.token == api_key,
            ApiToken.type == 'app'
        ).first()
        
        if not token_record:
            raise InvalidAPIKeyError("API key not found")
        
        if not token_record.is_active:
            raise InvalidAPIKeyError("API key is disabled")
        
        # 检查过期时间
        if token_record.expires_at and token_record.expires_at < datetime.utcnow():
            raise InvalidAPIKeyError("API key has expired")
        
        return {
            'app_id': token_record.app_id,
            'user_id': token_record.created_by,
            'permissions': token_record.permissions or []
        }
    
    @staticmethod
    def check_rate_limit(user_id: str, endpoint: str, limit: int = 100, window: int = 3600) -> bool:
        """检查频率限制"""
        
        key = f"rate_limit:{user_id}:{endpoint}"
        current_time = int(time.time())
        window_start = current_time - window
        
        # 使用Redis滑动窗口算法
        pipe = redis_client.pipeline()
        pipe.zremrangebyscore(key, 0, window_start)
        pipe.zcard(key)
        pipe.zadd(key, {str(current_time): current_time})
        pipe.expire(key, window)
        
        results = pipe.execute()
        current_requests = results[1]
        
        return current_requests < limit

def require_api_key(f):
    """API密钥验证装饰器"""
    @wraps(f)
    def decorated_function(*args, **kwargs):
        # 获取API密钥
        api_key = request.headers.get('Authorization')
        if api_key and api_key.startswith('Bearer '):
            api_key = api_key[7:]
        else:
            return {'error': 'Missing API key'}, 401
        
        try:
            # 验证API密钥
            key_info = SecurityManager.validate_api_key(api_key)
            
            # 检查频率限制
            if not SecurityManager.check_rate_limit(
                key_info['user_id'], 
                request.endpoint
            ):
                return {'error': 'Rate limit exceeded'}, 429
            
            # 注入密钥信息到请求上下文
            request.api_key_info = key_info
            
            return f(*args, **kwargs)
            
        except InvalidAPIKeyError as e:
            return {'error': str(e)}, 401
        except Exception as e:
            logger.exception(f"API key validation error: {e}")
            return {'error': 'Authentication failed'}, 500
    
    return decorated_function

# libs/validation.py
import re
import bleach
from marshmallow import Schema, fields, validate, ValidationError

class InputValidator:
    """输入验证器"""
    
    # 危险模式列表
    DANGEROUS_PATTERNS = [
        r'<script[^>]*>.*?</script>',  # XSS脚本
        r'javascript:',                # JavaScript协议
        r'on\w+\s*=',                 # 事件处理器
        r'expression\s*\(',           # CSS表达式
        r'import\s+os',               # Python导入
        r'exec\s*\(',                 # 代码执行
        r'eval\s*\(',                 # 代码评估
    ]
    
    @classmethod
    def sanitize_html(cls, content: str) -> str:
        """净化HTML内容"""
        
        # 允许的标签和属性
        allowed_tags = ['p', 'br', 'strong', 'em', 'u', 'ol', 'ul', 'li', 'a', 'img']
        allowed_attributes = {
            'a': ['href', 'title'],
            'img': ['src', 'alt', 'width', 'height']
        }
        
        # 使用bleach净化
        clean_content = bleach.clean(
            content,
            tags=allowed_tags,
            attributes=allowed_attributes,
            strip=True
        )
        
        return clean_content
    
    @classmethod
    def validate_user_input(cls, content: str, max_length: int = 10000) -> str:
        """验证用户输入"""
        
        if not content:
            raise ValidationError("Content cannot be empty")
        
        if len(content) > max_length:
            raise ValidationError(f"Content exceeds maximum length of {max_length}")
        
        # 检查危险模式
        for pattern in cls.DANGEROUS_PATTERNS:
            if re.search(pattern, content, re.IGNORECASE):
                raise ValidationError("Content contains potentially dangerous code")
        
        # 净化内容
        sanitized_content = cls.sanitize_html(content)
        
        return sanitized_content
    
    @classmethod
    def validate_file_upload(cls, file_data: bytes, filename: str, allowed_types: list) -> bool:
        """验证文件上传"""
        
        # 检查文件大小
        if len(file_data) > 10 * 1024 * 1024:  # 10MB
            raise ValidationError("File size exceeds 10MB limit")
        
        # 检查文件类型
        import magic
        file_type = magic.from_buffer(file_data, mime=True)
        
        if file_type not in allowed_types:
            raise ValidationError(f"File type {file_type} not allowed")
        
        # 检查文件名
        if not re.match(r'^[a-zA-Z0-9._-]+$', filename):
            raise ValidationError("Invalid filename")
        
        return True

# Marshmallow模式定义
class ChatMessageSchema(Schema):
    """聊天消息验证模式"""
    
    inputs = fields.Dict(
        required=True,
        validate=validate.Length(max=100)
    )
    
    query = fields.Str(
        required=True,
        validate=[
            validate.Length(min=1, max=4000),
            lambda x: InputValidator.validate_user_input(x, 4000)
        ]
    )
    
    response_mode = fields.Str(
        validate=validate.OneOf(['blocking', 'streaming']),
        missing='streaming'
    )
    
    conversation_id = fields.Str(
        validate=validate.Regexp(r'^[a-f0-9-]{36}$'),
        allow_none=True
    )
    
    user = fields.Str(
        required=True,
        validate=[
            validate.Length(min=1, max=255),
            validate.Regexp(r'^[a-zA-Z0-9_-]+$')
        ]
    )
    
    files = fields.List(
        fields.Dict(),
        validate=validate.Length(max=10),
        missing=[]
    )