VDAR-Router：基于查询难度分析的LLM智能路由方案设计与实现

在实际大语言模型应用开发中直接调用单一模型处理所有用户查询往往不是最优选择。不同模型在成本、响应速度、专业领域和推理能力上存在显著差异而用户查询的复杂度也千差万别。VDAR-Router 提出了一种基于查询难度分析的智能路由方案通过语言化分析查询难度结合检索机制实现将不同复杂度的查询动态分配到最合适的 LLM 上执行。这种方案的核心价值在于既避免了用昂贵的高性能模型处理简单问题造成的资源浪费也防止了能力有限的轻量模型无法胜任复杂任务的情况。对于需要平衡成本、响应时间和准确性的生产系统来说这种自适应路由机制能够显著提升整体效率。1. 理解 VDAR-Router 的核心工作机制VDAR-Router 的工作流程可以分解为三个关键阶段查询难度分析、候选模型检索和最终路由决策。每个阶段都有明确的技术目标和实现逻辑。1.1 查询难度分析的语言化表达传统的查询难度评估通常依赖于数值评分或分类标签但 VDAR-Router 采用了语言化分析的方式。这种方法不是简单输出简单或复杂的二分判断而是生成一段自然语言描述详细说明查询的难点所在。例如面对查询请解释量子纠缠的基本原理并举例说明其在量子通信中的应用系统可能生成如下难度分析该查询涉及多个知识层面需要先解释基础物理概念然后建立概念与实际应用的连接最后还要提供具体案例。回答需要确保准确性、连贯性和易懂性的平衡。这种语言化分析的优势在于为后续的模型匹配提供更丰富的语义信息分析过程本身可以作为可解释性输出帮助理解路由决策比简单数值评分更能捕捉查询的细微复杂度差异1.2 基于检索的候选模型匹配VDAR-Router 维护一个模型能力数据库其中存储了各个候选 LLM 的详细能力描述。这些描述同样采用自然语言形式例如模型A擅长处理基础概念解释类问题响应速度快成本低模型B具备深度推理能力适合处理多步骤复杂问题但响应较慢模型C在特定专业领域如法律、医疗有专门优化当收到查询难度分析后系统会通过语义检索技术从模型库中找出能力描述与查询难度最匹配的候选模型。这个过程不是简单的关键词匹配而是基于嵌入向量的相似度计算。1.3 综合考虑多因素的路由决策最终的路由决策需要平衡多个因素不仅仅是语义匹配度。VDAR-Router 通常会考虑语义匹配分数查询难度与模型能力的匹配程度成本约束当前可用的预算限制响应时间要求用户对延迟的敏感度当前负载各模型实例的实时负载情况历史表现该模型处理类似查询的成功率这些因素通过加权评分算法综合计算选出最优的目标模型。整个决策过程可以配置不同的策略权重适应不同的业务场景需求。2. 构建基础的 VDAR-Router 原型系统为了深入理解 VDAR-Router 的工作原理我们构建一个简化但完整可用的原型系统。这个原型包含核心的路由逻辑可以作为实际项目开发的基础。2.1 环境准备与依赖配置首先需要准备 Python 环境和支持的库依赖。建议使用 Python 3.8 版本主要依赖包括# requirements.txt openai1.0.0 numpy1.21.0 scikit-learn1.0.0 sentence-transformers2.2.0 fastapi0.100.0 uvicorn0.20.0 pydantic2.0.0安装命令pip install -r requirements.txt对于嵌入模型我们使用轻量级的all-MiniLM-L6-v2它在性能和资源消耗之间提供了良好平衡。生产环境可以考虑使用更大的模型获得更好的语义理解能力。2.2 项目结构与核心类设计创建以下项目结构vdar_router/ ├── __init__.py ├── models/ │ ├── __init__.py │ ├── model_registry.py # 模型注册与管理 │ └── capability_db.py # 能力数据库 ├── analysis/ │ ├── __init__.py │ └── difficulty_analyzer.py # 查询难度分析 ├── retrieval/ │ ├── __init__.py │ └── model_matcher.py # 模型匹配检索 ├── router/ │ ├── __init__.py │ └── decision_engine.py # 路由决策引擎 └── api/ ├── __init__.py └── router_api.py # API 接口层核心数据模型定义# models/model_registry.py from pydantic import BaseModel from typing import List, Dict, Optional from enum import Enum class ModelCapability(BaseModel): model_id: str capability_description: str cost_per_token: float avg_response_time: float max_tokens: int supported_domains: List[str] embedding: Optional[List[float]] None class QueryDifficulty(BaseModel): original_query: str verbalized_analysis: str complexity_score: float required_domains: List[str] estimated_token_count: int class RoutingDecision(BaseModel): selected_model: str confidence_score: float decision_reason: str fallback_model: Optional[str] None2.3 实现查询难度分析器难度分析器是系统的第一个关键组件负责将原始查询转化为结构化的难度分析。# analysis/difficulty_analyzer.py import openai from sentence_transformers import SentenceTransformer from models.model_registry import QueryDifficulty import re class DifficultyAnalyzer: def __init__(self, embedding_model_nameall-MiniLM-L6-v2): self.embedding_model SentenceTransformer(embedding_model_name) # 预定义的难度分析提示模板 self.analysis_prompt 请分析以下查询的难度从以下维度进行语言化描述 1. 知识深度要求 2. 推理复杂度 3. 回答结构要求 4. 潜在的专业领域需求 查询{query} 请用自然语言给出综合分析不要使用评分或等级标签。 def analyze_query(self, query: str) - QueryDifficulty: # 估算token数量简化版本 token_estimate len(query.split()) * 1.3 # 使用嵌入模型获取查询的语义向量 query_embedding self.embedding_model.encode([query])[0] # 这里简化处理实际应该调用LLM生成详细分析 verbalized_analysis self._generate_verbalized_analysis(query) # 基于分析文本计算复杂度分数 complexity_score self._calculate_complexity_score(verbalized_analysis) # 识别可能涉及的专业领域 domains self._identify_domains(query) return QueryDifficulty( original_queryquery, verbalized_analysisverbalized_analysis, complexity_scorecomplexity_score, required_domainsdomains, estimated_token_countint(token_estimate) ) def _generate_verbalized_analysis(self, query: str) - str: # 简化实现实际项目中应该调用配置的LLM if len(query.split()) 10: return 查询相对简单涉及基础概念或事实性信息需要直接明确的回答。 else: return 查询涉及多个概念或要求多步骤推理需要深入分析和结构化回答。 def _calculate_complexity_score(self, analysis: str) - float: # 基于分析文本的长度和关键词计算复杂度 length_factor min(len(analysis) / 100, 1.0) complexity_keywords [深入, 多步骤, 复杂, 分析, 推理] keyword_count sum(1 for keyword in complexity_keywords if keyword in analysis) keyword_factor min(keyword_count / len(complexity_keywords), 1.0) return (length_factor keyword_factor) / 2 def _identify_domains(self, query: str) - List[str]: domain_keywords { 技术: [编程, 代码, 算法, 系统, 软件], 学术: [研究, 理论, 论文, 实验, 学术], 商业: [市场, 营销, 战略, 财务, 商业], 生活: [日常, 生活, 家庭, 健康, 娱乐] } domains [] for domain, keywords in domain_keywords.items(): if any(keyword in query for keyword in keywords): domains.append(domain) return domains if domains else [通用]3. 构建模型能力数据库与匹配引擎模型能力数据库存储所有可用LLM的详细信息匹配引擎负责找到与查询难度最契合的候选模型。3.1 模型能力数据库实现# models/capability_db.py import json from typing import List, Dict from sentence_transformers import SentenceTransformer from models.model_registry import ModelCapability class CapabilityDatabase: def __init__(self): self.models: Dict[str, ModelCapability] {} self.embedding_model SentenceTransformer(all-MiniLM-L6-v2) self._initialize_sample_models() def _initialize_sample_models(self): # 示例模型配置实际项目应从配置文件或数据库加载 sample_models [ { model_id: gpt-3.5-turbo, capability_description: 适合处理中等复杂度的通用问题平衡成本与性能响应速度快, cost_per_token: 0.002, avg_response_time: 2.5, max_tokens: 4096, supported_domains: [通用, 技术, 商业, 生活] }, { model_id: gpt-4, capability_description: 处理高度复杂的推理任务多步骤问题解决深度分析能力强, cost_per_token: 0.06, avg_response_time: 8.0, max_tokens: 8192, supported_domains: [通用, 技术, 学术, 商业] }, { model_id: claude-instant, capability_description: 快速响应简单查询成本效益高适合事实性问答和基础任务, cost_per_token: 0.00163, avg_response_time: 1.8, max_tokens: 4096, supported_domains: [通用, 生活, 商业] } ] for model_data in sample_models: capability ModelCapability(**model_data) # 为每个模型的能力描述生成嵌入向量 capability.embedding self.embedding_model.encode( [capability.capability_description] )[0].tolist() self.models[capability.model_id] capability def get_all_models(self) - List[ModelCapability]: return list(self.models.values()) def get_model(self, model_id: str) - ModelCapability: return self.models.get(model_id) def add_model(self, capability: ModelCapability): if capability.embedding is None: capability.embedding self.embedding_model.encode( [capability.capability_description] )[0].tolist() self.models[capability.model_id] capability3.2 基于语义相似度的模型匹配# retrieval/model_matcher.py import numpy as np from sklearn.metrics.pairwise import cosine_similarity from typing import List, Tuple from models.model_registry import ModelCapability, QueryDifficulty from sentence_transformers import SentenceTransformer class ModelMatcher: def __init__(self, capability_db): self.capability_db capability_db self.embedding_model SentenceTransformer(all-MiniLM-L6-v2) def find_best_matches(self, difficulty: QueryDifficulty, top_k: int 3) - List[Tuple[ModelCapability, float]]: # 将查询难度分析转换为嵌入向量 query_embedding self.embedding_model.encode([difficulty.verbalized_analysis])[0] models self.capability_db.get_all_models() similarities [] for model in models: if model.embedding is not None: # 计算语义相似度 semantic_similarity cosine_similarity( [query_embedding], [model.embedding] )[0][0] # 领域匹配度 domain_overlap self._calculate_domain_overlap( difficulty.required_domains, model.supported_domains ) # 综合评分 combined_score (semantic_similarity * 0.6 domain_overlap * 0.4) similarities.append((model, combined_score)) # 按综合评分排序并返回top_k similarities.sort(keylambda x: x[1], reverseTrue) return similarities[:top_k] def _calculate_domain_overlap(self, query_domains: List[str], model_domains: List[str]) - float: if not query_domains or not model_domains: return 0.0 intersection set(query_domains) set(model_domains) union set(query_domains) | set(model_domains) return len(intersection) / len(union) if union else 0.04. 实现综合路由决策引擎决策引擎需要综合考虑语义匹配、成本约束、性能要求等多个因素做出最终的路由选择。4.1 路由决策引擎实现# router/decision_engine.py from typing import List, Tuple, Optional from models.model_registry import QueryDifficulty, RoutingDecision, ModelCapability from retrieval.model_matcher import ModelMatcher class DecisionEngine: def __init__(self, capability_db, matcher: ModelMatcher): self.capability_db capability_db self.matcher matcher # 配置决策权重 self.weights { semantic_match: 0.4, cost_efficiency: 0.25, performance: 0.2, domain_specialization: 0.15 } def make_routing_decision(self, difficulty: QueryDifficulty, budget_constraint: Optional[float] None, max_response_time: Optional[float] None) - RoutingDecision: # 获取候选模型 candidates self.matcher.find_best_matches(difficulty, top_k5) if not candidates: return self._get_fallback_decision() scored_candidates [] for model, base_score in candidates: # 应用约束过滤 if not self._satisfies_constraints(model, difficulty, budget_constraint, max_response_time): continue # 计算综合得分 final_score self._calculate_comprehensive_score(model, difficulty, base_score) scored_candidates.append((model, final_score)) if not scored_candidates: return self._get_fallback_decision() # 选择得分最高的模型 best_model, best_score max(scored_candidates, keylambda x: x[1]) fallback_model self._select_fallback_model(scored_candidates, best_model.model_id) return RoutingDecision( selected_modelbest_model.model_id, confidence_scorebest_score, decision_reasonself._generate_decision_reason(best_model, difficulty, best_score), fallback_modelfallback_model ) def _satisfies_constraints(self, model: ModelCapability, difficulty: QueryDifficulty, budget_constraint: Optional[float], max_response_time: Optional[float]) - bool: # 估算成本 estimated_cost model.cost_per_token * difficulty.estimated_token_count if budget_constraint and estimated_cost budget_constraint: return False if max_response_time and model.avg_response_time max_response_time: return False # 检查token限制 if difficulty.estimated_token_count model.max_tokens * 0.8: # 保留20%余量 return False return True def _calculate_comprehensive_score(self, model: ModelCapability, difficulty: QueryDifficulty, base_score: float) - float: # 成本效率得分成本越低得分越高 cost_score 1.0 / (model.cost_per_token 0.001) # 避免除零 normalized_cost_score min(cost_score / 1000, 1.0) # 归一化 # 性能得分响应时间越短得分越高 performance_score 1.0 / (model.avg_response_time 0.1) normalized_performance_score min(performance_score, 1.0) # 领域专业化得分 domain_score len(set(difficulty.required_domains) set(model.supported_domains)) / max( len(set(difficulty.required_domains)), 1 ) # 加权综合得分 comprehensive_score ( base_score * self.weights[semantic_match] normalized_cost_score * self.weights[cost_efficiency] normalized_performance_score * self.weights[performance] domain_score * self.weights[domain_specialization] ) return comprehensive_score def _select_fallback_model(self, candidates: List[Tuple[ModelCapability, float]], selected_model_id: str) - Optional[str]: # 选择得分第二高的不同模型作为备选 other_models [(model, score) for model, score in candidates if model.model_id ! selected_model_id] if other_models: fallback_model, _ max(other_models, keylambda x: x[1]) return fallback_model.model_id return None def _generate_decision_reason(self, model: ModelCapability, difficulty: QueryDifficulty, score: float) - str: reasons [] if score 0.8: reasons.append(查询复杂度与模型能力高度匹配) elif score 0.6: reasons.append(模型能力适合处理此类查询) else: reasons.append(在约束条件下选择相对合适的模型) if model.cost_per_token 0.01: reasons.append(成本效益优良) if model.avg_response_time 3.0: reasons.append(响应性能良好) return ; .join(reasons) def _get_fallback_decision(self) - RoutingDecision: # 默认回退到成本最低的可用模型 all_models self.capability_db.get_all_models() if all_models: fallback_model min(all_models, keylambda x: x.cost_per_token) return RoutingDecision( selected_modelfallback_model.model_id, confidence_score0.1, decision_reason无合适匹配选择成本最低的默认模型, fallback_modelNone ) else: raise ValueError(没有可用的模型配置)4.2 完整的路由服务集成# api/router_api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional from models.model_registry import QueryDifficulty, RoutingDecision from analysis.difficulty_analyzer import DifficultyAnalyzer from models.capability_db import CapabilityDatabase from retrieval.model_matcher import ModelMatcher from router.decision_engine import DecisionEngine app FastAPI(titleVDAR-Router API, version1.0.0) # 初始化组件 capability_db CapabilityDatabase() difficulty_analyzer DifficultyAnalyzer() model_matcher ModelMatcher(capability_db) decision_engine DecisionEngine(capability_db, model_matcher) class RoutingRequest(BaseModel): query: str max_budget: Optional[float] None max_response_time: Optional[float] None user_preferences: Optional[dict] None class RoutingResponse(BaseModel): decision: RoutingDecision difficulty_analysis: QueryDifficulty app.post(/route, response_modelRoutingResponse) async def route_query(request: RoutingRequest): try: # 1. 分析查询难度 difficulty difficulty_analyzer.analyze_query(request.query) # 2. 做出路由决策 decision decision_engine.make_routing_decision( difficulty, budget_constraintrequest.max_budget, max_response_timerequest.max_response_time ) return RoutingResponse( decisiondecision, difficulty_analysisdifficulty ) except Exception as e: raise HTTPException(status_code500, detailf路由处理失败: {str(e)}) app.get(/models) async def list_available_models(): models capability_db.get_all_models() return {models: [model.model_id for model in models]} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)5. 系统测试与验证方法构建完成后需要系统性地测试路由器的各项功能确保其在不同场景下都能做出合理决策。5.1 测试用例设计与执行创建测试脚本验证系统功能# test_router.py import asyncio from api.router_api import RoutingRequest, route_query async def test_router(): test_cases [ { name: 简单事实查询, query: 法国的首都是哪里, expected_model: claude-instant # 期望选择成本低的模型 }, { name: 复杂推理问题, query: 请比较机器学习中监督学习和无监督学习的优缺点并举例说明各自适用场景, expected_model: gpt-4 # 期望选择能力强的模型 }, { name: 成本约束测试, query: 需要详细分析当前人工智能技术的发展趋势和未来展望, max_budget: 0.01, # 设置较低预算 expected_model: gpt-3.5-turbo # 期望在预算内选择 } ] for test_case in test_cases: print(f\n测试用例: {test_case[name]}) print(f查询: {test_case[query]}) request RoutingRequest( querytest_case[query], max_budgettest_case.get(max_budget) ) # 这里简化调用实际应该通过HTTP调用API response await route_query(request) print(f难度分析: {response.difficulty_analysis.verbalized_analysis}) print(f选择模型: {response.decision.selected_model}) print(f置信度: {response.decision.confidence_score:.2f}) print(f决策理由: {response.decision.decision_reason}) if test_case.get(expected_model): if response.decision.selected_model test_case[expected_model]: print(✅ 测试通过) else: print(f❌ 测试失败期望 {test_case[expected_model]}) if __name__ __main__: asyncio.run(test_router())5.2 性能与准确性评估指标建立系统的评估体系监控路由决策的质量# evaluation/metrics_calculator.py import time from typing import List, Dict from dataclasses import dataclass dataclass class RoutingMetrics: decision_latency: float # 决策耗时 cost_savings: float # 相比总是使用最强模型的成本节省 accuracy_score: float # 人工评估的路由准确性 fallback_rate: float # 回退到默认模型的比例 class MetricsCalculator: def __init__(self): self.history: List[Dict] [] def record_decision(self, query: str, selected_model: str, ideal_model: str, actual_cost: float, max_model_cost: float, latency: float): record { timestamp: time.time(), query: query, selected_model: selected_model, ideal_model: ideal_model, cost_savings: max_model_cost - actual_cost, latency: latency, is_correct: selected_model ideal_model } self.history.append(record) def calculate_metrics(self, time_window: int 3600) - RoutingMetrics: # 计算指定时间窗口内的指标 window_start time.time() - time_window recent_records [r for r in self.history if r[timestamp] window_start] if not recent_records: return RoutingMetrics(0, 0, 0, 0) total_latency sum(r[latency] for r in recent_records) total_savings sum(r[cost_savings] for r in recent_records) correct_decisions sum(1 for r in recent_records if r[is_correct]) fallback_count sum(1 for r in recent_records if fallback in r[selected_model]) return RoutingMetrics( decision_latencytotal_latency / len(recent_records), cost_savingstotal_savings, accuracy_scorecorrect_decisions / len(recent_records), fallback_ratefallback_count / len(recent_records) )6. 生产环境部署与优化建议将 VDAR-Router 部署到生产环境需要考虑更多工程化因素确保系统的可靠性、可扩展性和可维护性。6.1 配置管理与外部化生产环境应该将配置外部化支持动态更新# config/production.yaml router: weights: semantic_match: 0.4 cost_efficiency: 0.25 performance: 0.2 domain_specialization: 0.15 constraints: max_decision_latency: 1.0 # 最大决策耗时秒 min_confidence_threshold: 0.3 # 最低置信度阈值 models: refresh_interval: 300 # 模型配置刷新间隔秒 logging: level: INFO format: %(asctime)s - %(name)s - %(levelname)s - %(message)s monitoring: enabled: true metrics_port: 9090 health_check_interval: 306.2 缓存策略优化为提升性能实现多级缓存机制# optimization/query_cache.py import time from typing import Optional import hashlib class QueryDifficultyCache: def __init__(self, max_size: int 10000, ttl: int 3600): self.cache {} self.max_size max_size self.ttl ttl def get_cache_key(self, query: str) - str: # 使用查询内容的哈希作为缓存键 return hashlib.md5(query.encode()).hexdigest() def get(self, query: str) - Optional[QueryDifficulty]: key self.get_cache_key(query) if key in self.cache: entry self.cache[key] if time.time() - entry[timestamp] self.ttl: return entry[difficulty] else: # 缓存过期删除条目 del self.cache[key] return None def set(self, query: str, difficulty: QueryDifficulty): key self.get_cache_key(query) # 如果缓存已满删除最旧的条目 if len(self.cache) self.max_size: oldest_key min(self.cache.keys(), keylambda k: self.cache[k][timestamp]) del self.cache[oldest_key] self.cache[key] { difficulty: difficulty, timestamp: time.time() }6.3 监控与告警配置建立完整的监控体系及时发现和处理问题# monitoring/health_check.py import psutil import time from typing import Dict class SystemHealthMonitor: def __init__(self): self.start_time time.time() def get_system_metrics(self) - Dict: return { uptime: time.time() - self.start_time, cpu_percent: psutil.cpu_percent(interval1), memory_percent: psutil.virtual_memory().percent, disk_usage: psutil.disk_usage(/).percent, active_connections: len(psutil.net_connections()) } def check_health(self) - Dict: metrics self.get_system_metrics() health_status healthy issues [] if metrics[cpu_percent] 80: issues.append(CPU使用率过高) health_status degraded if metrics[memory_percent] 85: issues.append(内存使用率过高) health_status degraded if metrics[disk_usage] 90: issues.append(磁盘空间不足) health_status critical return { status: health_status, metrics: metrics, issues: issues, timestamp: time.time() }7. 常见问题排查与解决方案在实际运行过程中可能会遇到各种问题以下是典型问题的排查路径。7.1 路由决策质量问题排查问题现象可能原因检查方式解决方案简单查询被路由到昂贵模型难度分析过度复杂化检查难度分析输出日志调整分析提示词增加简单模式识别复杂查询选择能力不足模型模型能力描述不准确验证模型能力向量质量重新生成模型能力嵌入优化描述文本路由置信度持续偏低语义匹配效果差检查嵌入模型质量升级嵌入模型增加训练数据多样性决策延迟过高嵌入计算或检索耗时分析各阶段性能日志引入缓存优化检索算法考虑近似匹配7.2 系统性能问题排查# troubleshooting/performance_profiler.py import time import cProfile import pstats from io import StringIO class RouterProfiler: def __init__(self): self.profiler cProfile.Profile() def profile_route_decision(self, query: str): 分析单次路由决策的性能瓶颈 self.profiler.enable() start_time time.time() # 执行路由决策流程 difficulty self.difficulty_analyzer.analyze_query(query) decision self.decision_engine.make_routing_decision(difficulty) end_time time.time() self.profiler.disable() # 生成性能报告 s StringIO() ps pstats.Stats(self.profiler, streams).sort_stats(cumulative) ps.print_stats() return { total_time: end_time - start_time, profile_report: s.getvalue(), decision: decision }7.3 模型配置管理问题生产环境中模型配置需要版本控制和回滚机制# management/model_config_manager.py import json from typing import List, Dict from datetime import datetime class ModelConfigManager: def __init__(self, config_file: str): self.config_file config_file self.version_history: List[Dict] [] self.load_config() def load_config(self): with open(self.config_file, r) as f: self.current_config json.load(f) self.version_history.append({ timestamp: datetime.now(), config: self.current_config.copy(), version: len(self.version_history) 1 }) def update_model_config(self, model_id: str, updates: Dict): # 创建配置备份 backup self.current_config.copy() try: if model_id in self.current_config[models]: self.current_config[models][model_id].update(updates) # 验证新配置 self._validate_config() # 保存更新 self._save_config() # 记录版本历史 self.version_history.append({ timestamp: datetime.now(), config: self.current_config.copy(), version: len(self.version_history) 1, changes: {model_id: updates} }) else: raise ValueError(f模型 {model_id} 不存在) except Exception as e: # 配置更新失败回滚到备份 self.current_config backup raise e def rollback_config(self, version: int): 回滚到指定版本配置 if 1 version len(self.version_history): self.current_config self.version_history[version-1][config].copy() self._save_config()VDAR-Router 的实现展示了如何将复杂的LLM路由问题分解为可管理的组件通过查询难度分析、语义匹配和多因素决策实现智能化的模型选择。在实际项目中还需要根据具体业务需求调整权重参数、优化匹配算法并建立完善的监控运维体系。这种架构不仅适用于LLM路由也可以扩展到其他类型的服务路由场景为构建高效可靠的AI应用基础设施提供重要参考。