1011 字
3 分钟
构建你的第一个AI应用：架构与工程实践
学会了AI的基础知识，接下来是动手实践：构建一个真正的AI应用。
这篇文章不讲理论，讲实操：从架构设计到技术选型，从开发到上线，用一个完整的智能客服案例带你走完全流程。
本文要点
- 四种AI应用架构模式
- 技术栈选择指南
- 工程化实践要点
- 完整案例：智能客服从0到1
- 评估与监控体系
- 安全与合规
一、AI应用架构模式
1.1 模式1：LLM as Generator（生成器）
flowchart TD
N0["输入"]
N1["LLM"]
N0 --> N1
N1["LLM"]
N2["输出"]
N1 --> N2
1.2 模式2：LLM as Orchestrator（协调器）
flowchart TD
N0["用户请求"]
N1["LLM协调多个工具"]
N0 --> N1
N1["LLM协调多个工具"]
N2["完成任务"]
N1 --> N2
1.3 模式3：LLM as Interface（接口层）
flowchart TD
N0["用户自然语言"]
N1["LLM解析意图"]
N0 --> N1
N1["LLM解析意图"]
N2["调用后端服务"]
N1 --> N2
N2["调用后端服务"]
N3["返回结果"]
N2 --> N3
1.4 模式4：RAG增强模式
flowchart TD
N0["用户问题"]
N1["检索知识库"]
N0 --> N1
N1["检索知识库"]
N2["LLM+知识生成"]
N1 --> N2
N2["LLM+知识生成"]
N3["回答"]
N2 --> N3
二、技术栈选择
2.1 应用框架对比
| 框架 | 语言 | 特点 | 适用场景 |
|---|---|---|---|
| LangChain | Python/JS | 生态丰富、组件多 | 快速原型、通用应用 |
| LlamaIndex | Python | RAG专精、文档处理好 | 知识库应用 |
| Semantic Kernel | C#/Python | 微软出品、企业级 | 企业集成、.NET项目 |
| Haystack | Python | 搜索增强、模块化 | 搜索类应用 |
| FastAPI + OpenAI SDK | Python | 灵活、轻量 | 自定义需求强 |
选择建议：
- 快速上手：LangChain
- 知识库应用：LlamaIndex
- 企业项目：Semantic Kernel
- 追求灵活：直接用SDK
2.2 向量数据库对比
| 数据库 | 部署方式 | 特点 | 适用场景 |
|---|---|---|---|
| Pinecone | 云托管 | 简单、可扩展 | 生产环境 |
| Chroma | 本地/Docker | 轻量、开源 | 开发测试 |
| Milvus | 自部署 | 高性能、分布式 | 大规模生产 |
| Qdrant | 自部署/云 | 高效、开源 | 性能敏感 |
| Weaviate | 自部署/云 | 语义强 | 复杂查询 |
2.3 模型选择
| 任务类型 | 推荐模型 | 理由 |
|---|---|---|
| 通用对话 | GPT-4o / Claude 3.5 Sonnet | 综合能力强 |
| 代码生成 | Claude 3.5 Sonnet / DeepSeek V3 | 代码能力突出 |
| 知识问答 | GPT-4o-mini / DeepSeek V3 | 性价比高 |
| 推理任务 | o3-mini / DeepSeek R1 | 推理能力强 |
| 私有部署 | Qwen 2.5 / LLaMA 3.1 | 开源可控 |
三、工程化实践
3.1 Prompt版本管理
mindmap
root((不要把Prompt硬编码在代码里！))
推荐结构：
prompts/
v1/
system_prompt.txt
user_prompt_template.txt
v2/
system_prompt.txt
user_prompt_template.txt
current -> v2 # 软链接指向当前版本
好处：
可追溯：知道什么时候改了什么
可回滚：出问题切回旧版本
可A/B测试：不同版本对比效果
from pathlib import Path
class PromptManager: def __init__(self, version="current"): self.base_path = Path("prompts") / version
def get_system_prompt(self) -> str: return (self.base_path / "system_prompt.txt").read_text()
def get_user_prompt(self, **kwargs) -> str: template = (self.base_path / "user_prompt_template.txt").read_text() return template.format(**kwargs)3.2 评估与监控体系
┌─────────────────────────────────────────────────────────────┐│ AI应用评估体系 │├─────────────────────────────────────────────────────────────┤│ ││ 功能评估： ││ ├── 准确率：答案正确的比例 ││ ├── 相关性：答案与问题的相关程度 ││ ├── 完整性：是否完整回答了问题 ││ └── 格式正确率：输出格式是否符合预期 ││ ││ 性能评估： ││ ├── 延迟（P50/P95/P99） ││ ├── 吞吐量（QPS） ││ └── Token消耗 ││ ││ 用户体验： ││ ├── 满意度评分 ││ ├── 反馈率（点赞/点踩） ││ └── 对话轮数 ││ ││ 成本监控： ││ ├── API调用费用 ││ ├── 基础设施成本 ││ └── 成本/查询 ││ │└─────────────────────────────────────────────────────────────┘# 监控示例import timefrom dataclasses import dataclass
@dataclassclass QueryMetrics: query_id: str latency_ms: float input_tokens: int output_tokens: int model: str success: bool
def track_query(func): async def wrapper(*args, **kwargs): start = time.time() try: result = await func(*args, **kwargs) success = True except Exception as e: success = False raise finally: latency = (time.time() - start) * 1000 metrics = QueryMetrics( query_id=kwargs.get("query_id"), latency_ms=latency, input_tokens=result.usage.input_tokens, output_tokens=result.usage.output_tokens, model=result.model, success=success ) # 发送到监控系统 send_metrics(metrics) return result return wrapper3.3 成本优化策略
flowchart TD
N0["- 简单问题"]
N1["小模型（GPT-4o-mini）"]
N0 --> N1
N2["- 复杂问题"]
N3["大模型（GPT-4o）"]
N2 --> N3
3.4 安全措施
mindmap
root((输入安全：))
敏感信息检测（PII、密钥）
恶意Prompt注入防护
内容审核（违规内容）
长度限制
输出安全：
有害内容过滤
幻觉检测
格式验证
来源标注
访问控制：
用户认证
权限管理
调用限流
审计日志
四、完整案例：智能客服从0到1
4.1 需求分析
背景：某电商公司需要智能客服系统，处理用户咨询
核心需求：1. 回答产品相关问题2. 处理订单查询3. 处理退换货申请4. 无法解决时转人工
约束条件：- 响应时间 < 3秒- 准确率 > 85%- 7x24小时服务- 成本可控4.2 架构设计
┌─────────────────────────────────────────────────────────────┐│ 智能客服系统架构 │├─────────────────────────────────────────────────────────────┤│ ││ ┌──────────┐ ││ │ 用户 │ ││ └────┬─────┘ ││ ↓ ││ ┌────────────────────────────────────────────┐ ││ │ API网关 │ ││ │ 认证 / 限流 / 路由 │ ││ └────────────────────┬───────────────────────┘ ││ ↓ ││ ┌────────────────────────────────────────────┐ ││ │ 意图分类器 │ ││ │ 产品咨询 / 订单查询 / 退换货 / 其他 │ ││ └────────────────────┬───────────────────────┘ ││ ↓ ││ ┌────────────────────────────────────────────┐ ││ │ 路由层 │ ││ ├────────────────────────────────────────────┤ ││ │ 产品咨询 → RAG知识库 │ ││ │ 订单查询 → 订单系统API │ ││ │ 退换货 → 退换货流程 │ ││ │ 其他 → 通用对话 / 转人工 │ ││ └────────────────────┬───────────────────────┘ ││ ↓ ││ ┌────────────────────────────────────────────┐ ││ │ LLM服务层 │ ││ │ GPT-4o-mini / DeepSeek V3 │ ││ └────────────────────┬───────────────────────┘ ││ ↓ ││ ┌────────────────────────────────────────────┐ ││ │ 响应生成 │ ││ │ 格式化 / 安全检查 / 缓存 │ ││ └────────────────────┬───────────────────────┘ ││ ↓ ││ ┌──────────┐ ││ │ 用户 │ ││ └──────────┘ ││ ││ ┌────────────────────────────────────────────┐ ││ │ 支撑服务 │ ││ ├────────────────────────────────────────────┤ ││ │ 向量数据库（产品知识库） │ ││ │ Redis（缓存/会话） │ ││ │ MySQL（日志/分析） │ ││ │ 监控系统（Prometheus + Grafana） │ ││ └────────────────────────────────────────────┘ ││ │└─────────────────────────────────────────────────────────────┘4.3 核心代码实现
项目结构：
mindmap
root((smart-customer-service/))
app/
__init__.py
main.py # FastAPI入口
routers/
chat.py # 对话接口
health.py # 健康检查
services/
intent_classifier.py # 意图分类
rag_service.py # RAG服务
order_service.py # 订单服务
llm_service.py # LLM封装
models/
request.py # 请求模型
response.py # 响应模型
utils/
cache.py # 缓存工具
metrics.py # 监控工具
prompts/
v1/
system_prompt.txt
intent_prompt.txt
tests/
requirements.txt
Dockerfile
意图分类器：
from enum import Enumfrom openai import OpenAI
class Intent(Enum): PRODUCT_INQUIRY = "product_inquiry" # 产品咨询 ORDER_QUERY = "order_query" # 订单查询 RETURN_REQUEST = "return_request" # 退换货 GENERAL = "general" # 通用/其他
class IntentClassifier: def __init__(self): self.client = OpenAI() self.prompt = """分析用户意图，返回以下之一：- product_inquiry: 产品相关问题- order_query: 订单查询- return_request: 退换货申请- general: 其他
用户消息：{message}意图："""
def classify(self, message: str) -> Intent: response = self.client.chat.completions.create( model="gpt-4o-mini", messages=[{ "role": "user", "content": self.prompt.format(message=message) }], temperature=0, max_tokens=20 ) intent_str = response.choices[0].message.content.strip() return Intent(intent_str)RAG服务：
from langchain_openai import OpenAIEmbeddings, ChatOpenAIfrom langchain_community.vectorstores import Chromafrom langchain.chains import RetrievalQA
class RAGService: def __init__(self): self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small") self.vectorstore = Chroma( persist_directory="./chroma_db", embedding_function=self.embeddings ) self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) self.qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", retriever=self.vectorstore.as_retriever(search_kwargs={"k": 3}), return_source_documents=True )
def query(self, question: str) -> dict: result = self.qa_chain({"query": question}) return { "answer": result["result"], "sources": [doc.metadata for doc in result["source_documents"]] }主服务：
from fastapi import FastAPI, Dependsfrom app.routers import chat, healthfrom app.services.intent_classifier import IntentClassifier, Intentfrom app.services.rag_service import RAGServicefrom app.services.order_service import OrderServicefrom app.utils.cache import Cache
app = FastAPI(title="Smart Customer Service")
# 初始化服务classifier = IntentClassifier()rag_service = RAGService()order_service = OrderService()cache = Cache()
@app.post("/chat")async def chat(message: str, user_id: str): # 1. 检查缓存 cached = cache.get(f"chat:{user_id}:{message}") if cached: return {"response": cached, "cached": True}
# 2. 意图分类 intent = classifier.classify(message)
# 3. 路由处理 if intent == Intent.PRODUCT_INQUIRY: result = rag_service.query(message) response = result["answer"] elif intent == Intent.ORDER_QUERY: response = order_service.handle_query(message, user_id) elif intent == Intent.RETURN_REQUEST: response = order_service.handle_return(message, user_id) else: # 通用处理或转人工 response = "抱歉，我暂时无法回答这个问题。正在为您转接人工客服..."
# 4. 缓存结果 cache.set(f"chat:{user_id}:{message}", response, ttl=3600)
return { "response": response, "intent": intent.value, "cached": False }4.4 测试与评估
测试用例：
import pytestfrom app.services.intent_classifier import IntentClassifier, Intent
class TestIntentClassifier: def test_product_inquiry(self): classifier = IntentClassifier()
assert classifier.classify("这个手机多少钱") == Intent.PRODUCT_INQUIRY assert classifier.classify("这款衣服有L码吗") == Intent.PRODUCT_INQUIRY assert classifier.classify("产品有什么颜色") == Intent.PRODUCT_INQUIRY
def test_order_query(self): classifier = IntentClassifier()
assert classifier.classify("我的订单到哪了") == Intent.ORDER_QUERY assert classifier.classify("查一下订单123456") == Intent.ORDER_QUERY
def test_return_request(self): classifier = IntentClassifier()
assert classifier.classify("我想退货") == Intent.RETURN_REQUEST assert classifier.classify("衣服不合适，怎么换货") == Intent.RETURN_REQUEST评估脚本：
import jsonfrom app.main import appfrom sklearn.metrics import accuracy_score, precision_recall_fscore_support
def evaluate_intent_classification(): """评估意图分类准确率""" test_data = json.load(open("tests/intent_test_data.json"))
y_true = [] y_pred = []
for item in test_data: pred = classifier.classify(item["message"]) y_true.append(item["expected_intent"]) y_pred.append(pred.value)
accuracy = accuracy_score(y_true, y_pred) precision, recall, f1, _ = precision_recall_fscore_support( y_true, y_pred, average='weighted' )
print(f"准确率: {accuracy:.2%}") print(f"精确率: {precision:.2%}") print(f"召回率: {recall:.2%}") print(f"F1分数: {f1:.2%}")
def evaluate_rag_quality(): """评估RAG回答质量""" test_data = json.load(open("tests/rag_test_data.json"))
correct = 0 for item in test_data: result = rag_service.query(item["question"]) # 人工或自动评估 if is_answer_correct(result["answer"], item["expected_answer"]): correct += 1
print(f"RAG准确率: {correct/len(test_data):.2%}")4.5 部署与运维
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", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]docker-compose.yml：
version: '3.8'services: api: build: . ports: - "8000:8000" environment: - OPENAI_API_KEY=${OPENAI_API_KEY} depends_on: - redis - chroma
redis: image: redis:alpine ports: - "6379:6379"
chroma: image: chromadb/chroma ports: - "8001:8000" volumes: - chroma_data:/chroma/data
volumes: chroma_data:五、评估与监控
5.1 评估指标体系
┌─────────────────────────────────────────────────────────────┐│ 评估指标体系 │├─────────────────────────────────────────────────────────────┤│ ││ 在线指标（实时监控）： ││ ├── 响应延迟：P50 < 1s, P95 < 3s ││ ├── 成功率：> 99% ││ ├── QPS：峰值处理能力 ││ └── 错误率：< 1% ││ ││ 离线指标（定期评估）： ││ ├── 意图分类准确率：> 90% ││ ├── 回答准确率：> 85% ││ ├── 用户满意度：> 4.0/5.0 ││ └── 转人工率：< 15% ││ ││ 业务指标： ││ ├── 问题解决率 ││ ├── 平均对话轮数 ││ ├── 用户留存率 ││ └── 成本/查询 ││ │└─────────────────────────────────────────────────────────────┘5.2 监控告警
groups: - name: ai-service rules: - alert: HighLatency expr: histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m])) > 3 for: 5m annotations: summary: "响应延迟过高"
- alert: HighErrorRate expr: rate(http_requests_total{status="500"}[5m]) / rate(http_requests_total[5m]) > 0.01 for: 5m annotations: summary: "错误率过高"
- alert: HighCost expr: rate(llm_token_cost_total[1h]) > 100 for: 1h annotations: summary: "LLM调用成本过高"六、安全与合规
6.1 安全措施
mindmap
root((1. 数据安全))
敏感信息脱敏（手机号、地址）
数据传输加密（HTTPS）
日志脱敏
数据保留策略
2. 应用安全
认证鉴权
限流防刷
输入验证
SQL/命令注入防护
3. AI安全
Prompt注入防护
输出内容审核
幻觉检测
来源追溯
6.2 合规考量
mindmap
root((1. 数据合规))
个人信息保护法（PIPL）
数据本地化存储
用户授权同意
2. 内容合规
违禁词过滤
敏感话题识别
内容审核机制
3. 行业合规
金融行业：数据隔离
医疗行业：专业审核
电商行业：消费者保护
常见问题 FAQ
Q1: 如何选择框架？
A:
- 快速原型：LangChain
- 知识库应用：LlamaIndex
- 企业集成：Semantic Kernel
- 追求灵活：直接用SDK
Q2: 如何评估AI应用效果？
A:
- 建立评估基准（测试集）
- 多维度指标（准确率、满意度、成本）
- 定期评估迭代
- A/B测试不同方案
Q3: 如何控制成本？
A:
- 选择合适模型（小模型优先）
- 使用缓存策略
- 监控Token消耗
- 批量处理非实时任务
Q4: 有哪些安全风险？
A:
- 数据泄露：敏感信息保护
- Prompt注入：输入过滤
- 幻觉输出：内容审核
- 恶意使用：权限控制
Q5: 如何处理模型升级？
A:
- 保持API版本兼容
- 新模型先灰度测试
- 监控效果变化
- 准备回滚方案
小结
构建AI应用的关键：
- 选择合适的架构模式：Generator/Orchestrator/Interface/RAG
- 技术栈匹配场景：框架、数据库、模型
- 工程化保障稳定性：Prompt管理、监控、测试
- 从简单开始，逐步迭代：先MVP，再优化
AI应用开发不是一次性工程，而是持续优化的过程。
下篇预告
通用模型不够用？需要定制专属能力？
参考资料
支持与分享
如果这篇文章对你有帮助，欢迎支持作者或分享给更多人
构建你的第一个AI应用：架构与工程实践
https://blog.souloss.com/posts/machine-learning/llm/build-ai-application/ 部分信息可能已经过时
相关文章 智能推荐