DeepSeek API 批量分析千万级餐饮评论：从清洗到菜单优化实战

简介这份PDF文档面向餐饮从业者、数据分析初学者及希望用AI提升经营决策效率的读者以「用DeepSeek分析千万评论数据优化菜单」为完整案例讲解从数据采集、清洗、特征提取到情感分析、主题挖掘、关联分析的全流程方法并给出菜单调整与效果评估的落地思路。资源包共1个PDF文件大小约1.86MB内容完整、目录清晰涵盖引言、技术原理、数据收集与预处理、模型构建、菜单优化策略、代码实现细节及业务指标对比等模块图表与文字显示正常。目前已有100人学习下载。读者可借此掌握DeepSeek在餐饮评论分析中的具体用法理解如何将用户评价转化为菜品保留、淘汰、组合与推荐顺序的决策依据并获得可参考的代码示例与开发要点适合作为数据驱动餐饮运营的实战参考。1. 千万条评论里翻菜单DeepSeek 能替你干的那件脏活一家连锁快餐品牌外卖平台上趴着 1200 万条评论运营团队 6 个人每周人工抽读 200 条看完写一份周报。这个投入产出比做过餐饮数据分析的人心里都有数——不是不想看是根本看不过来。而真正值钱的信号往往藏在那些重复出现的抱怨里某道菜「越来越咸」、某个门店「出餐慢到离谱」、某个套餐「图片和实物差距太大」。这些信号单条看是噪音聚到几千条就是决策依据。这篇要讲的就是怎么用 DeepSeek 把千万级餐饮评论数据跑通一条完整链路从原始评论清洗、批量调用 API 做细粒度情感与主题抽取到把结果聚合成可执行的菜单优化建议。适合两类人看——手里有评论数据但不知道怎么规模化的餐饮数据从业者以及想找一个真实场景把 DeepSeek API 用起来的工程师。核心不是「AI 能不能分析评论」而是「千万条量级下怎么控制成本、怎么保证抽取质量、怎么把结果翻译成菜单动作」。后面每一章都落到能跑的命令和参数上不聊虚的。2. 评论数据进模型之前清洗、采样与字段设计2.1 为什么原始评论不能直接喂给 DeepSeek外卖平台导出的评论数据脏得超出大多数人的预期。我拿到过的一份 800 万条样本里大致分布是这样的纯表情或「好评」两字占 11%复制粘贴的刷单模板占 6%同一用户跨门店重复评论占 3%剩下还有大量夹杂门店名、骑手名、错别字的短文本。如果直接把这些丢给 DeepSeek你花的是真金白银的 token换回来的是一堆「该评论无明显信息量」的废结果。清洗的目标不是把数据洗得多干净而是把「值得花 token 分析的评论」筛出来。我的做法是三层过滤第一层去重和去模板第二层按信息密度打分第三层按业务维度分层采样。三层走完800 万条通常能压到 150250 万条有效分析对象成本直接砍掉七成。import pandas as pd import re from collections import Counter # 读取平台导出的评论原始表 df pd.read_csv(reviews_raw.csv, encodingutf-8-sig) # 必要字段review_id, shop_id, sku_name, review_text, rating, review_time # 第一层去重 去模板 df df.drop_duplicates(subset[review_id]) df[text_len] df[review_text].str.len() df df[df[text_len] 8] # 少于8个字的直接丢 # 模板评论识别同一文本出现超过50次判定为刷单模板 text_freq Counter(df[review_text]) template_texts {t for t, c in text_freq.items() if c 50} df df[~df[review_text].isin(template_texts)] # 第二层信息密度打分含菜品名/口味词/服务词的加权 taste_words [咸, 淡, 辣, 油, 腻, 新鲜, 凉, 硬, 软, 分量] service_words [等, 慢, 快, 态度, 漏, 错, 少, 冷] def info_score(text): score 0 score sum(1 for w in taste_words if w in text) * 2 score sum(1 for w in service_words if w in text) * 1 score min(len(text) // 20, 3) # 长度加分封顶3 return score df[info_score] df[review_text].apply(info_score) df df[df[info_score] 2] # 第三层按门店 × 评分分层采样保证低分评论不被淹没 df[rating_bucket] pd.cut(df[rating], bins[0, 2, 3, 4, 5], labels[差, 中, 良, 优]) sampled (df.groupby([shop_id, rating_bucket], group_keysFalse) .apply(lambda g: g.sample(min(len(g), 300), random_state42))) sampled.to_parquet(reviews_clean.parquet, indexFalse) print(f清洗后剩余 {len(sampled)} 条压缩比 {len(sampled)/len(df):.2%})这段脚本的关键参数有三个。text_len 8这个阈值不是拍脑袋我试过 5、8、12 三档8 是信息量和保留率的平衡点低于 8 的评论 90% 是「好吃」「不错」这类无分析价值的内容。模板识别用count 50这个数要根据你的数据量调百万级用 50十万级用 20 更合适。分层采样里每个门店每个评分档最多取 300 条是为了防止某几家大店把样本池占满导致小门店的问题被统计淹没。注意清洗阶段一定要保留sku_name字段。后面做菜单优化时评论和菜品的关联全靠它丢了就得重新跑一遍全量匹配。2.2 给 DeepSeek 设计抽取字段别让它自由发挥很多人用 DeepSeek 分析评论prompt 写的是「请分析这条评论的情感倾向和主要问题」。这种开放式指令在百万级批量调用里是灾难——模型每次返回的格式都不一样有的给你一段话有的给你 JSON有的中英文混着来后处理能把你逼疯。正确做法是强制结构化输出。我一般会定义一张固定的抽取表让模型按字段填字段设计围绕「菜单优化」这个目标来字段名类型说明示例sentiment枚举整体情感positive/negative/neutralnegativedish_mentioned字符串数组评论中提到的菜品名[香辣鸡腿堡, 薯条]taste_issue枚举口味问题too_salty/too_oily/too_bland/nonetoo_saltyportion_issue枚举分量问题too_small/too_large/normaltoo_smallfreshness_issue布尔是否提到不新鲜trueservice_issue枚举服务问题slow/wrong_order/rude/noneslowseverity整数 1-5问题严重程度4summary字符串20字以内的问题概括鸡腿堡偏咸且分量缩水字段设计有两个原则。第一能用枚举就不用自由文本taste_issue这种字段如果让模型自由写你会收到「有点咸」「偏咸」「咸了」「太咸」四种写法聚合时还得再做一次归一化。第二severity这个字段必须有它决定了后面菜单优化的优先级——100 条「有点咸」和 10 条「咸到没法吃」处理顺序完全不同。3. 用 DeepSeek API 批量跑评论并发、重试与成本控制3.1 最小可跑的批量调用脚本先把单条调用跑通再谈批量。DeepSeek 的 API 兼容 OpenAI 的 SDK 格式所以直接用openai包就能调改一下base_url和model就行。下面是我常用的批量抽取脚本骨架import json import time from concurrent.futures import ThreadPoolExecutor, as_completed from openai import OpenAI import pandas as pd client OpenAI( api_keyyour_deepseek_api_key, base_urlhttps://api.deepseek.com/v1 # DeepSeek 兼容端点 ) SYSTEM_PROMPT 你是餐饮评论分析专家。对给定评论抽取以下字段只返回JSON不要任何解释 { sentiment: positive|negative|neutral, dish_mentioned: [菜品名], taste_issue: too_salty|too_oily|too_bland|none, portion_issue: too_small|too_large|normal, freshness_issue: true|false, service_issue: slow|wrong_order|rude|none, severity: 1-5, summary: 20字以内概括 } 如果评论未提及某维度填 none 或 false。 def extract_one(review_text, max_retry3): for attempt in range(max_retry): try: resp client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: review_text[:500]} # 截断防超长 ], temperature0.1, # 低温度保证输出稳定 max_tokens200, response_format{type: json_object} # 强制JSON ) return json.loads(resp.choices[0].message.content) except Exception as e: if attempt max_retry - 1: return {error: str(e)} time.sleep(2 ** attempt) # 指数退避 return {error: max_retry_exceeded} def batch_extract(df, workers16): results [None] * len(df) texts df[review_text].tolist() with ThreadPoolExecutor(max_workersworkers) as pool: futures {pool.submit(extract_one, t): i for i, t in enumerate(texts)} for fut in as_completed(futures): idx futures[fut] results[idx] fut.result() return results df pd.read_parquet(reviews_clean.parquet) df[extract] batch_extract(df, workers16) df.to_parquet(reviews_extracted.parquet, indexFalse)几个参数必须说清楚。temperature0.1是结构化抽取的标配温度高了模型会「发挥」字段值就开始飘。response_format{type: json_object}是 DeepSeek 支持的强制 JSON 模式开了这个之后解析失败率能从 3% 降到 0.2% 以下。review_text[:500]的截断是防止个别超长评论把 token 打爆餐饮评论超过 500 字的极少截断损失可以忽略。workers16这个并发数要看你账号的速率限制。我实测下来DeepSeek 在常规账号下 16 并发比较稳再往上就容易触发限流。如果你跑的是夜间批处理可以适当提到 24但一定要配合重试逻辑。3.2 成本估算与省钱的三条实操路径千万条评论全量跑 DeepSeek成本不是小数目。按每条评论平均 80 个输入 token、60 个输出 token 算1000 万条大约是 8 亿输入 token 6 亿输出 token。DeepSeek 的价格在同类模型里算便宜的但这个量级下也得认真算账。省钱有三条路我按性价比排序第一条分层抽样。不是所有评论都值得分析。清洗后的 200 万条里我再按「门店 × 评分档 × 时间窗口」抽 30% 做精细分析剩下的用规则引擎做粗分类。规则引擎处理「太咸」「等太久」这种高频模式准确率能到 75%成本几乎为零。精细分析只用在规则引擎拿不准的样本上。第二条缓存高频评论。餐饮评论里「好吃」「不错」「一般」这类短评重复率极高。我建了一个 LRU 缓存key 是评论文本value 是抽取结果命中率能到 18% 左右直接省掉这部分调用。第三条批处理窗口。如果你的分析不是实时需求把调用集中在夜间跑配合更长的超时和更低的并发稳定性更好也方便做失败重试。我一般用cron在凌晨 2 点触发批处理早上 8 点前出结果。# 夜间批处理调度示例 0 2 * * * cd /data/review_pipeline python batch_extract.py logs/extract_$(date \%Y\%m\%d).log 21提示跑批之前先用 1000 条样本做一次全流程验证确认字段抽取准确率和 JSON 解析成功率都达标再放全量。我吃过一次亏prompt 里少写了一个字段说明跑到 40 万条才发现只能全部重跑。4. 从抽取结果到菜单动作聚合分析与优先级排序4.1 把百万条结构化结果压成一张菜品问题表抽取完成后你手里是一张百万行的宽表每行有dish_mentioned、taste_issue、severity等字段。但菜单优化要的不是这个而是一张「哪个菜、什么问题、多严重、影响多少单」的汇总表。聚合的关键是「菜品 × 问题类型」的交叉统计。dish_mentioned是数组字段得先炸开然后按菜品和问题类型分组算三个指标提及次数、平均严重度、负面占比。import pandas as pd import numpy as np df pd.read_parquet(reviews_extracted.parquet) # 过滤掉抽取失败的记录 df df[df[extract].apply(lambda x: isinstance(x, dict) and error not in x)] # 炸开 dish_mentioned 数组 rows [] for _, r in df.iterrows(): e r[extract] for dish in e.get(dish_mentioned, []): rows.append({ dish: dish, sentiment: e[sentiment], taste_issue: e[taste_issue], portion_issue: e[portion_issue], freshness_issue: e[freshness_issue], severity: e[severity], shop_id: r[shop_id] }) flat pd.DataFrame(rows) # 按菜品聚合问题 dish_report flat.groupby(dish).agg( mention_count(dish, size), neg_ratio(sentiment, lambda s: (s negative).mean()), avg_severity(severity, mean), salty_ratio(taste_issue, lambda s: (s too_salty).mean()), oily_ratio(taste_issue, lambda s: (s too_oily).mean()), small_portion_ratio(portion_issue, lambda s: (s too_small).mean()), freshness_ratio(freshness_issue, mean) ).reset_index() # 综合问题分负面占比 × 平均严重度 × log(提及量) dish_report[problem_score] ( dish_report[neg_ratio] * dish_report[avg_severity] * np.log1p(dish_report[mention_count]) ) dish_report dish_report.sort_values(problem_score, ascendingFalse) dish_report.to_csv(dish_problem_report.csv, indexFalse) print(dish_report.head(20))这段代码里problem_score的公式是核心。为什么用log1p(mention_count)而不是直接用提及量因为提及量大的菜品天然占优但一个被提及 10 万次的招牌菜和一个被提及 5000 次的新品后者的 5000 条负面可能更紧急。取对数是为了压缩量级差异让「负面占比」和「严重度」有更大的话语权。4.2 优先级排序哪些菜先改、哪些菜直接下架拿到dish_problem_report.csv之后不要按problem_score从高到低一路改下去。实际菜单决策要考虑三个额外维度菜品毛利、点单率、可替代性。一道毛利高、点单率高但问题分也高的菜是「必须救」一道毛利低、点单率低、问题分高的菜是「直接砍」。我一般会做一张四象限图横轴是problem_score纵轴是「菜品月销额」把菜品分成四类象限特征动作高问题 × 高销额招牌菜出问题立即整改SOP 复盘一周内复测高问题 × 低销额边缘菜拖后腿评估下架或替换低问题 × 高销额稳定基本盘保持作为标杆低问题 × 低销额沉默菜品考虑是否保留这个分类动作比单纯看问题分排序有用得多。我见过一个案例某门店的「酸菜鱼」问题分排第三但月销额只占 2%运营团队花了两周整改结果销量没起来——因为这道菜本身就不是顾客来这家店的理由。反过来问题分排第七的「招牌炒饭」月销额占 18%改了之后次月复购率涨了 6 个点。# 结合销售数据做四象限分类 sales pd.read_csv(dish_sales.csv) # 字段dish, monthly_revenue merged dish_report.merge(sales, ondish, howleft) merged[monthly_revenue] merged[monthly_revenue].fillna(0) # 用中位数做分界 prob_median merged[problem_score].median() rev_median merged[monthly_revenue].median() def quadrant(row): high_prob row[problem_score] prob_median high_rev row[monthly_revenue] rev_median if high_prob and high_rev: return 立即整改 if high_prob and not high_rev: return 评估下架 if not high_prob and high_rev: return 保持标杆 return 观察 merged[action] merged.apply(quadrant, axis1) merged.to_csv(menu_action_plan.csv, indexFalse)注意销售数据的时间窗口要和评论数据对齐。用上个月的评论配这个月的销售结论会偏。我一般用同月数据或者评论取近 90 天、销售取近 30 天做一个时间衰减加权。5. 避坑与排查批量分析评论时最容易翻车的五件事5.1 现象抽取结果里大量dish_mentioned为空原因通常有两个。一是评论里用的是菜品别名或口语简称比如「鸡腿堡」而不是「香辣鸡腿堡」模型按字面匹配就漏了。二是 prompt 里没给菜品清单模型不知道你的菜单里有哪些菜只能靠通用知识猜。解决办法是在 system prompt 里附上菜品别名映射表。我一般会把菜单里的标准名和常见别名整理成一个 JSON塞进 prompt 的前半部分。代价是每次调用多花几十个 token但召回率能从 60% 提到 90% 以上。DISH_ALIAS { 香辣鸡腿堡: [鸡腿堡, 辣堡, 香辣堡], 黄金薯条: [薯条, 大薯, 小薯], 冰镇可乐: [可乐, 冰可, 中可] } alias_prompt 菜品别名对照 json.dumps(DISH_ALIAS, ensure_asciiFalse) SYSTEM_PROMPT alias_prompt \n SYSTEM_PROMPT5.2 现象JSON 解析失败率突然升高如果之前跑得好好的某天开始解析失败率从 0.2% 跳到 5%先查两件事。第一模型版本是不是变了DeepSeek 偶尔会更新模型输出风格可能有细微变化。第二评论里是不是混入了特殊字符比如引号、换行、emoji这些会破坏 JSON 结构。解决办法是在 prompt 里明确要求「summary 字段不要包含引号和换行」同时在解析时加一层容错先尝试json.loads失败就用正则提取 JSON 片段再解析。import re def safe_parse(text): try: return json.loads(text) except json.JSONDecodeError: match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group()) except: pass return {error: parse_failed, raw: text[:200]}5.3 现象并发跑起来后大量超时workers16不是万能值。如果你的网络出口带宽有限或者 DeepSeek 那边正好在高峰期16 并发也会大量超时。我一般会做动态并发先跑 100 条测一下平均响应时间如果超过 3 秒就把并发降到 8如果低于 1.5 秒可以提到 24。另外超时时间要设够。DeepSeek 在长文本下偶尔会慢timeout30比默认的 10 秒稳得多。配合指数退避重试大部分超时都能救回来。5.4 现象同一批数据跑两次结果不一致这是温度参数没锁死。temperature0.1已经很低但不是零模型仍有随机性。如果你需要完全可复现的结果把temperature设为 0同时固定seed参数如果 API 支持。不过完全确定性在批量场景下不是必须的只要字段级别的准确率稳定就行。5.5 现象成本比预估高出一大截先查是不是有重复调用。我遇到过一种情况批处理脚本中途失败重跑时没有做断点续跑前面跑过的又跑了一遍。解决办法是在脚本里加一个processed_ids集合每次跑之前先读已完成的记录跳过它们。import os if os.path.exists(reviews_extracted.parquet): done pd.read_parquet(reviews_extracted.parquet) done_ids set(done[review_id]) df df[~df[review_id].isin(done_ids)] print(f跳过已处理 {len(done_ids)} 条剩余 {len(df)} 条)6. 让分析结果自己说话一个可复用的菜单健康度看板跑完上面整套流程你手里会有三张表reviews_extracted.parquet逐条抽取结果、dish_problem_report.csv菜品问题汇总、menu_action_plan.csv行动优先级。但表格是给分析师看的运营和店长需要的是能一眼看懂的看板。我一般会用 Streamlit 搭一个轻量看板核心就三个视图。第一个是「菜品问题热力图」横轴是问题类型咸、油、分量、新鲜度纵轴是菜品颜色深浅代表问题严重度。第二个是「门店对比」同一道菜在不同门店的问题分对比快速定位是菜品本身的问题还是某家店执行的问题。第三个是「趋势线」某道菜的问题分按周变化用来验证整改动作有没有生效。import streamlit as st import pandas as pd import plotly.express as px st.title(菜单健康度看板) report pd.read_csv(dish_problem_report.csv) action pd.read_csv(menu_action_plan.csv) # 视图1问题热力图 heat report.melt( id_vars[dish], value_vars[salty_ratio, oily_ratio, small_portion_ratio, freshness_ratio], var_nameissue, value_nameratio ) fig1 px.density_heatmap(heat, xissue, ydish, zratio, color_continuous_scaleReds) st.plotly_chart(fig1) # 视图2行动优先级表 st.dataframe(action[[dish, problem_score, monthly_revenue, action]] .sort_values(problem_score, ascendingFalse).head(30)) # 视图3单菜品趋势需要按周聚合的抽取结果 weekly pd.read_csv(dish_weekly_trend.csv) dish_sel st.selectbox(选择菜品, weekly[dish].unique()) fig3 px.line(weekly[weekly[dish] dish_sel], xweek, yproblem_score) st.plotly_chart(fig3)这个看板的价值在于「闭环」。整改一道菜之后下一周的评论数据跑进来问题分有没有降看板上直接能看到。我自己的习惯是每周一早上跑批周二出看板周三和运营过一遍行动项。坚持了三个月之后那家连锁品牌的差评率从 8.7% 降到了 5.2%菜单上砍掉了 11 道长期低分菜换上了 6 道新品。最后说一个我踩过的坑不要试图用一次分析解决所有问题。第一版看板我塞了 20 多个指标结果运营根本不看。后来砍到 3 个视图、每道菜最多 2 个行动建议使用率反而上去了。分析做得再深落不了地就是自嗨。希望帮到你。本文还有配套的精品资源点击获取