最近跟几个做餐饮的朋友聊天发现一个挺有意思的现象大家一边抱怨线下生意难做一边又对“数字化”这个词感到迷茫。开个店上个外卖平台、弄个扫码点餐就叫数字化了吗今天我想分享一个更“硬核”的案例——如何用一套开源的、可编程的“数字大脑”把一个传统的凉茶店变成一个能自动感知、决策和行动的智能体Agent。这听起来有点玄乎但背后的逻辑很实在。传统餐饮的痛点是什么人力成本高、经验依赖强、流程不标准、营销拍脑袋。一个老师傅走了口味可能就变了一个新店员不熟练出品速度就慢了今天该煮多少凉茶全凭店长“感觉”。我们想做的就是把这些“感觉”和“经验”变成代码和算法让机器来辅助甚至执行。所以这篇文章不是讲怎么煮凉茶也不是讲怎么选址装修。我们要解决的核心问题是一个技术开发者或具备技术思维的餐饮创业者如何利用现有的开源Agent框架为一家实体小店构建一个低成本、可迭代的自动化运营系统这个系统能干什么从自动根据天气和历史销量调整原料采购到识别常客自动推送优惠券再到后厨设备异常自动报修。我们将从零开始拆解这个“凉茶店数字大脑”的实现路径。读完本文你将获得一个完整的、可落地的技术方案思路理解如何将Agent思维应用于实体商业并得到一套可以直接参考或修改的代码框架。无论你是对AI应用感兴趣的开发者还是寻求降本增效的实体店主都能从中找到抓手。1. 为什么凉茶店需要“数字大脑”从痛点看技术价值在广东凉茶店遍布大街小巷生意模式看似简单煮茶、售卖、收银。但深入运营你会发现一堆琐碎却影响生死的问题库存浪费与缺货矛盾今天气温35度理论上清热祛湿的“癍痧”应该多备。但如果你只凭感觉多煮了50杯结果下午一场暴雨门店无人茶就馊了。反之如果备少了高峰期顾客买不到直接损失营业额和口碑。人力成本刚性上涨一个熟练店员要会认方、抓药、煲煮、销售培养周期长工资要求高。高峰期忙不过来闲时人力又浪费。顾客体验难以标准化老顾客王阿姨喜欢“廿四味”多加一点冰糖新店员可能不记得。这种个性化的服务依赖人的记忆极不稳定。营销活动效果不明今天在朋友圈发了一张“买一送一”的券来了多少人是新增客户还是老客薅羊毛对提升长期复购有帮助吗大多数小店店主算不清这笔账。设备管理滞后煲茶的电磁炉功率下降了煮一壶茶的时间从30分钟变成了50分钟直到某天彻底坏了才发现导致营业中断。传统的“门店ERP”或“SaaS收银系统”只能解决部分问题比如记录库存和收银。但它们缺乏感知-决策-执行的闭环能力。而“智能体Agent”技术恰恰擅长于此。一个智能体可以感知Perception通过API获取天气数据、通过IoT传感器获取设备状态、通过小程序获取用户订单行为。决策Decision基于规则或简单模型例如IF 气温30 AND 历史同期销量100 THEN 增加“癍痧”备料30%做出判断。执行Action调用另一个API向采购清单添加条目、向用户推送一条消息、向维修平台发送一个工单。为凉茶店打造“数字大脑”本质是构建一个由多个垂直领域智能体库存Agent、客服Agent、设备Agent协同工作的系统让机器处理可规则化的重复劳动让人专注于产品研发和顾客情感维系。2. 核心概念什么是智能体Agent与自动化脚本有何不同在开始构建之前必须厘清一个关键概念我们用的“Agent”和写一个Python定时脚本有什么区别自动化脚本是预定义流程的忠实执行者。例如一个每天上午10点检查库存低于阈值就发邮件的脚本。它没有“状态”概念不会因为昨天销量好就调整今天的阈值除非你手动改代码。智能体Agent则是一个更具自主性的程序。它通常包含几个核心组件状态State对当前环境的认知如“当前库存”、“实时天气”、“设备健康度”。策略Policy根据状态决定行动的规则或模型。这是其“智能”所在。执行器Actuator执行策略输出的动作如调用API、发送指令。学习可选根据动作的结果奖励或惩罚调整策略以追求长期目标如利润最大化。对于我们凉茶店场景我们主要构建的是“基于规则的智能体”它已经足够解决80%的问题。它的策略是一组“IF-THEN”规则但它的状态是实时更新的这使得它比静态脚本灵活。举个例子脚本思维每天固定采购黄芪10斤。Agent思维每小时获取“过去24小时黄芪相关产品的销量”和“未来48小时天气预报”。IF 销量增长趋势 15% AND 未来天气持续炎热 THEN 采购量 基础量10斤 * (1 增长趋势)。ELSE 采购量 10斤。Agent会根据环境变化动态调整行为这是其核心价值。3. 技术选型与环境准备我们不会使用庞大复杂的商业AI平台而是选择轻量、开源、易集成的技术栈确保小团队也能维护。核心框架LangChain虽然LangChain常与大语言模型LLM绑定但其提供的Agent、Tool、Chain抽象非常适合构建我们的系统。即使不使用LLM我们也可以用其框架来组织工作流。当然你也可以用纯Python调度器如schedule库和规则引擎如durable-rules自己搭建但LangChain生态更成熟。辅助工具栈数据感知层天气requests调用和风天气或高德天气API。库存/销售假设你有一个简单的数据库如SQLite或在线表格如腾讯文档、Airtable通过其API读取。设备IoT使用树莓派传感器模拟或商用IoT平台如阿里云IoT的SDK。决策与执行层规则引擎可以使用pyknow一个Python CLIPS实现或简单的Python-dict配置。动作执行requests调用各类第三方服务API如短信、邮件、工单系统。开发环境Python 3.9包管理pip或poetry代码编辑器VS Code 或 PyCharm环境搭建步骤创建项目目录并初始化虚拟环境。mkdir liangcha_agent cd liangcha_agent python -m venv venv # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate安装核心依赖。pip install langchain langchain-community requests schedule pydantic # 如果使用SQLitePython已内置sqlite3准备配置文件config.yaml存放API密钥、数据库连接等敏感信息。# config.yaml weather: api_key: your_hefeng_api_key city_id: 101280101 # 广州 inventory: db_path: ./data/inventory.db notification: email_host: smtp.qq.com email_user: your_emailqq.com email_pass: your_auth_code # 注意是授权码非密码4. 系统架构设计多个Agent如何协同工作我们的“凉茶店数字大脑”由多个单一职责的Agent组成它们通过共享状态数据库或消息队列进行松散耦合。[外部数据源] -- [感知层] -- [中央状态/事件总线] -- [智能体集群] -- [执行层] -- [外部世界] 天气API (SQLite/Redis) 库存Agent 采购API 销售系统 (或消息队列) 客服Agent 消息推送API IoT传感器 设备Agent 工单API库存Agent负责库存优化。感知销售速度、天气决策采购建议执行更新采购清单。客服Agent负责顾客互动。感知用户订单行为如通过小程序决策是否发送优惠券或问候执行调用消息推送。设备Agent负责设备监控。感知设备传感器数据温度、电流决策设备是否异常执行发送维修告警。我们首先实现最核心的库存Agent。5. 实战构建库存智能体Inventory Agent5.1 定义状态与工具Tools首先我们定义Agent能感知和操作的东西。在LangChain中这通过Tool来实现。# tools/weather_tool.py import requests import yaml from datetime import datetime from typing import Optional from pydantic import BaseModel, Field # 加载配置 with open(config.yaml, r) as f: config yaml.safe_load(f) class WeatherInput(BaseModel): 获取天气的输入参数这里简单处理只针对配置的城市 # 可以扩展为接收城市参数这里固定为配置城市 pass class WeatherTool: name get_current_weather description 获取当前城市的天气情况包括温度和天气现象。用于预测凉茶需求。 args_schema WeatherInput def _run(self) - str: 执行获取天气的逻辑 api_key config[weather][api_key] city_id config[weather][city_id] url fhttps://devapi.qweather.com/v7/weather/now?location{city_id}key{api_key} try: response requests.get(url, timeout10) data response.json() if data[code] 200: now data[now] temp now[temp] # 温度 text now[text] # 天气现象如“晴”、“雨” return f当前天气{text}温度{temp}摄氏度。 else: return f获取天气失败{data.get(message, 未知错误)} except Exception as e: return f调用天气API异常{str(e)} # tools/inventory_tool.py import sqlite3 import yaml from pydantic import BaseModel, Field with open(config.yaml, r) as f: config yaml.safe_load(f) class QueryInventoryInput(BaseModel): 查询特定凉茶库存的输入 tea_name: str Field(description凉茶名称例如癍痧、廿四味、菊花雪梨) class QueryInventoryTool: name query_inventory description 查询指定凉茶当前的库存杯数。 args_schema QueryInventoryInput def _run(self, tea_name: str) - str: conn sqlite3.connect(config[inventory][db_path]) cursor conn.cursor() cursor.execute(SELECT quantity FROM inventory WHERE name?, (tea_name,)) row cursor.fetchone() conn.close() if row: return f{tea_name}的当前库存是{row[0]}杯。 else: return f未找到凉茶{tea_name} 的库存记录。5.2 定义决策策略Policy我们用一个简单的规则引擎来做决策。这里为了直观用一个函数模拟。# policy/inventory_policy.py def make_inventory_decision(tea_name: str, current_inventory: int, weather_info: str, historical_sales: int) - dict: 基于规则做出库存决策。 返回一个字典包含动作和参数。 decision {action: do_nothing, message: 库存充足无需操作。} # 规则1安全库存检查 safety_stock 20 # 安全库存为20杯 if current_inventory safety_stock: decision[action] generate_purchase_order decision[tea_name] tea_name # 简单逻辑补货到50杯如果天气热且历史销量好多补一些 base_replenish 50 - current_inventory extra 0 # 规则2天气因素 if 晴 in weather_info or 热 in weather_info: extra 10 if 雨 in weather_info: extra - 5 # 下雨天可能需求减少 # 规则3历史销售趋势这里简化处理 if historical_sales 80: # 假设历史销量好 extra 15 final_replenish base_replenish extra decision[quantity] final_replenish decision[message] f{tea_name}库存低于安全线。建议采购{final_replenish}杯。依据当前库存{current_inventory}天气“{weather_info}”历史销量{historical_sales}。 return decision5.3 组装智能体并运行我们将工具、策略和记忆这里用简单变量组装成一个可执行的Agent循环。# agents/inventory_agent.py import schedule import time from tools.weather_tool import WeatherTool from tools.inventory_tool import QueryInventoryTool from policy.inventory_policy import make_inventory_decision # 假设有一个模拟历史销量的函数 from data.mock_sales import get_historical_sales_today class InventoryAgent: def __init__(self): self.weather_tool WeatherTool() self.query_tool QueryInventoryTool() # 模拟一个简单的“待办事项”列表作为Agent的记忆/输出 self.task_list [] def run_one_cycle(self): 执行一个完整的感知-决策周期 print(f\n 库存Agent运行周期 {time.strftime(%Y-%m-%d %H:%M:%S)} ) # 1. 感知 Perception weather_info self.weather_tool._run() print(f[感知] {weather_info}) # 监控的凉茶列表 teas_to_check [癍痧, 廿四味, 菊花雪梨] for tea in teas_to_check: inventory_status self.query_tool._run(tea) print(f[感知] {inventory_status}) # 从返回字符串中提取库存数字实际项目应从数据库直接取数字 # 这里简化处理假设我们通过另一个函数直接拿到数字 current_inventory 15 # 模拟当前库存实际应从query_tool结果解析或直接查库 historical_sales get_historical_sales_today(tea) # 模拟获取历史销量 # 2. 决策 Decision decision make_inventory_decision(tea, current_inventory, weather_info, historical_sales) print(f[决策] 针对{tea}{decision[message]}) # 3. 执行 Action (这里模拟实际是调用采购API) if decision[action] generate_purchase_order: task { type: PURCHASE, item: decision[tea_name], quantity: decision[quantity], reason: decision[message], timestamp: time.time() } self.task_list.append(task) print(f[执行] 已生成采购任务{task}) print(f本轮结束待处理任务数{len(self.task_list)}) def get_pending_tasks(self): 获取当前所有待执行的任务 return self.task_list.copy() # 主程序 if __name__ __main__: agent InventoryAgent() # 使用schedule库定时运行例如每2小时运行一次 schedule.every(2).hours.do(agent.run_one_cycle) # 立即运行一次 agent.run_one_cycle() print(\nAgent已启动按 CtrlC 退出。) try: while True: schedule.run_pending() time.sleep(1) except KeyboardInterrupt: print(\nAgent已停止。) # 打印最终的任务列表 print(最终待处理采购任务) for task in agent.get_pending_tasks(): print(task)6. 运行结果与效果验证运行上述inventory_agent.py你会在控制台看到类似以下的输出 库存Agent运行周期 2023-10-27 14:00:00 [感知] 当前天气晴温度32摄氏度。 [感知] 癍痧的当前库存是15杯。 [决策] 针对癍痧癍痧库存低于安全线。建议采购60杯。依据当前库存15天气“当前天气晴温度32摄氏度。”历史销量90。 [执行] 已生成采购任务{type: PURCHASE, item: 癍痧, quantity: 60, reason: ..., timestamp: 1698398400.123456} [感知] 廿四味的当前库存是15杯。 [决策] 针对廿四味廿四味库存低于安全线。建议采购55杯。依据当前库存15天气“当前天气晴温度32摄氏度。”历史销量70。 [执行] 已生成采购任务{type: PURCHASE, item: 廿四味, quantity: 55, reason: ..., timestamp: 1698398400.234567} [感知] 菊花雪梨的当前库存是15杯。 [决策] 针对菊花雪梨菊花雪梨库存低于安全线。建议采购50杯。依据当前库存15天气“当前天气晴温度32摄氏度。”历史销量40。 [执行] 已生成采购任务{type: PURCHASE, item: 菊花雪梨, quantity: 50, reason: ..., timestamp: 1698398400.345678} 本轮结束待处理任务数3 Agent已启动按 CtrlC 退出。如何验证Agent工作正常感知验证检查日志中天气信息是否正常获取库存查询是否准确。决策验证手动修改inventory_policy.py中的安全库存值如改为50再次运行观察决策是否变化应不再触发采购。修改模拟的current_inventory值观察决策逻辑是否符合预期。执行验证将[执行]部分的模拟代码替换为真实的API调用如调用一个添加采购单到腾讯文档的接口查看采购单是否成功创建。至此一个最基础的、能自动运行的库存管理智能体就已经构建完成。它每隔2小时自动检查并根据规则生成采购建议。7. 扩展与进阶构建客服Agent与设备Agent库存Agent跑通后我们可以依葫芦画瓢构建其他Agent。7.1 客服Agent基于用户行为的自动化互动核心逻辑当用户在小程序完成一笔订单后系统触发一个事件。客服Agent监听该事件查询该用户的历史订单如最近30天购买次数根据规则决定是否发放优惠券。# agents/customer_agent.py class CustomerAgent: def __init__(self): self.rules [ {name: 新客首单奖励, condition: lambda user: user.order_count 1, action: send_coupon, coupon_type: WELCOME_10OFF}, {name: 老客回头促活, condition: lambda user: user.last_order_days_ago 30, action: send_coupon, coupon_type: COMEBACK_5OFF}, {name: 高价值客户关怀, condition: lambda user: user.total_spent 500, action: send_birthday_gift}, ] def on_order_completed(self, user_id, order_amount): 模拟订单完成事件处理 user self._get_user_profile(user_id) # 从数据库获取用户画像 for rule in self.rules: if rule[condition](user): self._execute_action(rule[action], user, rule) break # 简单处理只执行第一条匹配规则7.2 设备Agent基于IoT数据的预测性维护核心逻辑通过树莓派读取连接电磁炉的智能插座数据功率、用电量。设备Agent持续监控如果发现“平均功率持续下降”或“今日用电量异常高于历史同期”则判断设备可能老化或故障自动生成维修检查工单。# agents/device_agent.py import statistics from datetime import datetime, timedelta class DeviceAgent: def __init__(self, device_id): self.device_id device_id self.power_readings [] # 存储近期功率读数 self.alert_threshold 0.85 # 功率下降15%则告警 def ingest_reading(self, power_watt, timestamp): 接收设备上报的数据 self.power_readings.append({power: power_watt, ts: timestamp}) # 保留最近100条数据 if len(self.power_readings) 100: self.power_readings.pop(0) self._analyze() def _analyze(self): if len(self.power_readings) 20: return recent self.power_readings[-20:] historical self.power_readings[-100:-20] avg_recent statistics.mean([r[power] for r in recent]) avg_historical statistics.mean([r[power] for r in historical]) if avg_historical 0 and (avg_recent / avg_historical) self.alert_threshold: print(f[设备告警] 设备{self.device_id}近期平均功率({avg_recent:.1f}W)较历史({avg_historical:.1f}W)下降超过15%建议检查) # 调用工单API # self._create_maintenance_ticket(...)8. 常见问题与排查思路在开发和运行此类Agent系统时你会遇到一些典型问题。问题现象可能原因排查方式解决方案Agent定时任务不执行1.schedule库未正确调用run_pending()。2. 主线程被阻塞。3. 脚本因异常退出。1. 检查主循环是否在运行。2. 在周期函数开始打印日志。3. 查看是否有未捕获的异常。1. 确保while True循环存在且sleep时间短如1秒。2. 用try...except包裹周期函数体。3. 考虑使用systemd或supervisor托管进程。调用外部API失败1. 网络问题。2. API密钥无效或过期。3. 请求频率超限。4. API响应格式变化。1. 检查网络连通性ping/curl。2. 检查config.yaml中的密钥。3. 查看API提供商的后台统计。4. 打印API原始响应。1. 增加请求超时和重试机制。2. 将密钥存储在环境变量中更安全。3. 遵守API调用频率限制。4. 编写更健壮的响应解析代码。决策规则不生效或逻辑错误1. 规则条件condition写错。2. 输入数据格式与预期不符。3. 规则优先级或冲突。1. 在决策函数内打印中间变量。2. 使用单元测试验证单个规则。3. 检查规则集的遍历顺序。1. 采用更清晰的规则表达方式如将规则写入配置文件。2. 对输入数据进行清洗和类型验证。3. 引入简单的规则引擎库如pyknow管理复杂规则。多个Agent之间数据不同步Agent之间直接读写数据库存在并发问题。观察数据库记录是否出现错乱。1. 对于高频写操作使用数据库事务。2. 引入消息队列如Redis Pub/Sub, RabbitMQ让Agent通过事件通信而非直接共享状态。系统资源占用过高1. Agent运行频率太高。2. 单个周期内处理数据量过大。3. 存在内存泄漏。1. 使用top或htop监控进程。2. 使用memory_profiler等工具分析。1. 合理设置调度间隔非核心任务可以每天一次。2. 优化数据处理逻辑分批处理。3. 确保数据库连接、HTTP会话等资源在使用后正确关闭。9. 最佳实践与工程化建议将实验性的脚本升级为可维护、可扩展的生产系统需要注意以下几点配置中心化所有API密钥、数据库连接、决策阈值如安全库存数必须放在配置文件如config.yaml或环境变量中绝对不要硬编码在代码里。日志与监控替换print语句为标准的日志库如logging并设置不同级别INFO, WARNING, ERROR。关键动作如生成采购单、发送消息必须记录日志。可以考虑将运行状态最后一次成功运行时间、生成任务数写入数据库便于前端展示。错误处理与重试所有外部调用API、数据库都必须有完善的try...except和重试机制。对于非关键任务失败后可以记录错误并继续运行避免一个点失败导致整个Agent崩溃。版本控制与部署使用Git管理代码。生产环境部署建议使用容器化Docker便于环境一致性和快速回滚。人工审核与干预对于Agent生成的关键决策如大额采购、向客户发送消息初期应加入“人工审核”环节。可以设计一个简单的管理后台展示Agent推荐的动作由店长点击确认后再执行。循序渐进小步快跑不要试图一次性构建所有Agent。从库存Agent开始跑通数据流和决策执行闭环解决一个最痛的痛点。验证价值后再逐步扩展客服、设备等Agent。规则的可解释性确保每条决策规则都是可解释的就像我们make_inventory_decision函数中的reason字段。这有助于调试也能让业务人员店长信任这个“数字大脑”。为一家凉茶店构建“数字大脑”听起来像是一个大工程但通过智能体Agent的模块化思想我们可以将其拆解为一系列小而具体的自动化任务。本文带你从0到1实现了最核心的库存管理智能体并勾勒了客服、设备Agent的蓝图。这套方案的价值不在于用了多高深的技术而在于它提供了一种将业务知识转化为可执行代码的范式。你可以从今天开始用文中的代码框架替换掉你业务中的真实数据源和API先让一个环节比如每日销量预测自动运转起来。当机器帮你处理掉那些重复、枯燥的决策时你就能更专注于提升凉茶的口感、研究新的配方或者 simply享受一杯自己煮的凉茶。技术最终要回归到为人服务让生意更简单让生活更从容。