基于通义千问的智能点餐系统开发实践

1. 项目概述

1.1 项目背景与核心功能

这个基于通义千问(Qwen)的智能点餐系统是一个典型的AI应用落地案例，它通过大语言模型能力为餐厅场景提供智能化的交互服务。系统主要解决以下几个核心痛点：

1. 降低人力成本：传统餐厅需要大量服务员处理咨询、点餐等重复性工作
2. 提升服务效率：AI可以同时处理多个顾客请求，无等待时间
3. 个性化推荐：基于对话上下文理解顾客偏好，提供精准推荐
4. 24小时服务：不受营业时间限制，随时响应顾客需求

系统主要功能模块包括：

• 智能菜品推荐：根据用户描述推荐合适菜品（如"适合两个人的辣菜"）
• 餐厅信息服务：查询营业时间、地址、特色等基础信息
• 配送范围检查：自动判断用户地址是否在配送范围内
• 自然对话交互：支持多轮上下文对话，理解用户隐含需求

1.2 技术架构设计

整个系统采用分层架构设计，各组件职责明确：

code复制前端层(Web/App)
  ↑↓ HTTP/WebSocket
API服务层(FastAPI)
  ↑↓ 内部调用
AI能力层(LangChain + Qwen)
  ↑↓ 数据库访问
数据层(MySQL + Pinecone)

关键技术选型考量：

• LangChain框架：提供标准化的LLM应用开发范式，简化工具链集成
• 通义千问模型：在中文场景表现优异，特别适合餐饮领域的自然语言理解
• FastAPI后端：高性能异步框架，适合实时交互场景
• 混合数据库：MySQL存储结构化数据(菜单、订单)，Pinecone处理向量检索(菜品相似度)

2. 核心实现细节

2.1 环境配置与依赖管理

项目采用Python 3.10+环境，通过venv创建隔离环境：

bash复制python -m venv .venv  # 创建虚拟环境
.\.venv\Scripts\Activate.ps1  # 激活环境(Win)
source .venv/bin/activate  # 激活环境(Mac/Linux)

依赖管理使用requirements.txt固定版本：

python复制# requirements.txt
langchain==1.0.7  # LLM应用框架
dashscope>=1.14.0  # 通义千问SDK
pinecone-client~=7.3.0  # 向量数据库
mysql-connector-python~=9.4.0  # MySQL驱动

版本锁定策略：

• == 严格固定版本
• >= 允许自动升级小版本
• ~= 允许补丁版本升级(如7.3.0 → 7.3.1)

2.2 数据模型设计

MySQL表结构

sql复制CREATE TABLE menu_items (
    id VARCHAR(10) PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    description TEXT,
    price DECIMAL(10,2),
    category VARCHAR(20),
    spicy_level INT,
    is_recommended BOOLEAN
);

CREATE TABLE delivery_zones (
    id INT AUTO_INCREMENT PRIMARY KEY,
    zone_name VARCHAR(50),
    polygon_coordinates JSON  # 存储GeoJSON格式的多边形坐标
);

Pinecone索引配置

python复制import pinecone

pinecone.init(api_key="YOUR_KEY")
index = pinecone.Index("menu-items")

# 向量化配置
embedding_model = "text-embedding-3-small"  
dimension = 1536  # 与模型维度匹配

2.3 核心工具实现

菜品推荐工具

python复制from langchain.tools import tool
from typing import List
import json

@tool
def recommend_dishes(query: str, history: List[dict]) -> str:
    """
    根据用户描述推荐菜品，考虑历史对话上下文
    参数:
        query: 用户当前请求(如"推荐辣菜")
        history: 对话历史(用于理解上下文)
    返回:
        格式化推荐结果，包含MENU_IDS标记
    """
    # 1. 分析用户偏好
    context = _build_context(query, history)
    
    # 2. 向量搜索
    embedding = get_embedding(context)
    results = index.query(
        vector=embedding,
        top_k=5,
        include_metadata=True
    )
    
    # 3. LLM生成推荐理由
    recommendation = generate_recommendation(results)
    
    # 4. 返回结构化结果
    return f"{recommendation}\nMENU_IDS: {json.dumps([item.id for item in results])}"

关键实现细节：

1. 使用对话历史构建查询上下文
2. 通过向量搜索找到相似菜品
3. 让LLM生成自然语言推荐理由
4. 结构化返回结果供后续处理

配送检查工具

python复制@tool
def check_delivery(address: str) -> str:
    """
    检查地址是否在配送范围内
    参数:
        address: 用户提供的完整地址
    返回:
        配送检查结果和预计时间
    """
    # 1. 地址标准化
    normalized_addr = normalize_address(address)
    
    # 2. 地理编码
    location = amap.geocode(normalized_addr)
    if not location:
        return "地址解析失败，请确认地址是否正确"
    
    # 3. 多边形包含判断
    in_zone = check_polygon_contains(
        location['location'],
        get_delivery_zones()
    )
    
    # 4. 返回结果
    if in_zone:
        eta = calculate_eta(location)
        return f"地址在配送范围内，预计{eta}分钟送达"
    return "很抱歉，该地址不在当前配送范围内"

地理空间计算要点：

• 使用高德/百度地图API进行地理编码
• 射线法判断点是否在多边形内
• 配送时间基于距离和交通状况估算

3. 高级功能实现

3.1 对话记忆管理

python复制from typing import Dict, List
from collections import deque

class DialogueMemory:
    def __init__(self, max_history=10):
        self._sessions: Dict[str, deque] = {}
        self.max_history = max_history
    
    def get_history(self, session_id: str) -> List[dict]:
        """获取指定会话的历史记录"""
        return list(self._sessions.get(session_id, deque()))
    
    def add_message(self, session_id: str, role: str, content: str):
        """添加新消息到历史"""
        if session_id not in self._sessions:
            self._sessions[session_id] = deque(maxlen=self.max_history*2)
        
        self._sessions[session_id].append({
            "role": role,
            "content": content,
            "timestamp": time.time()
        })
    
    def build_context(self, session_id: str, query: str) -> str:
        """构建带上下文的Prompt"""
        history = self.get_history(session_id)
        if not history:
            return query
        
        context_lines = []
        for msg in history[-6:]:  # 最近3轮对话
            speaker = "用户" if msg["role"] == "user" else "助手"
            context_lines.append(f"{speaker}：{msg['content']}")
        
        return f"【历史对话】\n" + "\n".join(context_lines) + f"\n\n【当前问题】\n{query}"

内存优化技巧：

• 使用deque实现固定长度历史记录
• 按时间戳管理对话新鲜度
• 限制上下文长度避免Prompt过长

3.2 模型调用封装

python复制from dashscope import Generation
from langchain_community.llms import Tongyi

class QwenWrapper:
    def __init__(self, model="qwen-max"):
        self.model = model
        self.llm = Tongyi(model_name=model)
    
    def generate(self, prompt: str, temperature=0.7) -> str:
        """基础生成接口"""
        response = Generation.call(
            model=self.model,
            prompt=prompt,
            temperature=temperature
        )
        return response.output.text
    
    def tool_agent(self, tools: List[Tool], query: str) -> str:
        """工具调用代理"""
        agent = initialize_agent(
            tools,
            self.llm,
            agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
            verbose=True
        )
        return agent.run(query)

性能优化点：

• 设置合理的temperature平衡创造力和稳定性
• 使用结构化Agent提升工具调用准确率
• 实现请求批处理提高吞吐量

4. 部署与优化

4.1 服务部署方案

bash复制# 使用uvicorn部署
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

# 生产环境建议
uvicorn main:app --host 0.0.0.0 --port 8000 \
    --workers 8 \
    --limit-concurrency 100 \
    --timeout-keep-alive 30

部署配置建议：

• Worker数量 = CPU核心数 * 2 + 1
• 启用HTTP Keep-Alive减少连接开销
• 设置合理的并发限制防止过载

4.2 性能优化技巧

1. 缓存策略：

   • 高频查询结果缓存(如热门菜品)
   • 使用Redis缓存地理编码结果
   • 实现向量检索缓存层

2. 异步处理：

   python复制@app.post("/chat")
   async def chat_endpoint(query: str, session_id: str):
       # 异步调用LLM
       result = await asyncio.to_thread(llm.generate, query)
       return {"response": result}
   

3. 批量处理：

   python复制# 批量生成菜品描述
   def batch_generate_descriptions(items: List[MenuItem]):
       prompts = [f"生成{item.name}的吸引人描述" for item in items]
       return llm.batch_generate(prompts)
   

5. 实用技巧与问题排查

5.1 常见问题解决

问题1：LLM返回格式不稳定

现象：MENU_IDS有时返回JSON数组，有时返回纯文本

解决方案：

python复制# 添加输出解析器
from langchain.output_parsers import StructuredOutputParser

parser = StructuredOutputParser.from_used_parameters(
    parameters=[{
        "name": "menu_ids",
        "type": "array",
        "items": {"type": "string"},
        "description": "推荐菜品的ID列表"
    }]
)

@tool
def recommend_dishes(query: str) -> str:
    # ...
    response = llm.generate(prompt)
    return parser.parse(response)

问题2：地址识别错误

现象：把"朝阳区"误识别为城市名

优化方案：

python复制def normalize_address(address: str) -> str:
    """地址标准化处理"""
    # 1. 常见错误修正
    corrections = {
        "朝阳区": "北京市朝阳区",
        "浦东": "上海市浦东新区"
    }
    for wrong, right in corrections.items():
        address = address.replace(wrong, right)
    
    # 2. 去除特殊字符
    address = re.sub(r"[#@&]", "", address)
    
    return address

5.2 调试技巧

1. 对话历史检查：

   python复制def debug_history(session_id: str):
       history = memory.get_history(session_id)
       for i, msg in enumerate(history):
           print(f"[{i}] {msg['role']}: {msg['content']}")
   

2. Prompt模板验证：

   python复制def validate_prompt(template: str, inputs: dict):
       from langchain.prompts import PromptTemplate
       prompt = PromptTemplate.from_template(template)
       print("=== 完整Prompt ===")
       print(prompt.format(**inputs))
   

3. 性能分析：

   python复制import cProfile
   
   def profile_tool(tool_func, *args):
       profiler = cProfile.Profile()
       profiler.enable()
       result = tool_func(*args)
       profiler.disable()
       profiler.print_stats(sort="cumtime")
       return result
   

6. 扩展与改进方向

6.1 功能扩展建议

1. 多模态支持：

   • 上传菜品图片识别
   • 生成菜品视觉描述

2. 个性化推荐：

   python复制class UserPreference:
       def __init__(self):
           self.favorite_categories = set()
           self.allergies = set()
           self.price_range = (0, float('inf'))
   

3. 订单集成：

   • 直接通过对话下单
   • 支付状态查询
   • 订单进度跟踪

6.2 架构优化方向

1. 引入MCP协议：

   python复制from mcp_sdk import ToolServer
   
   server = ToolServer()
   server.register_tool(recommend_dishes)
   server.register_tool(check_delivery)
   server.start(port=8080)
   

2. 分布式会话管理：

   python复制from redis import Redis
   
   class RedisMemory(DialogueMemory):
       def __init__(self, redis_conn: Redis):
           self.redis = redis_conn
       
       def get_history(self, session_id: str) -> List[dict]:
           data = self.redis.get(f"dialogue:{session_id}")
           return json.loads(data) if data else []
   

3. 模型微调：

   python复制# 使用领域数据微调
   from transformers import Trainer
   
   trainer = Trainer(
       model=model,
       args=training_args,
       train_dataset=menu_dataset,
       eval_dataset=eval_dataset
   )
   trainer.train()
   

在实际部署中，我们还需要特别注意以下几点：

1. 对话服务的响应延迟要控制在3秒以内
2. 高峰期需要自动扩展计算资源
3. 实现完善的监控和告警系统
4. 定期更新菜品数据和配送范围

这个系统的优势在于将先进的AI技术与实际餐饮业务场景深度结合，既提升了用户体验，又为餐厅运营提供了智能化工具。随着持续迭代，可以逐步扩展成为完整的餐饮行业解决方案。