企业官网建设流程全解析

最近AI圈有个新玩法开始流行：让不同的大语言模型像接力赛一样协作完成复杂任务，然后把这个过程直播出来。你可能在Twitch或其他平台看到过"GPT-5.6与Claude Opus 5接力爬塔"这样的直播标题，但真正值得关注的不是娱乐效果，而是这种模式背后揭示的AI协作开发新范式。

作为一名长期关注AI工程化的开发者，我发现这种"AI接力"直播实际上是一个极佳的技术演示场。它不仅仅是为了展示哪个模型更强，更重要的是验证了多模型协作在实际项目中的可行性。如果你正在考虑如何将AI集成到自己的开发流程中，这种模式可能比单纯依赖单一模型更有价值。

本文将从一个技术实践者的角度，拆解这种AI接力直播的技术实现，并分享如何在自己的项目中应用类似的多模型协作策略。无论你是想优化现有的AI工作流，还是探索新的自动化方案，都能从中获得实用的技术洞察。

1. 为什么AI接力模式值得开发者关注

传统的AI应用大多采用单一模型架构：一个问题，一个模型，一个答案。但这种模式在面对复杂任务时存在明显局限——没有哪个模型在所有领域都表现完美。GPT系列可能在创意生成上更出色，Claude在逻辑推理上更严谨，而专门的小模型在特定任务上可能效率更高。

AI接力模式的核心价值在于任务分解与专业化分工。就像软件开发中的微服务架构，不同的模型负责自己最擅长的子任务，通过良好的接口设计实现无缝协作。在"爬塔"这个隐喻中，每一层代表一个难度递增的子任务，不同的模型根据自己的优势处理对应的层级。

从工程角度看，这种模式解决了几个关键问题：

• 能力边界问题：单个模型的知识盲区可以通过其他模型弥补
• 成本优化：不需要为简单任务使用昂贵的大模型
• 稳定性提升：一个模型的失败不会导致整个流程中断
• 可解释性增强：每个决策步骤都有明确的责任模型

在实际项目中，这意味着你可以用相对较小的成本构建出更强大的AI系统。比如让Claude负责需求分析和架构设计，GPT处理代码生成，专用模型进行代码审查和测试。

2. AI接力直播的技术架构解析

一个完整的AI接力直播系统涉及多个技术组件，理解这个架构是实现自定义方案的基础。

2.1 核心组件构成

典型的AI接力系统包含以下模块：

• 任务调度器：负责解析复杂任务，将其分解为子任务，并决定由哪个模型处理
• 模型接口层：统一不同AI模型的API调用，提供标准化的输入输出格式
• 状态管理器：跟踪任务执行进度，管理上下文传递 between models
• 直播集成模块：将执行过程可视化，并处理与直播平台的交互
• 错误处理与回退机制：当某个模型失败时，提供备选方案

2.2 数据流设计

用户输入 → 任务分解 → 模型A处理 → 结果评估 → 模型B处理 → 最终输出

关键设计考虑：

• 上下文如何在模型间传递
• 如何评估每个步骤的执行质量
• 什么时候应该切换模型
• 如何保证整个流程的连贯性

3. 环境准备与工具选型

要实现类似的AI接力系统，你需要准备以下环境和技术栈。

3.1 基础环境要求

# 推荐使用Python 3.8+环境 python --version # Python 3.8.10 # 必要的核心库 pip install openai anthropic-apis requests websocket-client

3.2 模型API配置

你需要准备相应模型的API访问权限：

# config.py - API配置管理 import os class APIConfig: # OpenAI GPT系列配置 OPENAI_API_KEY = os.getenv('OPENAI_API_KEY', 'your-openai-key') OPENAI_BASE_URL = os.getenv('OPENAI_BASE_URL', 'https://api.openai.com/v1') # Anthropic Claude配置 ANTHROPIC_API_KEY = os.getenv('ANTHROPIC_API_KEY', 'your-anthropic-key') # 其他模型配置...

3.3 开发工具建议

• IDE：VS Code with Python扩展
• 版本控制：Git
• API测试：Postman或curl
• 日志管理：Loguru或标准logging模块

4. 核心实现：多模型协作引擎

让我们从最核心的多模型协作引擎开始实现。

4.1 基础模型接口抽象

首先定义一个统一的模型接口，确保不同模型可以互换使用：

# models/base_model.py from abc import ABC, abstractmethod from typing import Dict, Any, List class BaseAIModel(ABC): """AI模型基础抽象类""" def __init__(self, model_name: str, api_key: str): self.model_name = model_name self.api_key = api_key self.client = None self._initialize_client() @abstractmethod def _initialize_client(self): """初始化模型客户端""" pass @abstractmethod async def generate(self, prompt: str, **kwargs) -> Dict[str, Any]: """生成回复""" pass @abstractmethod def estimate_cost(self, prompt_tokens: int, completion_tokens: int) -> float: """估算请求成本""" pass def validate_response(self, response: Dict[str, Any]) -> bool: """验证响应格式和内容""" required_fields = ['content', 'usage', 'model'] return all(field in response for field in required_fields)

4.2 GPT模型实现

# models/gpt_model.py import openai from .base_model import BaseAIModel class GPTModel(BaseAIModel): """OpenAI GPT系列模型实现""" def _initialize_client(self): self.client = openai.OpenAI(api_key=self.api_key) async def generate(self, prompt: str, **kwargs) -> Dict[str, Any]: try: response = self.client.chat.completions.create( model=self.model_name, messages=[{"role": "user", "content": prompt}], **kwargs ) return { 'content': response.choices[0].message.content, 'usage': { 'prompt_tokens': response.usage.prompt_tokens, 'completion_tokens': response.usage.completion_tokens, 'total_tokens': response.usage.total_tokens }, 'model': self.model_name } except Exception as e: return self._handle_error(e) def estimate_cost(self, prompt_tokens: int, completion_tokens: int) -> float: # GPT-4 Turbo定价示例（每1000个token） pricing = { 'gpt-4-turbo': {'input': 0.01, 'output': 0.03}, 'gpt-3.5-turbo': {'input': 0.001, 'output': 0.002} } model_pricing = pricing.get(self.model_name, pricing['gpt-3.5-turbo']) cost = (prompt_tokens / 1000 * model_pricing['input'] + completion_tokens / 1000 * model_pricing['output']) return cost def _handle_error(self, error: Exception) -> Dict[str, Any]: return { 'content': f"Error: {str(error)}", 'usage': {'prompt_tokens': 0, 'completion_tokens': 0, 'total_tokens': 0}, 'model': self.model_name, 'error': True }

4.3 Claude模型实现

# models/claude_model.py import anthropic from .base_model import BaseAIModel class ClaudeModel(BaseAIModel): """Anthropic Claude模型实现""" def _initialize_client(self): self.client = anthropic.Anthropic(api_key=self.api_key) async def generate(self, prompt: str, **kwargs) -> Dict[str, Any]: try: response = self.client.messages.create( model=self.model_name, max_tokens=4096, messages=[{"role": "user", "content": prompt}], **kwargs ) return { 'content': response.content[0].text, 'usage': { 'prompt_tokens': response.usage.input_tokens, 'completion_tokens': response.usage.output_tokens, 'total_tokens': response.usage.input_tokens + response.usage.output_tokens }, 'model': self.model_name } except Exception as e: return self._handle_error(e) def estimate_cost(self, prompt_tokens: int, completion_tokens: int) -> float: # Claude Opus定价示例 pricing = { 'claude-3-opus-20240229': {'input': 0.015, 'output': 0.075}, 'claude-3-sonnet-20240229': {'input': 0.003, 'output': 0.015} } model_pricing = pricing.get(self.model_name, pricing['claude-3-sonnet-20240229']) cost = (prompt_tokens / 1000 * model_pricing['input'] + completion_tokens / 1000 * model_pricing['output']) return cost def _handle_error(self, error: Exception) -> Dict[str, Any]: return { 'content': f"Error: {str(error)}", 'usage': {'prompt_tokens': 0, 'completion_tokens': 0, 'total_tokens': 0}, 'model': self.model_name, 'error': True }

5. 任务调度与接力逻辑实现

有了基础模型后，我们需要实现核心的接力逻辑。

5.1 任务分解策略

# core/task_decomposer.py import re from typing import List, Dict, Any class TaskDecomposer: """复杂任务分解器""" def __init__(self): self.decomposition_rules = { 'coding_task': self._decompose_coding_task, 'analysis_task': self._decompose_analysis_task, 'creative_task': self._decompose_creative_task } def decompose(self, task: str, task_type: str = 'general') -> List[Dict[str, Any]]: """根据任务类型分解复杂任务""" decomposer = self.decomposition_rules.get(task_type, self._decompose_general_task) return decomposer(task) def _decompose_coding_task(self, task: str) -> List[Dict[str, Any]]: """分解编码任务""" steps = [ { 'step': 1, 'description': '需求分析与技术选型', 'model': 'claude-3-opus-20240229', # Claude擅长分析 'role': '系统架构师', 'criteria': '明确技术栈和架构设计' }, { 'step': 2, 'description': '核心代码实现', 'model': 'gpt-4-turbo', # GPT擅长代码生成 'role': '高级开发工程师', 'criteria': '生成可运行的核心代码' }, { 'step': 3, 'description': '代码审查与优化', 'model': 'claude-3-sonnet-20240229', # Claude擅长逻辑检查 'role': '技术负责人', 'criteria': '检查代码质量和最佳实践' } ] return steps def _decompose_analysis_task(self, task: str) -> List[Dict[str, Any]]: """分解分析任务""" # 实现分析任务分解逻辑 pass def _decompose_creative_task(self, task: str) -> List[Dict[str, Any]]: """分解创意任务""" # 实现创意任务分解逻辑 pass def _decompose_general_task(self, task: str) -> List[Dict[str, Any]]: """通用任务分解""" # 实现通用分解逻辑 pass

5.2 接力执行引擎

# core/relay_engine.py import asyncio from typing import List, Dict, Any from models.gpt_model import GPTModel from models.claude_model import ClaudeModel class RelayEngine: """AI接力执行引擎""" def __init__(self, api_config: Dict[str, str]): self.models = self._initialize_models(api_config) self.task_history = [] def _initialize_models(self, api_config: Dict[str, str]) -> Dict[str, Any]: """初始化可用模型池""" models = {} if api_config.get('openai_api_key'): models['gpt-4-turbo'] = GPTModel('gpt-4-turbo', api_config['openai_api_key']) models['gpt-3.5-turbo'] = GPTModel('gpt-3.5-turbo', api_config['openai_api_key']) if api_config.get('anthropic_api_key'): models['claude-3-opus-20240229'] = ClaudeModel('claude-3-opus-20240229', api_config['anthropic_api_key']) models['claude-3-sonnet-20240229'] = ClaudeModel('claude-3-sonnet-20240229', api_config['anthropic_api_key']) return models async def execute_relay(self, task: str, steps: List[Dict[str, Any]]) -> Dict[str, Any]: """执行接力任务""" results = [] total_cost = 0.0 context = "" # 上下文传递 for step in steps: model_name = step['model'] if model_name not in self.models: raise ValueError(f"模型 {model_name} 未初始化") # 构建包含上下文的提示词 prompt = self._build_prompt(task, step, context) # 执行当前步骤 result = await self.models[model_name].generate(prompt) # 更新上下文和成本 context = result['content'] step_cost = self.models[model_name].estimate_cost( result['usage']['prompt_tokens'], result['usage']['completion_tokens'] ) total_cost += step_cost # 记录步骤结果 step_result = { 'step': step['step'], 'model': model_name, 'role': step['role'], 'prompt': prompt, 'response': result['content'], 'usage': result['usage'], 'cost': step_cost, 'success': not result.get('error', False) } results.append(step_result) # 如果当前步骤失败，考虑终止或回退 if not step_result['success']: break return { 'task': task, 'results': results, 'total_cost': total_cost, 'success': all(r['success'] for r in results) } def _build_prompt(self, task: str, step: Dict[str, Any], context: str) -> str: """构建包含上下文的提示词""" base_prompt = f""" 你正在参与一个AI协作任务，当前你的角色是：{step['role']} 总体任务：{task} 当前步骤：{step['description']} 成功标准：{step['criteria']} """ if context: base_prompt += f"之前步骤的上下文：\n{context}\n\n" base_prompt += "请基于以上信息完成当前步骤的任务：" return base_prompt

6. 直播集成与实时展示

为了让过程可视化，我们需要集成直播功能。

6.1 实时状态推送

#直播/live_broadcaster.py import asyncio import json from websockets.server import serve from typing import Dict, Any, List class LiveBroadcaster: """直播状态广播器""" def __init__(self, host: str = "localhost", port: int = 8765): self.host = host self.port = port self.connections = set() async def broadcast(self, message: Dict[str, Any]): """向所有连接广播消息""" if self.connections: message_json = json.dumps(message) await asyncio.gather( *[conn.send(message_json) for conn in self.connections], return_exceptions=True ) async def start_server(self): """启动WebSocket服务器""" async with serve(self._handle_connection, self.host, self.port): await asyncio.Future() # 永久运行 async def _handle_connection(self, websocket, path): """处理新连接""" self.connections.add(websocket) try: await websocket.wait_closed() finally: self.connections.remove(websocket)

6.2 直播界面设计

<!--直播界面示例：templates/live_dashboard.html--> <!DOCTYPE html> <html> <head> <title>AI接力直播看板</title> <style> .container { max-width: 1200px; margin: 0 auto; padding: 20px; } .step { border: 1px solid #ddd; margin: 10px 0; padding: 15px; } .current { background-color: #f0f8ff; border-left: 4px solid #007bff; } .completed { background-color: #f8fff8; border-left: 4px solid #28a745; } .failed { background-color: #fff8f8; border-left: 4px solid #dc3545; } .model-badge { display: inline-block; padding: 2px 8px; border-radius: 12px; font-size: 12px; } .gpt { background: #10a37f; color: white; } .claude { background: #d4af37; color: white; } </style> </head> <body> <div class="container"> <h1>AI接力任务直播</h1> <div id="taskInfo"></div> <div id="stepsContainer"></div> <div id="statsPanel"></div> </div> <script> // WebSocket连接和界面更新逻辑 const ws = new WebSocket('ws://localhost:8765'); ws.onmessage = function(event) { const data = JSON.parse(event.data); updateDashboard(data); }; function updateDashboard(data) { // 更新任务信息 document.getElementById('taskInfo').innerHTML = ` <h2>当前任务: ${data.task}</h2> <p>总成本: $${data.total_cost.toFixed(4)}</p> `; // 更新步骤状态 const stepsHtml = data.results.map(step => ` <div class="step ${step.success ? 'completed' : 'failed'}"> <h3>步骤 ${step.step}: ${step.role}</h3> <span class="model-badge ${step.model.includes('gpt') ? 'gpt' : 'claude'}"> ${step.model} </span> <p><strong>响应:</strong> ${step.response}</p> <p><small>Tokens: ${step.usage.total_tokens} | 成本: $${step.cost.toFixed(4)}</small></p> </div> `).join(''); document.getElementById('stepsContainer').innerHTML = stepsHtml; } </script> </body> </html>

7. 完整示例：代码生成接力实战

让我们通过一个完整的示例来演示整个流程。

7.1 示例任务定义

# examples/coding_relay_example.py import asyncio import os from core.relay_engine import RelayEngine from core.task_decomposer import TaskDecomposer async def main(): # 配置API密钥 api_config = { 'openai_api_key': os.getenv('OPENAI_API_KEY'), 'anthropic_api_key': os.getenv('ANTHROPIC_API_KEY') } # 初始化引擎 engine = RelayEngine(api_config) decomposer = TaskDecomposer() # 定义复杂任务 task = """ 开发一个Python Web应用，功能包括： 1. 用户注册登录系统 2. 文件上传和下载 3. 实时聊天功能 要求使用FastAPI框架，包含数据库设计和API文档。 """ # 分解任务 steps = decomposer.decompose(task, 'coding_task') # 执行接力 result = await engine.execute_relay(task, steps) # 输出结果 print("=== AI接力任务完成 ===") print(f"任务: {result['task']}") print(f"总成本: ${result['total_cost']:.4f}") print(f"成功率: {sum(1 for r in result['results'] if r['success'])}/{len(result['results'])}") for step_result in result['results']: status = "✅" if step_result['success'] else "❌" print(f"\n{status} 步骤{step_result['step']} - {step_result['role']}") print(f"模型: {step_result['model']}") print(f"响应: {step_result['response'][:200]}...") print(f"成本: ${step_result['cost']:.4f}") if __name__ == "__main__": asyncio.run(main())

7.2 运行与监控

# 设置环境变量 export OPENAI_API_KEY="your-openai-key" export ANTHROPIC_API_KEY="your-anthropic-key" # 运行示例 python examples/coding_relay_example.py

预期输出示例：

=== AI接力任务完成 === 任务: 开发一个Python Web应用... 总成本: $0.1245 成功率: 3/3 ✅ 步骤1 - 系统架构师 模型: claude-3-opus-20240229 响应: 基于需求分析，建议技术栈：FastAPI + SQLAlchemy + PostgreSQL... 成本: $0.0450 ✅ 步骤2 - 高级开发工程师 模型: gpt-4-turbo 响应: 以下是核心代码实现：from fastapi import FastAPI... 成本: $0.0675 ✅ 步骤3 - 技术负责人 模型: claude-3-sonnet-20240229 响应: 代码审查完成，发现以下优化点：1. 需要添加错误处理... 成本: $0.0120

8. 性能优化与成本控制

在实际使用中，性能和成本是需要重点考虑的因素。

8.1 成本优化策略

# optimization/cost_optimizer.py from typing import Dict, Any, List class CostOptimizer: """AI接力成本优化器""" def __init__(self, budget: float = 1.0): self.budget = budget self.cost_strategies = { 'aggressive': self._aggressive_optimization, 'balanced': self._balanced_optimization, 'conservative': self._conservative_optimization } def optimize_steps(self, steps: List[Dict[str, Any]], strategy: str = 'balanced') -> List[Dict[str, Any]]: """根据策略优化步骤配置""" optimizer = self.cost_strategies.get(strategy, self._balanced_optimization) return optimizer(steps) def _aggressive_optimization(self, steps: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """激进优化：尽可能使用便宜模型""" optimized_steps = [] for step in steps: optimized_step = step.copy() # 将昂贵模型替换为性价比更高的版本 if step['model'] == 'claude-3-opus-20240229': optimized_step['model'] = 'claude-3-sonnet-20240229' elif step['model'] == 'gpt-4-turbo': optimized_step['model'] = 'gpt-3.5-turbo' optimized_steps.append(optimized_step) return optimized_steps def _balanced_optimization(self, steps: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """平衡优化：关键步骤用强模型，简单步骤用便宜模型""" # 实现平衡优化逻辑 return steps def _conservative_optimization(self, steps: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """保守优化：保持原配置，通过其他方式节约""" return steps

8.2 缓存与重用策略

# optimization/response_cache.py import hashlib import json from typing import Dict, Any, Optional class ResponseCache: """响应缓存管理器""" def __init__(self, cache_file: str = "ai_cache.json"): self.cache_file = cache_file self.cache = self._load_cache() def get_cache_key(self, model: str, prompt: str, parameters: Dict[str, Any]) -> str: """生成缓存键""" content = f"{model}:{prompt}:{json.dumps(parameters, sort_keys=True)}" return hashlib.md5(content.encode()).hexdigest() def get(self, key: str) -> Optional[Dict[str, Any]]: """获取缓存响应""" return self.cache.get(key) def set(self, key: str, response: Dict[str, Any]): """设置缓存响应""" self.cache[key] = response self._save_cache() def _load_cache(self) -> Dict[str, Any]: """加载缓存文件""" try: with open(self.cache_file, 'r') as f: return json.load(f) except FileNotFoundError: return {} def _save_cache(self): """保存缓存到文件""" with open(self.cache_file, 'w') as f: json.dump(self.cache, f, indent=2)

9. 常见问题与解决方案

在实际实施过程中，你可能会遇到以下典型问题。

9.1 模型响应不一致问题

问题现象：不同模型对同一提示词产生完全不同的理解方向。

解决方案：

• 标准化提示词模板，明确角色和任务边界
• 添加上下文约束，确保后续模型基于前序结果工作
• 实现响应验证机制，检测偏离预期的输出

# validation/response_validator.py import re from typing import Dict, Any class ResponseValidator: """响应验证器""" def validate_code_response(self, response: str) -> Dict[str, Any]: """验证代码生成响应""" issues = [] # 检查是否包含代码块 if '```' not in response: issues.append("响应中未找到代码块") # 检查基础语法（简单验证） if 'def ' in response and ': ' not in response: issues.append("函数定义可能缺少冒号") return { 'valid': len(issues) == 0, 'issues': issues, 'score': max(0, 10 - len(issues)) # 简单评分 }

9.2 上下文传递失真问题

问题现象：在多步传递后，原始任务需求被逐渐扭曲。

解决方案：

• 定期重新注入原始任务描述
• 实现上下文摘要机制，避免过长历史
• 添加偏差检测和纠正逻辑

9.3 API限速与稳定性问题

问题现象：频繁的API调用触发限流或服务不稳定。

解决方案：

• 实现指数退避重试机制
• 添加请求队列和速率限制
• 准备备用模型方案

# utils/retry_handler.py import asyncio import random from typing import Callable, Any class RetryHandler: """重试处理器""" def __init__(self, max_retries: int = 3, base_delay: float = 1.0): self.max_retries = max_retries self.base_delay = base_delay async def execute_with_retry(self, func: Callable, *args, **kwargs) -> Any: """带重试的执行""" last_exception = None for attempt in range(self.max_retries): try: return await func(*args, **kwargs) except Exception as e: last_exception = e if attempt < self.max_retries - 1: delay = self.base_delay * (2 ** attempt) + random.uniform(0, 0.1) await asyncio.sleep(delay) raise last_exception

10. 生产环境最佳实践

将AI接力系统部署到生产环境时，需要考虑以下关键点。

10.1 安全与权限管理

• API密钥管理：使用环境变量或专业密钥管理服务
• 请求审计：记录所有AI交互用于监控和调试
• 输入验证：防止提示词注入攻击

10.2 监控与告警

# monitoring/system_monitor.py import time import logging from dataclasses import dataclass from typing import Dict, Any @dataclass class SystemMetrics: """系统监控指标""" total_requests: int = 0 successful_requests: int = 0 total_cost: float = 0.0 average_response_time: float = 0.0 class SystemMonitor: """系统监控器""" def __init__(self): self.metrics = SystemMetrics() self.logger = logging.getLogger('ai_relay_monitor') def record_request(self, success: bool, cost: float, response_time: float): """记录请求指标""" self.metrics.total_requests += 1 if success: self.metrics.successful_requests += 1 self.metrics.total_cost += cost # 更新平均响应时间 old_avg = self.metrics.average_response_time count = self.metrics.successful_requests self.metrics.average_response_time = ( (old_avg * (count - 1) + response_time) / count if count > 0 else response_time ) # 触发告警检查 self._check_alerts() def _check_alerts(self): """检查是否需要触发告警""" success_rate = (self.metrics.successful_requests / self.metrics.total_requests if self.metrics.total_requests > 0 else 1.0) if success_rate < 0.8: # 成功率低于80% self.logger.warning(f"系统成功率下降: {success_rate:.2%}") if self.metrics.total_cost > 100: # 月度成本超过100美元 self.logger.warning(f"月度成本预警: ${self.metrics.total_cost:.2f}")

10.3 版本控制与回滚

• 配置版本化：所有模型配置和提示词模板应该版本控制
• 渐进式部署：新配置先在小范围测试
• 快速回滚：准备一键回滚到稳定版本

11. 扩展应用场景

AI接力模式不仅适用于代码生成，还可以扩展到更多场景。

11.1 技术文档编写

• 步骤1：Claude分析需求，制定文档结构
• 步骤2：GPT生成详细内容
• 步骤3：专用模型进行格式检查和链接验证

11.2 数据分析流水线

• 步骤1：Claude理解分析需求，制定分析计划
• 步骤2：GPT生成数据处理代码
• 步骤3：结果解释和可视化建议

11.3 产品需求分析

• 步骤1：多角度需求分析
• 步骤2：竞品调研和差异化分析
• 步骤3：功能优先级排序

这种AI接力直播的技术模式，本质上是在探索如何让不同的AI模型更好地协作。对于开发者来说，重要的不是复现某个具体的直播效果，而是理解这种协作模式背后的工程价值，并将其应用到实际的项目开发中。

从技术演进的角度看，多模型协作很可能成为未来AI应用的标准架构。现在开始积累相关经验，将为你在AI工程化浪潮中占据有利位置。建议从小的实验项目开始，逐步验证不同模型组合的效果，找到最适合自己业务场景的协作模式。