Appearance
第18天:LLM模块总结与项目
学习目标
- 回顾本周所学知识
- 掌握技术选型方法
- 完成智能问答系统项目
- 总结学习成果
课程内容
1. 知识点回顾
1.1 LLM原理与架构
核心概念:
- Transformer架构
- Self-Attention机制
- 位置编码
- 多头注意力
关键代码:
python
class SelfAttention(nn.Module):
def __init__(self, d_model, d_k, d_v):
super().__init__()
self.W_q = nn.Linear(d_model, d_k, bias=False)
self.W_k = nn.Linear(d_model, d_k, bias=False)
self.W_v = nn.Linear(d_model, d_v, bias=False)
self.W_o = nn.Linear(d_v, d_model, bias=False)
def forward(self, x, mask=None):
Q = self.W_q(x)
K = self.W_k(x)
V = self.W_v(x)
scores = torch.matmul(Q, K.transpose(-2, -1))
scores = scores / torch.sqrt(torch.tensor(self.d_k, dtype=torch.float32))
if mask is not None:
scores = scores.masked_fill(mask == 0, -1e9)
weights = F.softmax(scores, dim=-1)
output = torch.matmul(weights, V)
output = self.W_o(output)
return output, weights1.2 主流LLM架构
架构对比:
| 模型 | 架构 | 特点 | 适用场景 |
|---|---|---|---|
| GPT | Decoder-only | 生成能力强 | 文本生成、代码生成 |
| BERT | Encoder-only | 理解能力强 | NLU任务、分类 |
| T5 | Encoder-Decoder | 统一框架 | 多任务处理 |
| LLaMA | Decoder-only | 开源友好 | 本地部署 |
| Claude | Decoder-only | 上下文长 | 长文档处理 |
| Gemini | Multi-modal | 多模态 | 多模态任务 |
1.3 国内大模型
国内模型对比:
| 模型 | 开发者 | 特点 | API价格 |
|---|---|---|---|
| 文心一言 | 百度 | 知识增强 | 中 |
| 通义千问 | 阿里 | 多尺寸 | 低 |
| GLM | 智谱AI | 开源 | 低 |
| Kimi | 月之暗面 | 超长上下文 | 中 |
| DeepSeek | 深度求索 | MoE架构 | 低 |
| Yi | 零一万物 | 开源 | 低 |
1.4 LLM API开发
核心功能:
- API封装
- 流式输出
- Function Calling
- 多模型支持
代码示例:
python
class LLMClient:
def __init__(self, api_key, base_url, model):
self.api_key = api_key
self.base_url = base_url
self.model = model
def call(self, prompt, **kwargs):
url = f"{self.base_url}/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
}
payload = {
"model": self.model,
"messages": [
{"role": "user", "content": prompt}
],
**kwargs
}
response = requests.post(url, headers=headers, data=json.dumps(payload))
response.raise_for_status()
return response.json()
def stream(self, prompt, **kwargs):
payload = {
"model": self.model,
"messages": [
{"role": "user", "content": prompt}
],
"stream": True,
**kwargs
}
response = requests.post(url, headers=headers, data=json.dumps(payload), stream=True)
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
break
try:
chunk = json.loads(data)
if 'choices' in chunk and len(chunk['choices']) > 0:
delta = chunk['choices'][0].get('delta', {})
content = delta.get('content', '')
if content:
yield content
except json.JSONDecodeError:
pass1.5 Prompt Engineering
核心技巧:
- 清晰明确的指令
- 提供充足的上下文
- 指定输出格式
- 使用分隔符
- Few-shot Learning
- Chain of Thought
示例:
python
prompt = """
你是一个专业的Python编程助手。
请编写一个快速排序算法,要求:
1. 使用递归实现
2. 添加详细的中文注释
3. 包含时间复杂度和空间复杂度分析
4. 提供测试代码
输出格式:
```python
# 快速排序实现
def quick_sort(arr):
# 你的代码
pass
# 测试代码
if __name__ == "__main__":
# 测试
pass时间复杂度:O(n log n) 空间复杂度:O(log n) """
#### 1.6 LLM评估
**评估指标**:
- 准确性:准确率、BLEU、ROUGE
- 效率:响应时间、吞吐量
- 成本:API费用、部署成本
**评估流程**:
1. 定义评估目标
2. 选择评估指标
3. 准备测试数据
4. 执行评估
5. 分析结果
6. 做出选择
### 2. 技术选型
#### 2.1 选型原则
**原则1:需求导向**
- 明确应用场景
- 确定性能要求
- 评估成本预算
**原则2:技术匹配**
- 选择合适的架构
- 匹配功能需求
- 考虑扩展性
**原则3:成本优化**
- 对比API价格
- 评估部署成本
- 考虑维护成本
#### 2.2 选型决策树开始 │ ├─ 需要本地部署? │ ├─ 是 → 选择开源模型(LLaMA、GLM、Yi) │ └─ 否 → 继续判断 │ ├─ 需要处理长文档? │ ├─ 是 → 选择Claude、Kimi │ └─ 否 → 继续判断 │ ├─ 需要多模态? │ ├─ 是 → 选择GPT-4、Gemini │ └─ 否 → 继续判断 │ ├─ 成本敏感? │ ├─ 是 → 选择GPT-3.5、DeepSeek │ └─ 否 → 选择GPT-4、Claude-3 │ └─ 需要中文优化? ├─ 是 → 选择国内模型 └─ 否 → 选择国外模型
#### 2.3 选型案例
**案例1:智能客服系统**
- 需求:问答准确、响应快速、成本可控
- 推荐:GPT-3.5-turbo + RAG
- 理由:性价比高、响应快、准确率足够
**案例2:代码生成工具**
- 需求:代码质量高、支持多种语言
- 推荐:Claude-3-Opus 或 GPT-4
- 理由:代码生成能力强、注释详细
**案例3:长文档分析**
- 需求:处理长文档、上下文长
- 推荐:Claude-3-Opus(200K token)
- 理由:上下文长、理解能力强
### 3. 实战项目:智能问答系统
#### 3.1 项目概述
**项目目标**:
- 构建一个智能问答系统
- 支持多种问答类型
- 提供友好的用户界面
- 优化响应速度和成本
**技术栈**:
- 后端:Python + FastAPI
- LLM:GPT-3.5-turbo
- 前端:HTML + JavaScript
- 部署:Docker
#### 3.2 系统架构用户界面 ↓ API网关 ↓ 问答引擎 ├─ 通用问答 ├─ 代码问答 ├─ 文档问答 └─ 知识库问答 ↓ LLM API
#### 3.3 后端实现
**3.3.1 项目结构**qa_system/ ├── app/ │ ├── init.py │ ├── main.py │ ├── config.py │ ├── models/ │ │ ├── init.py │ │ ├── question.py │ │ └── answer.py │ ├── services/ │ │ ├── init.py │ │ ├── llm_service.py │ │ ├── qa_service.py │ │ └── cache_service.py │ ├── prompts/ │ │ ├── init.py │ │ └── qa_prompts.py │ └── utils/ │ ├── init.py │ └── logger.py ├── requirements.txt ├── Dockerfile └── README.md
**3.3.2 配置文件**
```python
# config.py
import os
from pydantic import BaseSettings
class Settings(BaseSettings):
# API配置
api_host: str = "0.0.0.0"
api_port: int = 8000
# LLM配置
openai_api_key: str = os.getenv("OPENAI_API_KEY", "")
openai_model: str = "gpt-3.5-turbo"
openai_temperature: float = 0.7
openai_max_tokens: int = 1000
# 缓存配置
cache_ttl: int = 3600 # 1小时
# 日志配置
log_level: str = "INFO"
log_file: str = "qa_system.log"
settings = Settings()3.3.3 LLM服务
python
# services/llm_service.py
from openai import OpenAI
from typing import Iterator
import logging
logger = logging.getLogger(__name__)
class LLMService:
def __init__(self, api_key: str, model: str):
self.client = OpenAI(api_key=api_key)
self.model = model
def call(self, messages: list, **kwargs) -> dict:
"""
调用LLM
Args:
messages: 消息列表
**kwargs: 其他参数
Returns:
response: 响应
"""
try:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
**kwargs
)
return response.model_dump()
except Exception as e:
logger.error(f"LLM调用失败: {e}")
raise
def stream(self, messages: list, **kwargs) -> Iterator[str]:
"""
流式输出
Args:
messages: 消息列表
**kwargs: 其他参数
Yields:
chunk: 数据块
"""
try:
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
stream=True,
**kwargs
)
for chunk in stream:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
except Exception as e:
logger.error(f"流式输出失败: {e}")
raise
def get_text(self, prompt: str, **kwargs) -> str:
"""
获取文本
Args:
prompt: 输入提示
**kwargs: 其他参数
Returns:
text: 输出文本
"""
messages = [{"role": "user", "content": prompt}]
response = self.call(messages, **kwargs)
return response['choices'][0]['message']['content']3.3.4 Prompt模板
python
# prompts/qa_prompts.py
GENERAL_QA_PROMPT = """
你是一个专业的问答助手。请根据用户的问题提供准确、详细的回答。
问题:{question}
回答:
"""
CODE_QA_PROMPT = """
你是一个专业的编程问答助手。请根据用户的问题提供准确的代码解决方案。
问题:{question}
要求:
1. 提供完整的代码
2. 添加详细注释
3. 包含测试代码
回答:
"""
DOC_QA_PROMPT = """
你是一个专业的文档问答助手。请根据提供的文档内容回答用户的问题。
文档:
{document}
问题:{question}
回答:
"""
KNOWLEDGE_QA_PROMPT = """
你是一个专业的知识库问答助手。请根据提供的知识库内容回答用户的问题。
知识库:
{knowledge}
问题:{question}
回答:
"""3.3.5 问答服务
python
# services/qa_service.py
from typing import Optional, Iterator
from .llm_service import LLMService
from .cache_service import CacheService
from .qa_prompts import (
GENERAL_QA_PROMPT,
CODE_QA_PROMPT,
DOC_QA_PROMPT,
KNOWLEDGE_QA_PROMPT
)
import logging
logger = logging.getLogger(__name__)
class QAService:
def __init__(
self,
llm_service: LLMService,
cache_service: CacheService,
temperature: float = 0.7,
max_tokens: int = 1000
):
self.llm_service = llm_service
self.cache_service = cache_service
self.temperature = temperature
self.max_tokens = max_tokens
def answer(
self,
question: str,
qa_type: str = "general",
document: Optional[str] = None,
knowledge: Optional[str] = None,
use_cache: bool = True
) -> str:
"""
回答问题
Args:
question: 问题
qa_type: 问答类型(general、code、doc、knowledge)
document: 文档内容
knowledge: 知识库内容
use_cache: 是否使用缓存
Returns:
answer: 答案
"""
# 检查缓存
if use_cache:
cached_answer = self.cache_service.get(question)
if cached_answer:
logger.info(f"从缓存获取答案: {question}")
return cached_answer
# 选择Prompt
if qa_type == "general":
prompt = GENERAL_QA_PROMPT.format(question=question)
elif qa_type == "code":
prompt = CODE_QA_PROMPT.format(question=question)
elif qa_type == "doc":
if not document:
raise ValueError("文档问答需要提供document参数")
prompt = DOC_QA_PROMPT.format(document=document, question=question)
elif qa_type == "knowledge":
if not knowledge:
raise ValueError("知识库问答需要提供knowledge参数")
prompt = KNOWLEDGE_QA_PROMPT.format(knowledge=knowledge, question=question)
else:
raise ValueError(f"未知的问答类型: {qa_type}")
# 调用LLM
answer = self.llm_service.get_text(
prompt,
temperature=self.temperature,
max_tokens=self.max_tokens
)
# 缓存答案
if use_cache:
self.cache_service.set(question, answer)
return answer
def stream_answer(
self,
question: str,
qa_type: str = "general",
document: Optional[str] = None,
knowledge: Optional[str] = None
) -> Iterator[str]:
"""
流式回答
Args:
question: 问题
qa_type: 问答类型
document: 文档内容
knowledge: 知识库内容
Yields:
chunk: 数据块
"""
# 选择Prompt
if qa_type == "general":
prompt = GENERAL_QA_PROMPT.format(question=question)
elif qa_type == "code":
prompt = CODE_QA_PROMPT.format(question=question)
elif qa_type == "doc":
if not document:
raise ValueError("文档问答需要提供document参数")
prompt = DOC_QA_PROMPT.format(document=document, question=question)
elif qa_type == "knowledge":
if not knowledge:
raise ValueError("知识库问答需要提供knowledge参数")
prompt = KNOWLEDGE_QA_PROMPT.format(knowledge=knowledge, question=question)
else:
raise ValueError(f"未知的问答类型: {qa_type}")
# 流式输出
for chunk in self.llm_service.stream(
prompt,
temperature=self.temperature,
max_tokens=self.max_tokens
):
yield chunk3.3.6 缓存服务
python
# services/cache_service.py
from typing import Optional
import time
import hashlib
import logging
logger = logging.getLogger(__name__)
class CacheService:
def __init__(self, ttl: int = 3600):
self.cache = {}
self.ttl = ttl
def _generate_key(self, key: str) -> str:
"""
生成缓存键
Args:
key: 原始键
Returns:
cache_key: 缓存键
"""
return hashlib.md5(key.encode()).hexdigest()
def get(self, key: str) -> Optional[str]:
"""
获取缓存
Args:
key: 键
Returns:
value: 值
"""
cache_key = self._generate_key(key)
if cache_key in self.cache:
value, timestamp = self.cache[cache_key]
# 检查是否过期
if time.time() - timestamp < self.ttl:
logger.info(f"缓存命中: {cache_key}")
return value
else:
# 删除过期缓存
del self.cache[cache_key]
return None
def set(self, key: str, value: str):
"""
设置缓存
Args:
key: 键
value: 值
"""
cache_key = self._generate_key(key)
self.cache[cache_key] = (value, time.time())
logger.info(f"缓存设置: {cache_key}")
def clear(self):
"""清空缓存"""
self.cache.clear()
logger.info("缓存已清空")3.3.7 API接口
python
# main.py
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional
import uvicorn
import logging
from config import settings
from services.llm_service import LLMService
from services.qa_service import QAService
from services.cache_service import CacheService
from utils.logger import setup_logger
# 设置日志
setup_logger(settings.log_level, settings.log_file)
logger = logging.getLogger(__name__)
# 创建应用
app = FastAPI(title="智能问答系统")
# 配置CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 初始化服务
llm_service = LLMService(
api_key=settings.openai_api_key,
model=settings.openai_model
)
cache_service = CacheService(ttl=settings.cache_ttl)
qa_service = QAService(
llm_service=llm_service,
cache_service=cache_service,
temperature=settings.openai_temperature,
max_tokens=settings.openai_max_tokens
)
# 数据模型
class QuestionRequest(BaseModel):
question: str
qa_type: str = "general"
document: Optional[str] = None
knowledge: Optional[str] = None
use_cache: bool = True
class AnswerResponse(BaseModel):
answer: str
qa_type: str
# API接口
@app.post("/api/answer", response_model=AnswerResponse)
async def answer(request: QuestionRequest):
"""
回答问题
"""
try:
answer = qa_service.answer(
question=request.question,
qa_type=request.qa_type,
document=request.document,
knowledge=request.knowledge,
use_cache=request.use_cache
)
return AnswerResponse(
answer=answer,
qa_type=request.qa_type
)
except Exception as e:
logger.error(f"回答失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/answer/stream")
async def stream_answer(
question: str,
qa_type: str = "general",
document: Optional[str] = None,
knowledge: Optional[str] = None
):
"""
流式回答
"""
try:
from fastapi.responses import StreamingResponse
async def generate():
for chunk in qa_service.stream_answer(
question=question,
qa_type=qa_type,
document=document,
knowledge=knowledge
):
yield chunk
return StreamingResponse(generate(), media_type="text/plain")
except Exception as e:
logger.error(f"流式回答失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/health")
async def health():
"""
健康检查
"""
return {"status": "ok"}
if __name__ == "__main__":
uvicorn.run(
"main:app",
host=settings.api_host,
port=settings.api_port,
reload=True
)3.3.8 依赖文件
txt
# requirements.txt
fastapi==0.104.1
uvicorn==0.24.0
pydantic==2.5.0
pydantic-settings==2.1.0
openai==1.3.0
python-multipart==0.0.63.3.9 Dockerfile
dockerfile
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]3.4 前端实现
html
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>智能问答系统</title>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: Arial, sans-serif;
background-color: #f5f5f5;
padding: 20px;
}
.container {
max-width: 800px;
margin: 0 auto;
background-color: white;
padding: 20px;
border-radius: 8px;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
h1 {
text-align: center;
color: #333;
margin-bottom: 20px;
}
.form-group {
margin-bottom: 15px;
}
label {
display: block;
margin-bottom: 5px;
color: #555;
}
select,
textarea,
input[type="text"] {
width: 100%;
padding: 10px;
border: 1px solid #ddd;
border-radius: 4px;
font-size: 14px;
}
textarea {
min-height: 100px;
resize: vertical;
}
button {
background-color: #007bff;
color: white;
border: none;
padding: 10px 20px;
border-radius: 4px;
cursor: pointer;
font-size: 14px;
}
button:hover {
background-color: #0056b3;
}
.answer {
margin-top: 20px;
padding: 15px;
background-color: #f9f9f9;
border-radius: 4px;
border-left: 4px solid #007bff;
}
.answer h3 {
color: #333;
margin-bottom: 10px;
}
.answer p {
color: #555;
line-height: 1.6;
}
.loading {
display: none;
text-align: center;
padding: 20px;
}
.loading.show {
display: block;
}
</style>
</head>
<body>
<div class="container">
<h1>智能问答系统</h1>
<div class="form-group">
<label for="question">问题:</label>
<textarea id="question" placeholder="请输入您的问题..."></textarea>
</div>
<div class="form-group">
<label for="qa-type">问答类型:</label>
<select id="qa-type">
<option value="general">通用问答</option>
<option value="code">代码问答</option>
<option value="doc">文档问答</option>
<option value="knowledge">知识库问答</option>
</select>
</div>
<div class="form-group" id="document-group" style="display: none;">
<label for="document">文档内容:</label>
<textarea id="document" placeholder="请输入文档内容..."></textarea>
</div>
<div class="form-group" id="knowledge-group" style="display: none;">
<label for="knowledge">知识库内容:</label>
<textarea id="knowledge" placeholder="请输入知识库内容..."></textarea>
</div>
<div class="form-group">
<button id="submit-btn">提交问题</button>
</div>
<div class="loading" id="loading">
<p>正在思考...</p>
</div>
<div class="answer" id="answer" style="display: none;">
<h3>答案:</h3>
<p id="answer-content"></p>
</div>
</div>
<script>
// 显示/隐藏输入框
document.getElementById('qa-type').addEventListener('change', function() {
const qaType = this.value;
const documentGroup = document.getElementById('document-group');
const knowledgeGroup = document.getElementById('knowledge-group');
if (qaType === 'doc') {
documentGroup.style.display = 'block';
knowledgeGroup.style.display = 'none';
} else if (qaType === 'knowledge') {
documentGroup.style.display = 'none';
knowledgeGroup.style.display = 'block';
} else {
documentGroup.style.display = 'none';
knowledgeGroup.style.display = 'none';
}
});
// 提交问题
document.getElementById('submit-btn').addEventListener('click', async function() {
const question = document.getElementById('question').value;
const qaType = document.getElementById('qa-type').value;
const document = document.getElementById('document').value;
const knowledge = document.getElementById('knowledge').value;
if (!question) {
alert('请输入问题');
return;
}
// 显示加载
document.getElementById('loading').classList.add('show');
document.getElementById('answer').style.display = 'none';
try {
const response = await fetch('/api/answer', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({
question: question,
qa_type: qaType,
document: document || null,
knowledge: knowledge || null
})
});
const data = await response.json();
// 显示答案
document.getElementById('answer-content').textContent = data.answer;
document.getElementById('answer').style.display = 'block';
} catch (error) {
alert('请求失败: ' + error.message);
} finally {
// 隐藏加载
document.getElementById('loading').classList.remove('show');
}
});
</script>
</body>
</html>3.5 部署
3.5.1 构建Docker镜像
bash
docker build -t qa-system .3.5.2 运行容器
bash
docker run -d -p 8000:8000 --env OPENAI_API_KEY=your-api-key qa-system3.5.3 访问系统
打开浏览器访问:http://localhost:8000(假设您使用8000端口)
4. 学习总结
4.1 本周学习成果
掌握的知识:
- LLM原理与架构
- 主流LLM架构对比
- 国内大模型详解
- 国外大模型详解
- LLM API开发
- Prompt Engineering
- LLM评估与选择
完成的任务:
- 从零实现Self-Attention
- 对比不同模型输出
- 开发LLM API封装库
- 优化Prompt提升效果
- 评估模型性能
- 完成智能问答系统项目
4.2 学习建议
持续学习:
- 关注LLM最新进展
- 实践更多项目
- 参与开源社区
- 分享学习心得
进阶方向:
- RAG技术
- AI Agent开发
- 模型微调
- 多模态应用
课后作业
作业1:扩展问答系统
题目:为问答系统添加新功能
要求:
- 添加历史记录功能
- 添加导出功能
- 添加评分功能
- 优化用户界面
作业2:性能优化
题目:优化问答系统性能
要求:
- 优化缓存策略
- 实现批量处理
- 添加负载均衡
- 优化数据库查询
作业3:部署优化
题目:优化问答系统部署
要求:
- 使用Kubernetes部署
- 添加监控和日志
- 实现自动扩缩容
- 添加备份和恢复
参考资料
项目资源
FastAPI: https://fastapi.tiangolo.com/
- FastAPI文档
OpenAI API: https://platform.openai.com/docs
- OpenAI API文档
扩展阅读
- RAG技术: 检索增强生成
- LangChain: AI应用开发框架
- LlamaIndex: 数据框架
下节预告
下一节我们将开始模块2:MCP协议开发,学习MCP协议深度解析、MCP Server开发、MCP工具开发等内容。

扫描二维码关注"架构师AI杜"公众号,获取更多技术内容和最新动态
