微软开源的AI调度中枢，终于有人讲清楚了！JARVIS如何让10+种大模型自动协作，这篇实战教程说透了
导语
当ChatGPT、Stable Diffusion、Whisper等AI模型层出不穷时，开发者面临一个甜蜜的烦恼：如何让这些”各有所长”的模型协同工作？手动编写调用逻辑、处理格式转换、管理API配额——这些工作足以让一个中型项目变成维护噩梦。
微软JARVIS项目正是为解决这一痛点而生。它像一个智能交通枢纽，能够自动理解用户意图、调度合适的AI模型、处理模型间的数据流转，最终输出整合后的结果。想象一下，你只需说一句”帮我把这张图片里的人像抠出来，然后翻译图片中的文字，最后生成一段配图文案”，JARVIS就能自动规划任务流、调用CV模型处理图像、调用NLP模型翻译、调用LLM生成文案——整个过程对你透明。
本文将从原理出发，手把手教你从零搭建JARVIS环境，通过实战案例展示如何用它构建复杂的多模型协作应用。无论你是AI爱好者还是企业开发者，这篇教程都将帮你打开多模型协同的大门。
第一部分：为什么你应该关注JARVIS
解决多模型时代的”碎片化”困境
当前的AI生态呈现”诸侯割据”的格局：GPT-4在推理和对话上无出其右，Claude在长文本理解上表现优异，Stable Diffusion在图像生成上独领风骚，Whisper在语音识别上几乎无人能敌。每个模型都是某个领域的”专精选手”，但它们之间几乎没有交集。
传统解决方案是手动串联：先调API获取结果，格式化后传给下一个模型，再处理输出。这条链路一旦变长，代码就变得难以维护，更别提处理错误恢复、并发控制、结果缓存等工程问题了。
JARVIS的核心价值在于自动化和智能化：
用户需求 → JARVIS理解意图 → 选择合适模型 → 编排执行顺序 → 整合输出
它不是简单地把模型串起来，而是真正理解任务本质，自动规划最优执行路径。
技术架构的精妙之处
JARVIS基于HuggingGPT的核心理念，但做了大量工程优化。其架构可以概括为”一个大脑，两套机制”：
一个大脑：大型语言模型（LLM）作为核心调度器，负责理解用户指令、拆解任务、规划执行流程、整合结果。这通常使用GPT-3.5/4或同等能力的开源模型。
两套机制：
-
任务规划机制（Task Planning）：LLM分析用户请求，将其分解为可执行的子任务。例如，”分析这篇PDF并生成摘要”可能被分解为：提取文本→文本清洗→生成摘要。
-
模型选择机制（Model Selection）：根据每个子任务的性质，从模型池中选择最合适的模型。JARVIS维护了一个模型注册表，包含各模型的能力描述、输入输出格式、性能指标。
这种设计的精妙之处在于：它把”做什么”（由LLM决定）和”怎么做”（由专用模型执行）分离，兼顾了灵活性与专业性。
与其他方案的对比优势
| 特性 | 手动API调用 | LangChain | JARVIS |
|---|---|---|---|
| 代码复杂度 | 高（需处理所有细节） | 中等 | 低（声明式任务描述） |
| 模型调度 | 手动指定 | 半自动 | 全自动 |
| 错误恢复 | 需自行实现 | 有限支持 | 框架层支持 |
| 扩展性 | 差 | 好 | 优秀 |
| 学习成本 | 因模型而异 | 较高 | 较低 |
对于希望快速原型验证、同时又不失扩展性的团队，JARVIS是一个极具竞争力的选择。
第二部分：环境搭建与基础配置
系统要求与依赖准备
在开始之前，确保你的开发环境满足以下要求：
# Python版本要求（推荐3.10或更高）
python --version
# 输出应类似: Python 3.10.12
# 检查包管理器
pip --version
# 或使用 conda
conda --version
JARVIS依赖的核心库包括：
transformers：Hugging Face模型库，用于加载各类AI模型torch：深度学习框架openai或anthropic：API调用（如果你使用商业模型）gradio或fastapi：Web界面框架requests：HTTP请求处理
安装JARVIS主仓库
# 克隆官方仓库
git clone https://github.com/microsoft/JARVIS.git
# 进入项目目录
cd JARVIS
# 查看项目结构
ls -la
项目结构通常如下：
JARVIS/
├── jarvis/
│ ├── __init__.py
│ ├── llm/ # 大语言模型相关
│ ├── models/ # 模型调度相关
│ ├── planner/ # 任务规划相关
│ ├── tools/ # 工具接口
│ └── server/ # Web服务
├── examples/ # 示例代码
├── requirements.txt
└── README.md
安装依赖包
# 方式一：使用requirements文件
pip install -r requirements.txt
# 方式二：手动安装核心依赖（如果你需要更精细控制）
pip install torch>=2.0.0
pip install transformers>=4.30.0
pip install gradio>=3.40.0
pip install openai>=0.27.0
pip install anthropic>=0.5.0
pip install pydantic>=2.0.0
pip install python-dotenv>=1.0.0
配置API密钥
JARVIS支持多种模型调用方式。对于商业模型（如OpenAI、Anthropic），需要配置API密钥：
# 创建环境变量文件
touch .env
# 编辑.env文件，添加API密钥
cat >> .env << 'EOF'
# OpenAI配置
OPENAI_API_KEY=sk-your-openai-api-key-here
OPENAI_API_BASE=https://api.openai.com/v1
# Anthropic配置（如果使用Claude）
ANTHROPIC_API_KEY=sk-ant-your-anthropic-key-here
# Hugging Face配置（如果使用开源模型）
HF_TOKEN=hf_your-huggingface-token
EOF
安全提示：永远不要将包含真实密钥的.env文件提交到版本控制系统。确保.gitignore包含.env。
# 确保.gitignore包含敏感文件
echo ".env" >> .gitignore
echo "__pycache__/" >> .gitignore
echo "*.pyc" >> .gitignore
验证安装
创建一个验证脚本，确认环境配置正确：
# verify_installation.py
"""验证JARVIS安装是否成功的脚本"""
import sys
print("=" * 50)
print("JARVIS 环境验证")
print("=" * 50)
# 检查Python版本
print(f"\n[1] Python版本: {sys.version}")
# 检查核心依赖
dependencies = [
"torch",
"transformers",
"gradio",
"openai",
"anthropic",
"pydantic",
"dotenv"
]
print("\n[2] 依赖包检查:")
all_ok = True
for dep in dependencies:
try:
__import__(dep)
print(f" ✓ {dep}")
except ImportError:
print(f" ✗ {dep} - 未安装")
all_ok = False
# 检查JARVIS模块
print("\n[3] JARVIS模块检查:")
try:
from jarvis import JARVIS
print(" ✓ JARVIS主模块")
except ImportError as e:
print(f" ✗ JARVIS导入失败: {e}")
all_ok = False
# 检查环境变量
print("\n[4] 环境变量检查:")
from dotenv import load_dotenv
load_dotenv()
import os
api_vars = ["OPENAI_API_KEY", "ANTHROPIC_API_KEY", "HF_TOKEN"]
for var in api_vars:
value = os.getenv(var)
if value:
# 只显示前几位，保护密钥安全
masked = value[:8] + "..." if len(value) > 8 else "***"
print(f" ✓ {var}: {masked}")
else:
print(f" - {var}: 未设置")
print("\n" + "=" * 50)
if all_ok:
print("验证通过！JARVIS已准备就绪。")
else:
print("验证发现问题，请检查上述错误。")
print("=" * 50)
运行验证：
python verify_installation.py
第三部分：核心概念与架构解析
理解JARVIS的工作流程
JARVIS的处理流程可以分为四个阶段，每个阶段都有明确的职责：
阶段一：需求理解
# 用户输入自然语言请求
user_request = "分析这张图片中的商品，并生成一段适合朋友圈的推荐文案"
JARVIS首先将用户请求发送给LLM，让它理解用户的真实意图。这包括：
- 提取关键实体（图片、商品、文案）
- 识别任务类型（分析+生成）
- 理解约束条件（朋友圈风格）
阶段二：任务规划
LLM输出一个结构化的任务计划：
{
"tasks": [
{
"id": 1,
"type": "image_analysis",
"model": "Salesforce/blip-flan-xxl",
"input": "用户上传的图片",
"output": "图片描述和商品特征"
},
{
"id": 2,
"type": "text_generation",
"model": "gpt-3.5-turbo",
"input": "商品特征 + 朋友圈风格约束",
"output": "推荐文案"
}
],
"dependencies": {
"2": ["1"] # 任务2依赖任务1的输出
}
}
阶段三：模型执行
JARVIS按照计划依次执行任务：
# 执行流程伪代码
result_1 = await model_registry["Salesforce/blip-flan-xxl"].run(image)
result_2 = await model_registry["gpt-3.5-turbo"].run(
prompt=f"根据以下商品特征生成朋友圈文案: {result_1}"
)
final_result = result_2
阶段四：结果整合
将各任务输出整合为用户期望的格式：
# 输出最终结果
final_output = {
"商品分析": result_1,
"推荐文案": result_2
}
核心组件详解
1. Planner（任务规划器）
Planner是JARVIS的”大脑”，负责将自然语言转换为可执行计划：
# jarvis/planner/task_planner.py
"""任务规划器核心实现"""
class TaskPlanner:
"""
任务规划器：根据用户请求生成执行计划
"""
def __init__(self, llm_client):
self.llm = llm_client
self.task_templates = self._load_templates()
def plan(self, user_request: str) -> ExecutionPlan:
"""
生成执行计划
参数:
user_request: 用户的自然语言请求
返回:
ExecutionPlan: 包含任务列表和依赖关系的执行计划
"""
# 构造提示词，引导LLM输出结构化计划
prompt = self._build_planning_prompt(user_request)
# 调用LLM获取计划
raw_plan = self.llm.complete(prompt)
# 解析为结构化对象
plan = self._parse_plan(raw_plan)
return plan
def _build_planning_prompt(self, request: str) -> str:
"""构建规划提示词"""
return f"""
你是一个任务规划助手。用户将提出一个请求，你需要将其分解为可执行的子任务。
可用模型类型：
- 文本生成模型：GPT-3.5, GPT-4, LLaMA
- 图像分析模型：BLIP, CLIP, GPT-4V
- 图像生成模型：Stable Diffusion, DALL-E
- 语音识别模型：Whisper
- 文本翻译模型：NLLB, M2M-100
请按以下JSON格式输出任务计划：
{{
"tasks": [
{{"id": 1, "type": "任务类型", "model": "推荐模型", "description": "任务描述"}},
...
],
"dependencies": {{"任务ID": ["前置任务ID列表"]}}
}}
用户请求：{request}
任务计划：
"""
2. Model Registry（模型注册表）
Model Registry管理所有可用模型的信息：
# jarvis/models/model_registry.py
"""模型注册表核心实现"""
from dataclasses import dataclass
from typing import Dict, List, Optional
import json
@dataclass
class ModelCapability:
"""模型能力描述"""
task_type: str # 任务类型，如 "image-to-text"
input_format: str # 输入格式
output_format: str # 输出格式
max_input_length: int # 最大输入长度
strengths: List[str] # 模型优势
limitations: List[str] # 模型限制
@dataclass
class ModelInfo:
"""模型信息"""
model_id: str
provider: str # "openai", "huggingface", "local"
endpoint: Optional[str] # API端点
capability: ModelCapability
api_key_env: Optional[str] # 环境变量名，存储API密钥
is_local: bool # 是否本地模型
class ModelRegistry:
"""
模型注册表：管理系统中所有可用模型
"""
def __init__(self):
self.models: Dict[str, ModelInfo] = {}
self._register_default_models()
def _register_default_models(self):
"""注册默认模型"""
# 注册OpenAI模型
self.register(ModelInfo(
model_id="gpt-4",
provider="openai",
endpoint="https://api.openai.com/v1/chat/completions",
capability=ModelCapability(
task_type="text-generation",
input_format="text",
output_format="text",
max_input_length=8192,
strengths=["推理能力强", "上下文理解好", "多语言支持"],
limitations=["成本较高", "有速率限制"]
),
api_key_env="OPENAI_API_KEY",
is_local=False
))
# 注册HuggingFace图像分析模型
self.register(ModelInfo(
model_id="Salesforce/blip-image-captioning-large",
provider="huggingface",
endpoint=None,
capability=ModelCapability(
task_type="image-to-text",
input_format="image",
output_format="text",
max_input_length=1,
strengths=["图像描述准确", "支持多种图像"],
limitations=["无法理解复杂场景"]
),
api_key_env="HF_TOKEN",
is_local=True
))
# 注册语音识别模型
self.register(ModelInfo(
model_id="openai/whisper-large-v3",
provider="huggingface",
endpoint=None,
capability=ModelCapability(
task_type="speech-to-text",
input_format="audio",
output_format="text",
max_input_length=1,
strengths=["多语言支持", "识别准确率高"],
limitations=["需要清晰音频"]
),
api_key_env="HF_TOKEN",
is_local=True
))
def register(self, model_info: ModelInfo):
"""注册新模型"""
self.models[model_info.model_id] = model_info
def get(self, model_id: str) -> Optional[ModelInfo]:
"""获取模型信息"""
return self.models.get(model_id)
def find_models(self, task_type: str) -> List[ModelInfo]:
"""根据任务类型查找可用模型"""
return [
m for m in self.models.values()
if m.capability.task_type == task_type
]
def recommend_model(self, task_type: str,
constraints: Optional[Dict] = None) -> Optional[ModelInfo]:
"""
根据任务类型和约束推荐最佳模型
参数:
task_type: 任务类型
constraints: 约束条件，如 {"max_cost": "low", "speed": "fast"}
返回:
推荐的最合适模型
"""
candidates = self.find_models(task_type)
if not candidates:
return None
# 简单的选择逻辑（实际应用中会更复杂）
# 优先选择本地模型（更快、更便宜）
local_candidates = [m for m in candidates if m.is_local]
if local_candidates:
return local_candidates[0]
return candidates[0]
3. Executor（任务执行器）
Executor负责实际运行模型、处理输入输出：
# jarvis/executor/task_executor.py
"""任务执行器核心实现"""
import asyncio
from typing import Any, Dict, List
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
@dataclass
class TaskResult:
"""任务执行结果"""
task_id: str
success: bool
output: Any
error: Optional[str] = None
execution_time: float = 0.0
class TaskExecutor:
"""
任务执行器：负责执行具体的模型调用
"""
def __init__(self, model_registry, cache=None):
self.registry = model_registry
self.cache = cache # 可选的缓存层
self.executor = ThreadPoolExecutor(max_workers=4)
async def execute(self, plan: ExecutionPlan) -> Dict[str, TaskResult]:
"""
执行完整计划
参数:
plan: 任务执行计划
返回:
任务ID到结果的映射
"""
results = {}
# 构建依赖图
dependency_graph = self._build_dependency_graph(plan)
# 按依赖顺序执行（支持并行的任务会并行执行）
completed = set()
while len(completed) < len(plan.tasks):
# 找出当前可执行的任务（所有依赖都已完成）
ready_tasks = self._get_ready_tasks(plan, completed, dependency_graph)
if not ready_tasks:
# 理论上不应该发生，除非有循环依赖
break
# 并行执行准备好的任务
batch_results = await asyncio.gather(
*[self._execute_single(task, plan, results) for task in ready_tasks],
return_exceptions=True
)
# 处理结果
for task, result in zip(ready_tasks, batch_results):
if isinstance(result, Exception):
results[task.id] = TaskResult(
task_id=task.id,
success=False,
output=None,
error=str(result)
)
else:
results[task.id] = result
completed.add(task.id)
return results
async def _execute_single(self, task, plan, results) -> TaskResult:
"""执行单个任务"""
import time
start_time = time.time()
# 准备输入（可能需要从依赖任务获取）
inputs = self._prepare_inputs(task, plan, results)
# 获取模型
model_info = self.registry.get(task.model)
if not model_info:
return TaskResult(
task_id=task.id,
success=False,
output=None,
error=f"模型 {task.model} 未找到"
)
# 调用模型（这里简化了，实际实现会更复杂）
try:
output = await self._call_model(model_info, inputs)
return TaskResult(
task_id=task.id,
success=True,
output=output,
execution_time=time.time() - start_time
)
except Exception as e:
return TaskResult(
task_id=task.id,
success=False,
output=None,
error=str(e),
execution_time=time.time() - start_time
)
async def _call_model(self, model_info: ModelInfo, inputs: Dict) -> Any:
"""调用具体的模型"""
# 根据provider选择调用方式
if model_info.provider == "openai":
return await self._call_openai(model_info, inputs)
elif model_info.provider == "huggingface":
return await self._call_huggingface(model_info, inputs)
else:
raise ValueError(f"不支持的provider: {model_info.provider}")
async def _call_openai(self, model_info: ModelInfo, inputs: Dict) -> str:
"""调用OpenAI API"""
from openai import AsyncOpenAI
import os
from dotenv import load_dotenv
load_dotenv()
client = AsyncOpenAI(api_key=os.getenv(model_info.api_key_env))
response = await client.chat.completions.create(
model=model_info.model_id,
messages=inputs.get("messages", []),
temperature=inputs.get("temperature", 0.7)
)
return response.choices[0].message.content
async def _call_huggingface(self, model_info: ModelInfo, inputs: Dict) -> str:
"""调用HuggingFace模型"""
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
# 加载模型和tokenizer
tokenizer = AutoTokenizer.from_pretrained(model_info.model_id)
model = AutoModelForCausalLM.from_pretrained(
model_info.model_id,
torch_dtype=torch.float16,
device_map="auto"
)
# 处理输入
prompt = inputs.get("prompt", "")
inputs_tokens = tokenizer(prompt, return_tensors="pt").to(model.device)
# 生成
outputs = model.generate(**inputs_tokens, max_new_tokens=512)
generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
return generated_text
第四部分：实战教程——从入门到精通
案例一：构建一个多模态内容分析助手
本案例将创建一个能够分析图片、提取关键信息、生成评论的多模态助手。
完整代码：
# multi_modal_assistant.py
"""
多模态内容分析助手
功能：接收图片URL，提取内容，分析情感，生成评论
"""
import os
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
# 导入JARVIS核心组件
from jarvis import JARVIS
from jarvis.planner import TaskPlanner
from jarvis.models import ModelRegistry
from jarvis.executor import TaskExecutor
# ==================== 配置部分 ====================
class AssistantConfig:
"""助手配置"""
# LLM配置（用于任务规划）
PLANNER_LLM = "gpt-4"
PLANNER_API_KEY = os.getenv("OPENAI_API_KEY")
# 模型配置
IMAGE_MODEL = "Salesforce/blip-image-captioning-large"
SENTIMENT_MODEL = "nlptown/bert-base-multilingual-uncased-sentiment"
# 情感分析提示模板
SENTIMENT_PROMPT = """
请分析以下评论的情感倾向，返回简要分析结果：
评论内容：{comment}
分析要求：
1. 识别情感（积极/消极/中性）
2. 指出主要情感因素
3. 用中文输出
"""
# ==================== 核心逻辑 ====================
class MultiModalAssistant:
"""
多模态内容分析助手
工作流程：
1. 接收图片URL
2. 分析图片内容
3. 生成评论
4. 分析评论情感
5. 返回综合结果
"""
def __init__(self):
# 初始化各组件
print("=" * 50)
print("初始化多模态内容分析助手...")
print("=" * 50)
# 创建模型注册表
self.registry = ModelRegistry()
print("✓ 模型注册表初始化完成")
# 创建规划器（使用GPT-4）
from openai import OpenAI
self.llm_client = OpenAI(api_key=self.PLANNER_API_KEY)
print("✓ LLM客户端初始化完成")
# 创建执行器
self.executor = TaskExecutor(self.registry)
print("✓ 任务执行器初始化完成")
print("=" * 50)
print("助手初始化完成！")
print("=" * 50)
def analyze_image(self, image_url: str) -> dict:
"""
分析图片并生成评论
参数:
image_url: 图片的URL地址
返回:
包含分析结果的字典
"""
print(f"\n开始分析图片: {image_url}")
# 步骤1：分析图片内容
print("\n[步骤1/4] 分析图片内容...")
image_description = self._analyze_image_content(image_url)
print(f" 图片描述: {image_description[:100]}...")
# 步骤2：生成评论
print("\n[步骤2/4] 生成评论...")
comments = self._generate_comments(image_description)
print(f" 生成评论数: {len(comments)}")
# 步骤3：情感分析
print("\n[步骤3/4] 分析评论情感...")
sentiment_results = []
for i, comment in enumerate(comments, 1):
sentiment = self._analyze_sentiment(comment)
sentiment_results.append({
"comment": comment,
"sentiment": sentiment
})
print(f" 评论{i}情感: {sentiment['overall']}")
# 步骤4：整合结果
print("\n[步骤4/4] 整合分析结果...")
final_result = {
"image_url": image_url,
"image_description": image_description,
"comments_with_sentiment": sentiment_results,
"summary": self._generate_summary(sentiment_results)
}
return final_result
def _analyze_image_content(self, image_url: str) -> str:
"""分析图片内容（简化实现）"""
# 实际应用中，这里会调用图像分析模型
# 这里用模拟结果演示
return (
"一张精致的咖啡杯照片，放在木质桌面上，"
"背景是模糊的咖啡馆内景。咖啡杯上有精美的拉花，"
"旁边有一本翻开的书和一副眼镜。光线柔和，给人温馨的感觉。"
)
def _generate_comments(self, description: str) -> list:
"""生成多条评论"""
prompt = f"""基于以下图片描述，生成3条不同风格的社交媒体评论：
图片描述：{description}
要求：
1. 一条正面积极，一条中性描述，一条幽默调侃
2. 每条不超过50字
3. 用中文
4. 直接输出三条评论，用换行分隔
"""
response = self.llm_client.chat.completions.create(
model=self.PLANNER_LLM,
messages=[
{"role": "system", "content": "你是一个社交媒体内容创作者。"},
{"role": "user", "content": prompt}
],
temperature=0.8,
max_tokens=500
)
comments_text = response.choices[0].message.content
comments = [c.strip() for c in comments_text.split('\n') if c.strip()]
return comments
def _analyze_sentiment(self, text: str) -> dict:
"""分析文本情感"""
prompt = self.SENTIMENT_PROMPT.format(comment=text)
response = self.llm_client.chat.completions.create(
model=self.PLANNER_LLM,
messages=[
{"role": "system", "content": "你是一个情感分析专家。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=200
)
analysis = response.choices[0].message.content
# 简单判断情感极性
if "积极" in analysis or "正面" in analysis:
overall = "positive"
elif "消极" in analysis or "负面" in analysis:
overall = "negative"
else:
overall = "neutral"
return {
"overall": overall,
"analysis": analysis
}
def _generate_summary(self, results: list) -> str:
"""生成总结报告"""
positive_count = sum(1 for r in results if r['sentiment']['overall'] == 'positive')
neutral_count = sum(1 for r in results if r['sentiment']['overall'] == 'neutral')
negative_count = sum(1 for r in results if r['sentiment']['overall'] == 'negative')
return (
f"本次分析共生成{len(results)}条评论，其中"
f"积极评论{positive_count}条，中性评论{neutral_count}条，"
f"消极评论{negative_count}条。"
)
# ==================== 使用示例 ====================
def main():
"""主函数：演示多模态助手的使用"""
print("\n" + "=" * 60)
print("多模态内容分析助手 - 使用演示")
print("=" * 60)
# 创建助手实例
assistant = MultiModalAssistant()
# 准备测试图片
test_images = [
"https://example.com/coffee.jpg",
"https://example.com/landscape.jpg"
]
# 逐个分析
for i, image_url in enumerate(test_images, 1):
print(f"\n{'='*60}")
print(f"分析第 {i} 张图片")
print(f"{'='*60}")
result = assistant.analyze_image(image_url)
# 输出结果
print("\n" + "-" * 40)
print("分析结果：")
print("-" * 40)
print(f"\n图片描述：{result['image_description']}")
print(f"\n生成的评论：")
for i, item in enumerate(result['comments_with_sentiment'], 1):
print(f"\n 评论{i}：{item['comment']}")
print(f" 情感：{item['sentiment']['overall']}")
print(f"\n总结：{result['summary']}")
if __name__ == "__main__":
main()
案例二：构建文档处理流水线
本案例展示如何构建一个复杂的文档处理流水线，包括PDF解析、关键信息提取、翻译和摘要生成。
完整代码：
# document_pipeline.py
"""
文档处理流水线
功能：处理PDF文档，提取信息，翻译关键内容，生成结构化摘要
"""
import os
import re
from typing import List, Dict, Optional
from dataclasses import dataclass
from dotenv import load_dotenv
load_dotenv()
# ==================== 数据模型 ====================
@dataclass
class DocumentMetadata:
"""文档元数据"""
filename: str
page_count: int
language: str
file_size: int
created_date: Optional[str] = None
@dataclass
class ExtractedInfo:
"""提取的信息"""
title: str
authors: List[str]
key_phrases: List[str]
key_dates: List[str]
entities: Dict[str, List[str]] # 实体类型 -> 实体列表
@dataclass
class ProcessingResult:
"""处理结果"""
metadata: DocumentMetadata
original_summary: str
translated_summary: str
key_findings: List[str]
action_items: List[str]
# ==================== 核心处理类 ====================
class DocumentPipeline:
"""
文档处理流水线
处理流程：
1. 文档解析（PDF -> 文本）
2. 信息提取（关键实体、日期、术语）
3. 内容摘要（生成摘要）
4. 翻译处理（目标语言）
5. 结构化输出
"""
def __init__(self, target_language: str = "Chinese"):
self.target_language = target_language
self._setup_llm()
print("=" * 50)
print(f"文档处理流水线初始化完成")
print(f"目标语言: {target_language}")
print("=" * 50)
def _setup_llm(self):
"""初始化LLM客户端"""
from openai import OpenAI
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise ValueError("未设置 OPENAI_API_KEY")
self.llm = OpenAI(api_key=api_key)
self.model = "gpt-4"
def process(self, document_path: str) -> ProcessingResult:
"""
处理文档
参数:
document_path: 文档路径
返回:
ProcessingResult: 处理结果
"""
print(f"\n开始处理文档: {document_path}")
print("-" * 50)
# 步骤1：解析文档
print("[1/5] 解析文档...")
raw_text = self._parse_document(document_path)
metadata = self._extract_metadata(document_path, raw_text)
print(f" 解析完成，提取 {len(raw_text)} 字符")
# 步骤2：提取关键信息
print("\n[2/5] 提取关键信息...")
extracted_info = self._extract_information(raw_text)
print(f" 提取到 {len(extracted_info.key_phrases)} 个关键短语")
print(f" 识别到 {len(extracted_info.entities.get('PERSON', []))} 个人名")
# 步骤3：生成摘要
print("\n[3/5] 生成摘要...")
original_summary = self._generate_summary(raw_text, extracted_info)
print(f" 摘要生成完成 ({len(original_summary)} 字符)")
# 步骤4：翻译摘要
print("\n[4/5] 翻译摘要...")
translated_summary = self._translate_text(original_summary)
print(f" 翻译完成")
# 步骤5：提取行动项
print("\n[5/5] 提取行动项...")
action_items = self._extract_action_items(raw_text)
key_findings = self._extract_key_findings(raw_text, extracted_info)
print(f" 提取到 {len(action_items)} 个行动项")
print(f" 提取到 {len(key_findings)} 个关键发现")
print("-" * 50)
print("文档处理完成！")
return ProcessingResult(
metadata=metadata,
original_summary=original_summary,
translated_summary=translated_summary,
key_findings=key_findings,
action_items=action_items
)
def _parse_document(self, path: str) -> str:
"""
解析文档（这里简化处理，实际需要用PyPDF2或pdfplumber）
"""
# 模拟文档内容
sample_content = """
研究报告：人工智能在医疗领域的应用
作者：张伟、李明、王芳
摘要：
本研究探讨了人工智能技术在医疗诊断、药物研发和健康管理中的应用。
通过分析100家医院的临床数据，我们发现AI辅助诊断系统能够提高诊断准确率
约15%，同时减少30%的诊断时间。在药物研发领域，AI技术成功将新药研发周期
从平均10年缩短至5-7年。
主要发现：
1. AI辅助影像诊断准确率达到97.5%
2. 药物分子设计效率提升40%
3. 个性化治疗方案覆盖率提高至85%
建议行动：
- 加强医疗机构与AI公司的合作
- 建立标准化数据共享机制
- 完善AI医疗应用的监管框架
"""
return sample_content
def _extract_metadata(self, path: str, content: str) -> DocumentMetadata:
"""提取文档元数据"""
filename = os.path.basename(path)
return DocumentMetadata(
filename=filename,
page_count=len(content) // 2000 + 1, # 估算
language="zh-CN",
file_size=len(content.encode('utf-8')),
created_date="2024-01-15"
)
def _extract_information(self, text: str) -> ExtractedInfo:
"""提取关键信息"""
# 提取标题（第一行非空行）
lines = text.strip().split('\n')
title = next((l.strip() for l in lines if l.strip()), "未知标题")
# 提取作者
authors_match = re.search(r'作者[：:](.+?)(?:\n|$)', text)
authors = []
if authors_match:
authors = [a.strip() for a in authors_match.group(1).split('、')]
# 使用LLM提取关键短语
key_phrases = self._extract_key_phrases_with_llm(text)
# 提取日期
date_pattern = r'\d{4}年\d{1,2}月|\d{4}-\d{2}-\d{2}|\d+年'
key_dates = re.findall(date_pattern, text)
# 提取实体（使用LLM）
entities = self._extract_entities_with_llm(text)
return ExtractedInfo(
title=title,
authors=authors,
key_phrases=key_phrases,
key_dates=list(set(key_dates)),
entities=entities
)
def _extract_key_phrases_with_llm(self, text: str) -> List[str]:
"""使用LLM提取关键短语"""
prompt = f"""从以下文本中提取10个最重要的关键短语或术语。
文本：
{text[:2000]}
要求：
1. 每个短语用逗号分隔
2. 只输出短语，不要其他说明
3. 选择真正重要的术语和概念
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个信息提取专家。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=300
)
result = response.choices[0].message.content
phrases = [p.strip() for p in result.split('，') if p.strip()]
return phrases
def _extract_entities_with_llm(self, text: str) -> Dict[str, List[str]]:
"""使用LLM提取命名实体"""
prompt = f"""从以下文本中提取命名实体，按类型分类。
文本：
{text[:2000]}
请提取以下类型的实体：
- PERSON（人名）
- ORGANIZATION（组织机构）
- LOCATION（地点）
- DATE（日期）
- PRODUCT（产品名称）
输出格式（JSON）：
{{
"PERSON": ["人名1", "人名2"],
"ORGANIZATION": ["机构1"],
...
}}
只输出JSON，不要其他内容。
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个信息提取专家，擅长NER任务。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=500
)
result = response.choices[0].message.content
# 解析JSON
import json
try:
# 尝试提取JSON部分
json_match = re.search(r'\{.*\}', result, re.DOTALL)
if json_match:
entities = json.loads(json_match.group())
else:
entities = {}
except json.JSONDecodeError:
entities = {}
return entities
def _generate_summary(self, text: str, info: ExtractedInfo) -> str:
"""生成摘要"""
prompt = f"""为以下文档生成一个全面的摘要。
文档标题：{info.title}
作者：{', '.join(info.authors) if info.authors else '未知'}
关键短语：{', '.join(info.key_phrases[:5])}
文档内容：
{text[:3000]}
摘要要求：
1. 长度200-300字
2. 包含研究背景、主要发现、结论
3. 使用专业但易懂的语言
4. 用中文输出
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个专业的学术写作助手。"},
{"role": "user", "content": prompt}
],
temperature=0.5,
max_tokens=600
)
return response.choices[0].message.content
def _translate_text(self, text: str) -> str:
"""翻译文本"""
language_map = {
"Chinese": "中文",
"English": "英文",
"Japanese": "日文"
}
target = language_map.get(self.target_language, self.target_language)
prompt = f"""将以下文本翻译成{target}。
原文：
{text}
要求：
1. 保持原文风格和专业术语
2. 翻译自然流畅
3. 只输出翻译结果，不要其他说明
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": f"你是一个专业的翻译专家，擅长翻译成{target}。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=800
)
return response.choices[0].message.content
def _extract_key_findings(self, text: str, info: ExtractedInfo) -> List[str]:
"""提取关键发现"""
prompt = f"""从以下文档中提取关键发现和研究结论。
文档：
{text[:2500]}
要求：
1. 提取3-5个最重要的发现
2. 每个发现一句话概括
3. 重点关注数据、结论和洞见
4. 用中文输出
5. 每条发现用数字编号
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个研究分析专家。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=400
)
findings_text = response.choices[0].message.content
findings = [f.strip() for f in findings_text.split('\n') if f.strip()]
return findings
def _extract_action_items(self, text: str) -> List[str]:
"""提取行动项"""
prompt = f"""从以下文档中提取建议行动项。
文档：
{text[:2500]}
要求：
1. 提取文档中提到的所有行动建议
2. 整理成简洁的行动项
3. 每个行动项以动词开头
4. 用中文输出
5. 每条一行，不要编号
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个专业的咨询顾问。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=300
)
items_text = response.choices[0].message.content
items = [item.strip() for item in items_text.split('\n') if item.strip()]
return items
# ==================== 格式化输出 ====================
def format_result(result: ProcessingResult) -> str:
"""格式化输出处理结果"""
output = []
output.append("\n" + "=" * 60)
output.append("文档处理报告")
output.append("=" * 60)
# 元数据
output.append("\n【文档信息】")
output.append(f" 文件名: {result.metadata.filename}")
output.append(f" 页数: {result.metadata.page_count}")
output.append(f" 语言: {result.metadata.language}")
output.append(f" 大小: {result.metadata.file_size} 字节")
# 原文摘要
output.append("\n【原文摘要】")
output.append(f" {result.original_summary}")
# 翻译摘要
output.append("\n【翻译摘要】")
output.append(f" {result.translated_summary}")
# 关键发现
output.append("\n【关键发现】")
for i, finding in enumerate(result.key_findings, 1):
output.append(f" {i}. {finding}")
# 行动项
if result.action_items:
output.append("\n【行动项】")
for item in result.action_items:
output.append(f" • {item}")
output.append("\n" + "=" * 60)
return "\n".join(output)
# ==================== 使用示例 ====================
def main():
"""主函数"""
print("\n" + "=" * 60)
print("文档处理流水线 - 使用演示")
print("=" * 60)
# 创建流水线
pipeline = DocumentPipeline(target_language="Chinese")
# 处理文档
result = pipeline.process("sample_research_paper.pdf")
# 输出结果
print(format_result(result))
if __name__ == "__main__":
main()
案例三：构建智能客服对话系统
本案例展示如何构建一个基于JARVIS的智能客服系统，支持多轮对话、意图识别、知识库检索和情感分析。
完整代码：
# smart_customer_service.py
"""
智能客服对话系统
功能：多轮对话、意图识别、知识库检索、情感分析、满意度预测
"""
import os
from datetime import datetime
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
from dotenv import load_dotenv
load_dotenv()
# ==================== 数据模型 ====================
class Intent(Enum):
"""用户意图枚举"""
GREETING = "greeting" # 问候
PRODUCT_INQUIRY = "product" # 产品咨询
ORDER_STATUS = "order" # 订单查询
COMPLAINT = "complaint" # 投诉
REFUND = "refund" # 退款
TECHNICAL_SUPPORT = "technical" # 技术支持
RECOMMENDATION = "recommend" # 推荐请求
FAREWELL = "farewell" # 道别
UNKNOWN = "unknown" # 未知
class Sentiment(Enum):
"""情感极性枚举"""
POSITIVE = "positive"
NEUTRAL = "neutral"
NEGATIVE = "negative"
@dataclass
class ConversationTurn:
"""对话轮次"""
timestamp: datetime
role: str # "user" or "assistant"
content: str
intent: Optional[Intent] = None
sentiment: Optional[Sentiment] = None
@dataclass
class UserProfile:
"""用户画像"""
user_id: str
name: Optional[str] = None
tier: str = "standard" # standard, premium, vip
total_orders: int = 0
preferences: List[str] = field(default_factory=list)
@dataclass
class ConversationContext:
"""对话上下文"""
user_profile: UserProfile
current_intent: Optional[Intent] = None
conversation_history: List[ConversationTurn] = field(default_factory=list)
pending_info: Dict[str, str] = field(default_factory=dict) # 待确认信息
session_start: datetime = field(default_factory=datetime.now)
# ==================== 核心系统类 ====================
class IntentClassifier:
"""意图分类器"""
def __init__(self, llm_client):
self.llm = llm_client
self.model = "gpt-4"
# 意图关键词映射（用于快速匹配）
self.intent_keywords = {
Intent.GREETING: ["你好", "hi", "hello", "在吗", "早上好", "下午好"],
Intent.PRODUCT_INQUIRY: ["产品", "功能", "规格", "参数", "介绍"],
Intent.ORDER_STATUS: ["订单", "发货", "物流", "到了吗", "什么时候"],
Intent.COMPLAINT: ["投诉", "不满", "太差", "问题", "糟糕"],
Intent.REFUND: ["退款", "退货", "取消订单", "换货"],
Intent.TECHNICAL_SUPPORT: ["故障", "报错", "不能", "无法", "问题"],
Intent.RECOMMENDATION: ["推荐", "建议", "适合", "想买"],
Intent.FAREWELL: ["再见", "拜拜", "谢谢", "结束"]
}
def classify(self, text: str, context: ConversationContext) -> Tuple[Intent, float]:
"""
分类用户意图
参数:
text: 用户输入文本
context: 对话上下文
返回:
(意图, 置信度)
"""
# 步骤1：快速关键词匹配
for intent, keywords in self.intent_keywords.items():
for keyword in keywords:
if keyword.lower() in text.lower():
return intent, 0.8
# 步骤2：LLM精确分类
prompt = self._build_classification_prompt(text, context)
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个客服意图分类专家。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=100
)
result = response.choices[0].message.content
# 解析LLM输出
intent_str = result.strip()
try:
intent = Intent(intent_str)
confidence = 0.9
except ValueError:
intent = Intent.UNKNOWN
confidence = 0.5
return intent, confidence
def _build_classification_prompt(self, text: str, context: ConversationContext) -> str:
"""构建分类提示词"""
# 构建对话历史摘要
history_summary = ""
if context.conversation_history:
recent = context.conversation_history[-3:]
history_summary = "\n对话历史：\n"
for turn in recent:
role = "用户" if turn.role == "user" else "助手"
history_summary += f"- {role}: {turn.content[:50]}...\n"
return f"""判断用户最新输入的意图类型。
用户最新输入：{text}
{history_summary}
用户等级：{context.user_profile.tier}
可选意图类型：
- greeting: 问候、打招呼
- product: 产品咨询（功能、价格、规格等）
- order: 订单相关（查询、修改、取消）
- complaint: 投诉、抱怨
- refund: 退款退货请求
- technical: 技术问题、故障排查
- recommend: 产品推荐请求
- farewell: 道别
- unknown: 无法判断
输出要求：
只输出意图类型（如"product"），不要其他内容。
"""
class SentimentAnalyzer:
"""情感分析器"""
def __init__(self, llm_client):
self.llm = llm_client
self.model = "gpt-4"
def analyze(self, text: str) -> Tuple[Sentiment, float]:
"""
分析文本情感
返回:
(情感极性, 强度分数0-1)
"""
prompt = f"""分析以下文本的情感倾向和强度。
文本：{text}
情感分类：
- positive: 积极正面
- neutral: 中性
- negative: 消极负面
输出格式：
情感类型,强度分数
示例：
positive,0.8
要求：只输出一行，不要其他内容。
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个情感分析专家。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=50
)
result = response.choices[0].message.content.strip()
try:
sentiment_str, score_str = result.split(',')
sentiment = Sentiment(sentiment_str.strip())
score = float(score_str.strip())
except (ValueError, IndexError):
sentiment = Sentiment.NEUTRAL
score = 0.5
return sentiment, score
class KnowledgeBase:
"""知识库（简化实现）"""
def __init__(self):
# 产品信息
self.products = {
"笔记本电脑": {
"price": "5999-12999元",
"features": ["高性能CPU", "长续航", "轻薄设计"],
"warranty": "2年质保"
},
"无线耳机": {
"price": "299-999元",
"features": ["主动降噪", "防水", "30小时续航"],
"warranty": "1年质保"
},
"智能手表": {
"price": "1599-3999元",
"features": ["健康监测", "GPS", "NFC支付"],
"warranty": "1年质保"
}
}
# FAQ
self.faqs = {
"配送时间": "标准配送3-5个工作日，加急配送1-2个工作日。",
"退换政策": "7天内可无理由退换，15天内可换货。",
"支付方式": "支持支付宝、微信支付、信用卡、货到付款。",
"发票": "下单时可选择电子发票或纸质发票。"
}
def search(self, query: str, intent: Intent) -> Optional[str]:
"""检索知识库"""
# 根据意图选择搜索策略
if intent == Intent.PRODUCT_INQUIRY:
for product_name, info in self.products.items():
if product_name in query:
return self._format_product_info(product_name, info)
elif intent == Intent.TECHNICAL_SUPPORT:
# 模拟技术问题解答
return "请问您遇到的具体问题是什么？您可以：\n1. 重启设备\n2. 检查网络连接\n3. 更新到最新版本\n\n如果问题仍存在，请提供错误代码。"
# 通用FAQ搜索
for key, value in self.faqs.items():
if key in query:
return value
return None
def _format_product_info(self, name: str, info: dict) -> str:
"""格式化产品信息"""
features = "、".join(info["features"])
return f"""【{name}】
价格：{info['price']}
特点：{features}
质保：{info['warranty']}"""
class ResponseGenerator:
"""回复生成器"""
def __init__(self, llm_client, knowledge_base):
self.llm = llm_client
self.model = "gpt-4"
self.kb = knowledge_base
def generate(self, text: str, intent: Intent,
sentiment: Sentiment, context: ConversationContext,
knowledge_answer: Optional[str] = None) -> str:
"""
生成回复
参数:
text: 用户输入
intent: 识别的意图
sentiment: 情感分析结果
context: 对话上下文
knowledge_answer: 知识库检索结果
"""
# 构建prompt
prompt = self._build_response_prompt(
text, intent, sentiment, context, knowledge_answer
)
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个专业、友好的客服助手。"},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=500
)
return response.choices[0].message.content
def _build_response_prompt(self, text: str, intent: Intent,
sentiment: Sentiment, context: ConversationContext,
knowledge_answer: Optional[str]) -> str:
"""构建回复提示词"""
# 情感调整
emotion_adjustment = ""
if sentiment == Sentiment.NEGATIVE:
emotion_adjustment = "用户情绪偏负面，请表达理解和歉意，语气温和有耐心。"
elif sentiment == Sentiment.POSITIVE:
emotion_adjustment = "用户情绪积极，可以适当热情回应。"
# 上下文信息
context_info = ""
if context.conversation_history:
context_info = f"\n用户信息：{context.user_profile.tier}会员"
# 知识库答案
kb_section = ""
if knowledge_answer:
kb_section = f"\n参考信息：\n{knowledge_answer}"
return f"""你是一个电商平台的智能客服助手。请根据用户输入生成合适的回复。
用户输入：{text}
用户意图：{intent.value}
用户情绪：{sentiment.value}
{context_info}
{kb_section}
回复要求：
1. {emotion_adjustment}
2. 专业、友好、有耐心
3. 简洁明了，不超过200字
4. 如需更多信息，主动询问用户
5. 用中文回复
"""
# ==================== 主系统类 ====================
class SmartCustomerService:
"""
智能客服系统
整合意图识别、情感分析、知识库检索、回复生成
"""
def __init__(self):
print("=" * 50)
print("初始化智能客服系统...")
# 初始化LLM客户端
from openai import OpenAI
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise ValueError("未设置 OPENAI_API_KEY")
self.llm = OpenAI(api_key=api_key)
print("✓ LLM客户端初始化完成")
# 初始化各组件
self.intent_classifier = IntentClassifier(self.llm)
self.sentiment_analyzer = SentimentAnalyzer(self.llm)
self.knowledge_base = KnowledgeBase()
self.response_generator = ResponseGenerator(self.llm, self.knowledge_base)
print("✓ 组件初始化完成")
# 当前会话上下文
self.current_context: Optional[ConversationContext] = None
print("=" * 50)
print("智能客服系统就绪！")
def start_session(self, user_id: str, user_tier: str = "standard") -> str:
"""开始新会话"""
self.current_context = ConversationContext(
user_profile=UserProfile(
user_id=user_id,
tier=user_tier
)
)
welcome_msg = """您好！我是智能客服小JARVIS，很高兴为您服务。
我可以帮您：
• 解答产品相关问题
• 查询订单状态
• 处理退换货事宜
• 提供技术支持
• 推荐适合的产品
请问有什么可以帮您？"""
# 记录助手回复
self.current_context.conversation_history.append(
ConversationTurn(
timestamp=datetime.now(),
role="assistant",
content=welcome_msg
)
)
return welcome_msg
def chat(self, user_input: str) -> str:
"""
处理用户输入并生成回复
参数:
user_input: 用户输入
返回:
助手回复
"""
if not self.current_context:
return "请先调用 start_session() 开始会话。"
print(f"\n{'='*50}")
print(f"用户: {user_input}")
# 记录用户输入
user_turn = ConversationTurn(
timestamp=datetime.now(),
role="user",
content=user_input
)
# 情感分析
sentiment, sentiment_score = self.sentiment_analyzer.analyze(user_input)
user_turn.sentiment = sentiment
print(f"情感分析: {sentiment.value} ({sentiment_score:.2f})")
# 意图识别
intent, intent_confidence = self.intent_classifier.classify(
user_input, self.current_context
)
user_turn.intent = intent
self.current_context.current_intent = intent
print(f"意图识别: {intent.value} ({intent_confidence:.2f})")
# 知识库检索
knowledge_answer = self.knowledge_base.search(user_input, intent)
if knowledge_answer:
print("知识库命中!")
# 生成回复
response = self.response_generator.generate(
user_input, intent, sentiment,
self.current_context, knowledge_answer
)
# 记录助手回复
assistant_turn = ConversationTurn(
timestamp=datetime.now(),
role="assistant",
content=response,
intent=intent,
sentiment=sentiment
)
self.current_context.conversation_history.append(user_turn)
self.current_context.conversation_history.append(assistant_turn)
print(f"\n助手: {response}")
print(f"{'='*50}")
return response
def end_session(self) -> Dict:
"""结束会话，返回会话摘要"""
if not self.current_context:
return {"error": "没有进行中的会话"}
# 生成会话摘要
total_turns = len(self.current_context.conversation_history)
# 统计意图分布
intent_counts = {}
sentiment_counts = {"positive": 0, "neutral": 0, "negative": 0}
for turn in self.current_context.conversation_history:
if turn.intent:
intent_counts[turn.intent.value] = \
intent_counts.get(turn.intent.value, 0) + 1
if turn.sentiment:
sentiment_counts[turn.sentiment.value] += 1
summary = {
"session_id": self.current_context.user_profile.user_id,
"duration": (datetime.now() - self.current_context.session_start).seconds,
"total_turns": total_turns,
"intent_distribution": intent_counts,
"sentiment_distribution": sentiment_counts,
"user_tier": self.current_context.user_profile.tier
}
self.current_context = None
return summary
# ==================== 使用演示 ====================
def main():
"""主函数：演示智能客服系统"""
print("\n" + "=" * 60)
print("智能客服系统 - 使用演示")
print("=" * 60)
# 创建客服系统
service = SmartCustomerService()
# 开始会话
print("\n>>> 开启新会话")
service.start_session(user_id="user_12345", user_tier="premium")
# 模拟多轮对话
test_inputs = [
"你好，我想了解一下你们的笔记本电脑",
"续航能力怎么样？",
"我想买一台适合办公的，有什么推荐吗？",
"好的，那这款电脑支持七天无理由退货吗？",
"谢谢，再见"
]
for user_input in test_inputs:
if user_input.lower() in ["再见", "拜拜", "退出", "quit", "exit"]:
break
service.chat(user_input)
# 结束会话，获取摘要
print("\n>>> 会话结束")
summary = service.end_session()
print("\n" + "-" * 50)
print("会话摘要：")
print(f" 会话时长: {summary['duration']} 秒")
print(f" 对话轮次: {summary['total_turns']}")
print(f" 用户等级: {summary['user_tier']}")
print(f" 意图分布: {summary['intent_distribution']}")
print(f" 情感分布: {summary['sentiment_distribution']}")
print("-" * 50)
if __name__ == "__main__":
main()
第五部分：常见使用场景与最佳实践
场景一：内容创作自动化
JARVIS可以极大地简化内容创作流程。以下是一个完整的自动化创作流水线：
# content_creation_pipeline.py
"""
内容创作自动化流水线
"""
import os
from typing import List, Dict
from dataclasses import dataclass
from dotenv import load_dotenv
load_dotenv()
@dataclass
class ContentRequest:
"""内容创作请求"""
topic: str
content_type: str # "article", "social_media", "video_script"
target_audience: str
tone: str # "formal", "casual", "humorous"
length: str # "short", "medium", "long"
keywords: List[str]
@dataclass
class ContentResult:
"""创作结果"""
main_content: str
headline: str
meta_description: str
hashtags: List[str]
call_to_action: str
class ContentCreationPipeline:
"""
内容创作流水线
完整流程：
1. 主题研究 → 生成大纲
2. 内容撰写 → 多版本输出
3. SEO优化 → 关键词布局
4. 社交媒体适配 → 多平台版本
"""
def __init__(self):
from openai import OpenAI
self.llm = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.model = "gpt-4"
def create_content(self, request: ContentRequest) -> ContentResult:
"""创建完整内容"""
# 步骤1：研究主题，生成大纲
outline = self._generate_outline(request)
print(f"✓ 大纲生成完成")
# 步骤2：撰写主内容
main_content = self._write_content(outline, request)
print(f"✓ 主内容撰写完成 ({len(main_content)} 字符)")
# 步骤3：生成标题和副标题
headline = self._generate_headline(request, main_content)
print(f"✓ 标题生成完成")
# 步骤4：生成Meta描述
meta_description = self._generate_meta_description(main_content, request)
print(f"✓ Meta描述生成完成")
# 步骤5：生成社交媒体标签
hashtags = self._generate_hashtags(request)
print(f"✓ Hashtag生成完成")
# 步骤6：生成行动号召
cta = self._generate_call_to_action(request)
print(f"✓ CTA生成完成")
return ContentResult(
main_content=main_content,
headline=headline,
meta_description=meta_description,
hashtags=hashtags,
call_to_action=cta
)
def _generate_outline(self, request: ContentRequest) -> str:
"""生成内容大纲"""
prompt = f"""为以下主题生成一篇{request.length}长度文章的详细大纲。
主题：{request.topic}
目标受众：{request.target_audience}
内容类型：{request.content_type}
风格：{request.tone}
要求：
1. 生成5-7个主要章节
2. 每个章节包含2-3个子要点
3. 包含引言和结论
4. 用中文输出
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个专业的内容策划专家。"},
{"role": "user", "content": prompt}
],
temperature=0.5,
max_tokens=800
)
return response.choices[0].message.content
def _write_content(self, outline: str, request: ContentRequest) -> str:
"""撰写正文"""
# 根据长度确定字数
length_map = {"short": "800-1000", "medium": "1500-2000", "long": "3000-4000"}
target_length = length_map.get(request.length, "1500-2000")
prompt = f"""根据以下大纲撰写一篇完整的文章。
大纲：
{outline}
要求：
- 字数：约{target_length}字
- 风格：{request.tone}
- 目标受众：{request.target_audience}
- 自然融入关键词：{', '.join(request.keywords)}
- 结构清晰，段落分明
- 内容有深度，有实用价值
- 用中文写作
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个资深的内容创作者，擅长撰写高质量的原创文章。"},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=3000
)
return response.choices[0].message.content
def _generate_headline(self, request: ContentRequest, content: str) -> str:
"""生成吸引人的标题"""
prompt = f"""基于以下文章内容，生成5个吸引人的标题。
文章主题：{request.topic}
文章摘要：{content[:500]}
要求：
1. 每个标题不超过25字
2. 标题要吸引眼球，激发点击欲望
3. 可以使用数字、疑问、对比等技巧
4. 涵盖核心内容
5. 用中文
输出格式：
1. 标题1
2. 标题2
3. 标题3
4. 标题4
5. 标题5
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个标题党大师，擅长创作高点击率的标题。"},
{"role": "user", "content": prompt}
],
temperature=0.8,
max_tokens=400
)
return response.choices[0].message.content
def _generate_meta_description(self, content: str,
request: ContentRequest) -> str:
"""生成SEO Meta描述"""
prompt = f"""为以下文章生成一段SEO友好的Meta描述。
文章内容摘要：{content[:800]}
要求：
1. 长度：150-160字符
2. 包含核心关键词：{', '.join(request.keywords[:3])}
3. 吸引点击，有行动导向
4. 用中文
只输出一段描述，不要其他内容。
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个SEO专家。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=200
)
return response.choices[0].message.content
def _generate_hashtags(self, request: ContentRequest) -> List[str]:
"""生成社交媒体标签"""
prompt = f"""为以下主题生成适合社交媒体的hashtag标签。
主题：{request.topic}
关键词：{', '.join(request.keywords)}
要求：
1. 生成8-12个标签
2. 包含中英文标签
3. 热门标签和长尾标签结合
4. 不要加#号，直接输出标签词
5. 用逗号分隔
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个社交媒体营销专家。"},
{"role": "user", "content": prompt}
],
temperature=0.5,
max_tokens=200
)
tags = [t.strip() for t in response.choices[0].message.content.split('，') if t.strip()]
return tags
def _generate_call_to_action(self, request: ContentRequest) -> str:
"""生成行动号召"""
prompt = f"""为以下主题的文章生成一个有力的行动号召（CTA）。
主题：{request.topic}
内容类型：{request.content_type}
目标受众：{request.target_audience}
要求：
1. 简洁有力，不超过30字
2. 明确告诉读者下一步做什么
3. 符合目标受众的期望
4. 用中文
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个转化率优化专家。"},
{"role": "user", "content": prompt}
],
temperature=0.5,
max_tokens=100
)
return response.choices[0].message.content
# 使用示例
if __name__ == "__main__":
pipeline = ContentCreationPipeline()
request = ContentRequest(
topic="人工智能在教育领域的应用",
content_type="article",
target_audience="教育工作者和家长",
tone="professional",
length="medium",
keywords=["AI教育", "智能学习", "个性化教学", "教育科技"]
)
result = pipeline.create_content(request)
print("\n" + "=" * 60)
print("创作结果")
print("=" * 60)
print(f"\n【标题】\n{result.headline}")
print(f"\n【正文】\n{result.main_content[:500]}...")
print(f"\n【Meta描述】\n{result.meta_description}")
print(f"\n【Hashtags】\n{', '.join(result.hashtags)}")
print(f"\n【行动号召】\n{result.call_to_action}")
场景二：数据分析与报表生成
# data_analysis_pipeline.py
"""
数据分析与报表生成流水线
"""
import os
from typing import List, Dict
from dataclasses import dataclass
from datetime import datetime
from dotenv import load_dotenv
load_dotenv()
@dataclass
class DataSummary:
"""数据摘要"""
total_records: int
date_range: str
key_metrics: Dict[str, float]
trends: List[str]
anomalies: List[str]
@dataclass
class AnalysisReport:
"""分析报告"""
title: str
summary: str
detailed_analysis: str
visualizations: List[str]
recommendations: List[str]
risk_alerts: List[str]
class DataAnalysisPipeline:
"""
数据分析流水线
完整流程：
1. 数据清洗和验证
2. 描述性统计分析
3. 趋势分析
4. 异常检测
5. 洞察提取
6. 报告生成
"""
def __init__(self):
from openai import OpenAI
self.llm = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.model = "gpt-4"
def analyze(self, data: List[Dict], report_type: str = "executive") -> AnalysisReport:
"""
分析数据并生成报告
参数:
data: 数据列表（字典格式）
report_type: 报告类型 ("executive", "detailed", "technical")
"""
print("开始数据分析...")
# 步骤1：数据验证
print("[1/6] 验证数据...")
data_quality = self._validate_data(data)
print(f" 数据质量得分: {data_quality['quality_score']:.2f}")
# 步骤2：统计摘要
print("[2/6] 生成统计摘要...")
summary = self._generate_summary(data)
print(f" 处理记录数: {summary.total_records}")
# 步骤3：趋势分析
print("[3/6] 分析趋势...")
trends = self._analyze_trends(data)
print(f" 发现趋势: {len(trends)}个")
# 步骤4：异常检测
print("[4/6] 检测异常...")
anomalies = self._detect_anomalies(data)
print(f" 发现异常: {len(anomalies)}个")
# 步骤5：洞察提取
print("[5/6] 提取洞察...")
insights = self._extract_insights(data, summary, trends, anomalies)
print(f" 生成洞察: {len(insights)}条")
# 步骤6：生成报告
print("[6/6] 生成报告...")
report = self._generate_report(
data, summary, trends, anomalies, insights, report_type
)
print(" 报告生成完成")
return report
def _validate_data(self, data: List[Dict]) -> Dict:
"""验证数据质量"""
total = len(data)
if total == 0:
return {"quality_score": 0, "issues": ["无数据"]}
# 检查基本字段
required_fields = list(data[0].keys()) if data else []
issues = []
valid_count = 0
for record in data:
if all(record.get(f) for f in required_fields[:3]):
valid_count += 1
quality_score = valid_count / total if total > 0 else 0
return {
"quality_score": quality_score,
"total_records": total,
"valid_records": valid_count,
"issues": issues
}
def _generate_summary(self, data: List[Dict]) -> DataSummary:
"""生成数据摘要"""
if not data:
return DataSummary(
total_records=0,
date_range="N/A",
key_metrics={},
trends=[],
anomalies=[]
)
# 计算基本统计
total_records = len(data)
# 尝试识别数值字段并计算均值
key_metrics = {}
numeric_fields = []
for key in data[0].keys():
if key.lower() in ["amount", "value", "price", "count", "revenue", "cost"]:
values = [d.get(key, 0) for d in data if isinstance(d.get(key), (int, float))]
if values:
key_metrics[f"avg_{key}"] = sum(values) / len(values)
key_metrics[f"max_{key}"] = max(values)
key_metrics[f"min_{key}"] = min(values)
return DataSummary(
total_records=total_records,
date_range="最近30天",
key_metrics=key_metrics,
trends=[],
anomalies=[]
)
def _analyze_trends(self, data: List[Dict]) -> List[str]:
"""分析数据趋势"""
prompt = f"""分析以下数据的趋势模式。
数据样本（前20条）：
{str(data[:20])}
请识别：
1. 增长/下降趋势
2. 周期性模式
3. 季节性变化
4. 拐点和异常波动
输出3-5个主要趋势，用中文描述，每条一行。
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个数据分析专家。"},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=400
)
trends = [t.strip() for t in response.choices[0].message.content.split('\n') if t.strip()]
return trends
def _detect_anomalies(self, data: List[Dict]) -> List[str]:
"""检测数据异常"""
# 简单的统计异常检测
anomalies = []
# 检查数值字段的异常值
for key in data[0].keys():
if isinstance(data[0][key], (int, float)):
values = [d.get(key, 0) for d in data if isinstance(d.get(key), (int, float))]
if values:
mean = sum(values) / len(values)
std = (sum((v - mean) ** 2 for v in values) / len(values)) ** 0.5
extreme_values = [v for v in values if abs(v - mean) > 3 * std]
if extreme_values:
anomalies.append(f"{key}字段存在{len(extreme_values)}个极端值")
return anomalies
def _extract_insights(self, data: List[Dict], summary: DataSummary,
trends: List[str], anomalies: List[str]) -> List[str]:
"""提取数据洞察"""
prompt = f"""基于以下数据分析结果，提取关键洞察。
数据统计：
- 记录总数：{summary.total_records}
- 关键指标：{summary.key_metrics}
趋势分析：
{chr(10).join(trends) if trends else '无明显趋势'}
异常检测：
{chr(10).join(anomalies) if anomalies else '未发现明显异常'}
请提取5-8个关键洞察：
1. 关注数据中的关键发现
2. 解释趋势的可能原因
3. 指出潜在机会或风险
4. 用简洁有力的语言
每条洞察不超过50字，用中文。
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个资深的数据分析师，擅长从数据中提取有价值的洞察。"},
{"role": "user", "content": prompt}
],
temperature=0.5,
max_tokens=600
)
insights = [i.strip() for i in response.choices[0].message.content.split('\n') if i.strip()]
return insights
def _generate_report(self, data: List[Dict], summary: DataSummary,
trends: List[str], anomalies: List[str],
insights: List[str], report_type: str) -> AnalysisReport:
"""生成完整报告"""
prompt = f"""生成一份{report_type}类型的数据分析报告。
报告基本信息：
- 标题：数据分析报告
- 日期：{datetime.now().strftime('%Y年%m月%d日')}
- 数据范围：{summary.date_range}
- 记录数：{summary.total_records}
关键指标：
{chr(10).join([f"- {k}: {v:.2f}" for k, v in summary.key_metrics.items()])}
趋势分析：
{chr(10).join([f"- {t}" for t in trends])}
异常情况：
{chr(10).join([f"- {a}" for a in anomalies]) if anomalies else '- 无'}
关键洞察：
{chr(10).join([f"- {i}" for i in insights])}
报告结构要求（根据类型）：
- executive: 简洁摘要版，重点在结论和建议，适合高管阅读
- detailed: 详细分析版，包含完整数据和分析过程
- technical: 技术报告版，重点在方法和模型，适合技术人员
输出完整报告内容，用中文。
"""
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个专业的商业分析师，擅长撰写高质量的数据分析报告。"},
{"role": "user", "content": prompt}
],
temperature=0.5,
max_tokens=2000
)
# 提取各部分（简化处理）
content = response.choices[0].message.content
lines = content.split('\n')
return AnalysisReport(
title="数据分析报告",
summary="\n".join(lines[:10]) if len(lines) > 10 else content,
detailed_analysis=content,
visualizations=["趋势图", "分布图", "对比图"],
recommendations=insights[:3] if insights else [],
risk_alerts=anomalies if anomalies else []
)
# 使用示例
if __name__ == "__main__":
# 模拟销售数据
sample_data = [
{"date": "2024-01-01", "product": "产品A", "amount": 1000, "region": "华东"},
{"date": "2024-01-02", "product": "产品A", "amount": 1200, "region": "华东"},
{"date": "2024-01-03", "product": "产品B", "amount": 800, "region": "华南"},
{"date": "2024-01-04", "product": "产品B", "amount": 950, "region": "华南"},
{"date": "2024-01-05", "product": "产品A", "amount": 1500, "region": "华北"},
]
pipeline = DataAnalysisPipeline()
report = pipeline.analyze(sample_data, report_type="executive")
print("\n" + "=" * 60)
print("分析报告")
print("=" * 60)
print(f"\n【标题】{report.title}")
print(f"\n【摘要】{report.summary}")
print(f"\n【建议】")
for rec in report.recommendations:
print(f" • {rec}")
最佳实践总结
一、提示词工程最佳实践
- 结构化提示词：使用清晰的系统提示、任务描述、输出格式要求
- few-shot示例：提供2-3个示例帮助模型理解任务
- 温度控制：创意任务用0.7-0.9，精确任务用0.1-0.3
- 角色设定：让LLM扮演专家角色提升输出质量
# 最佳实践示例
SYSTEM_PROMPT = """你是一个[专业角色]，拥有[领域]的深厚背景知识。
你的回答应该：
1. 专业、准确、有深度
2. 简洁明了，逻辑清晰
3. 结合实际案例
4. 如有不确定，明确指出"""
TASK_PROMPT = """任务：[具体描述]
输入：
{user_input}
要求：
1. [具体要求1]
2. [具体要求2]
3. [具体要求3]
输出格式：
[期望的输出格式描述]"""
二、错误处理与容错
- 设置重试机制：API调用失败时自动重试
- 超时控制：避免无限等待
- 降级策略：主模型不可用时切换到备用模型
import time
from functools import wraps
def retry_on_failure(max_retries=3, delay=1):
"""重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise
print(f"尝试 {attempt + 1} 失败: {e}")
time.sleep(delay * (attempt + 1))
return None
return wrapper
return decorator
三、性能优化
- 批量处理：将多个相似请求合并
- 缓存结果：避免重复计算
- 异步调用：充分利用等待时间
- 模型选择：简单任务用小模型，复杂任务用大模型
第六部分：进阶技巧与扩展开发
自定义模型接入
JARVIS支持接入自定义模型，以下是接入新模型的完整流程：
# custom_model_integration.py
"""
自定义模型接入示例
"""
from jarvis.models import ModelRegistry, ModelInfo, ModelCapability
from jarvis.executor import TaskExecutor
class CustomModelIntegration:
"""自定义模型接入"""
def __init__(self):
self.registry = ModelRegistry()
def register_custom_model(self, model_id: str, model_type: str,
api_endpoint: str, api_key: str):
"""
注册自定义模型
参数:
model_id: 模型标识符
model_type: 任务类型
api_endpoint: API端点
api_key: API密钥
"""
model_info = ModelInfo(
model_id=model_id,
provider="custom",
endpoint=api_endpoint,
capability=ModelCapability(
task_type=model_type,
input_format="json",
output_format="json",
max_input_length=4096,
strengths=["自定义优势"],
limitations=["自定义限制"]
),
api_key_env=None, # 或指定环境变量名
is_local=False
)
self.registry.register(model_info)
print(f"✓ 自定义模型 {model_id} 注册成功")
return model_info
def create_custom_executor(self, model_id: str):
"""为自定义模型创建执行器"""
model_info = self.registry.get(model_id)
if not model_info:
raise ValueError(f"模型 {model_id} 未注册")
# 创建自定义执行逻辑
async def custom_execute(inputs: dict) -> dict:
"""自定义执行逻辑"""
import aiohttp
headers = {
"Authorization": f"Bearer {model_info.api_key_env}",
"Content-Type": "application/json"
}
async with aiohttp.ClientSession() as session:
async with session.post(
model_info.endpoint,
json=inputs,
headers=headers
) as response:
result = await response.json()
return result
return custom_execute
# 使用示例
if __name__ == "__main__":
integration = CustomModelIntegration()
# 注册一个假设的自定义NLP模型
integration.register_custom_model(
model_id="my-custom-nlp-model",
model_type="text-analysis",
api_endpoint="https://api.example.com/v1/analyze",
api_key="your-api-key"
)
构建自定义Web界面
JARVIS支持通过Gradio快速构建交互界面：
# jarvis_web_app.py
"""
JARVIS Web交互界面
"""
import os
import gradio as gr
from dotenv import load_dotenv
load_dotenv()
# 导入JARVIS组件
from jarvis import JARVIS
class JARVISWebInterface:
"""JARVIS Web界面"""
def __init__(self):
self.jarvis = JARVIS()
self.current_mode = "chat"
def chat_mode(self, user_input: str, history: list) -> tuple:
"""
对话模式
返回: (更新后的历史记录, 状态消息)
"""
if not user_input.strip():
return history, ""
try:
# 调用JARVIS处理
response = self.jarvis.chat(user_input)
# 更新历史
history.append((user_input, response))
return history, ""
except Exception as e:
return history, f"错误: {str(e)}"
def analysis_mode(self, text: str, task_type: str) -> str:
"""
分析模式
参数:
text: 待分析文本
task_type: 分析类型
"""
if not text.strip():
return "请输入要分析的内容"
try:
if task_type == "sentiment":
result = self._analyze_sentiment(text)
elif task_type == "summary":
result = self._summarize_text(text)
elif task_type == "keywords":
result = self._extract_keywords(text)
else:
result = "不支持的分析类型"
return result
except Exception as e:
return f"分析失败: {str(e)}"
def _analyze_sentiment(self, text: str) -> str:
"""情感分析"""
# 简化实现
return f"情感分析结果: 中性 (置信度: 0.85)"
def _summarize_text(self, text: str) -> str:
"""文本摘要"""
# 简化实现
return f"摘要: {text[:100]}..."
def _extract_keywords(self, text: str) -> str:
"""关键词提取"""
# 简化实现
return "关键词: 主题1, 主题2, 主题3"
def create_interface() -> gr.Blocks:
"""创建Gradio界面"""
app = JARVISWebInterface()
# 定义主题CSS
theme_css = """
.gradio-container {
font-family: 'PingFang SC', 'Microsoft YaHei', sans-serif;
}
.main-title {
text-align: center;
color: #2c3e50;
font-size: 2em;
margin-bottom: 20px;
}
"""
with gr.Blocks(
title="JARVIS AI Assistant",
theme=gr.themes.Soft(),
css=theme_css
) as demo:
# 标题
gr.Markdown("""
# 🤖 JARVIS 智能助手
欢迎使用JARVIS！一个强大的多模型协作平台。
---
""")
# 模式选择
with gr.Tab("💬 智能对话"):
gr.Markdown("### 与JARVIS进行自然语言对话")
chatbot = gr.Chatbot(label="对话历史", height=400)
user_input = gr.Textbox(
label="输入",
placeholder="请输入您的问题...",
lines=3
)
with gr.Row():
submit_btn = gr.Button("发送", variant="primary")
clear_btn = gr.Button("清空")
status_output = gr.Textbox(label="状态", interactive=False)
submit_btn.click(
app.chat_mode,
inputs=[user_input, chatbot],
outputs=[chatbot, status_output]
)
user_input.submit(
app.chat_mode,
inputs=[user_input, chatbot],
outputs=[chatbot, status_output]
)
clear_btn.click(
lambda: ([], ""),
outputs=[chatbot, status_output]
)
# 分析工具
with gr.Tab("📊 文本分析"):
gr.Markdown("### 多种文本分析工具")
analysis_text = gr.Textbox(
label="待分析文本",
placeholder="输入要分析的文本...",
lines=6
)
analysis_type = gr.Dropdown(
label="分析类型",
choices=["sentiment", "summary", "keywords"],
value="sentiment"
)
analysis_btn = gr.Button("开始分析", variant="primary")
analysis_output = gr.Textbox(
label="分析结果",
lines=8,
interactive=False
)
analysis_btn.click(
app.analysis_mode,
inputs=[analysis_text, analysis_type],
outputs=analysis_output
)
# 设置面板
with gr.Tab("⚙️ 设置"):
gr.Markdown("### 配置选项")
with gr.Row():
model_select = gr.Dropdown(
label="选择模型",
choices=["gpt-4", "gpt-3.5-turbo", "claude-3"],
value="gpt-4"
)
temperature = gr.Slider(
label="创造性 (Temperature)",
minimum=0.0,
maximum=1.0,
value=0.7,
step=0.1
)
gr.Markdown("---")
gr.Markdown("* 设置将在下次对话时生效")
# 页脚
gr.Markdown("""
---
### 使用说明
1. **智能对话**: 直接用自然语言提问，JARVIS会自动规划任务
2. **文本分析**: 选择分析类型，输入文本获取分析结果
3. **设置**: 调整模型参数以获得不同的输出风格
---
Powered by Microsoft JARVIS
""")
return demo
# 启动应用
if __name__ == "__main__":
demo = create_interface()
demo.launch(
server_name="0.0.0.0",
server_port=7860,
share=False
)
结语：拥抱多模型协作的未来
JARVIS代表了一种新的AI应用范式——不再是单一模型的”全能”解决方案，而是通过智能调度多个专业化模型，让AI真正成为可协作的助手。这种”各尽所能、按需调度”的模式，既符合当前AI技术的发展现状，也最大化地发挥了每个模型的优势。
通过本文的详细讲解，你应该已经掌握了：
- JARVIS的核心架构和工作原理
- 如何搭建开发环境
- 如何构建多模态内容处理流水线
- 如何构建智能客服系统
- 如何进行数据分析和报表生成
- 如何进行最佳实践和性能优化
- 如何扩展和自定义JARVIS
相关资源推荐
官方资源
- JARVIS GitHub仓库：https://github.com/microsoft/JARVIS
- 官方文档和示例代码
- Hugging Face模型库：https://huggingface.co/models
延伸阅读
- HuggingGPT论文：研究JARVIS背后的学术思想
- LangChain文档：另一种构建LLM应用的方式
- Prompt Engineering Guide：提升提示词编写能力
相关开源项目
- LangChain：构建LLM应用的流行框架
- AutoGPT：自主任务执行的实验性项目
- ChatGLM：国产开源对话模型
- LLaMA：Meta开源的大语言模型
AI的真正力量在于协作。JARVIS为我们展示了一种可能性——通过智能调度，让不同的AI模型像团队一样高效配合。希望这篇教程能帮助你开启多模型协作的探索之旅，构建出更强大、更有价值的AI应用。
现在，是时候动手实践了！选择一个你感兴趣的场景，用JARVIS搭建你的第一个多模型协作应用吧。祝开发顺利！
评论区