Appearance
第18天:LLM模块总结与项目
学习目标
本节的学习目标包括回顾本周所学知识、掌握技术选型方法、完成智能问答系统项目、总结学习成果等方面。
课程内容
1. 知识点回顾
1.1 LLM原理与架构
核心概念:
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开发
核心功能:
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
核心技巧:
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费用、部署成本)等方面。
评估流程:
- 定义评估目标
- 选择评估指标
- 准备测试数据
- 执行评估
- 分析结果
- 做出选择
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 API3.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.md3.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杜"公众号,获取更多技术内容和最新动态
