背景:毕设里“看天穿衣”到底卡在哪
去年做毕设,选题是“根据天气给用户搭衣服”,听起来简单,真动手才发现坑比衣柜还深。
- 天气字段多、单位杂,光一个“feels_like”就能返回摄氏度、华氏度、开尔文三种格式,手动 if-else 写到眼花。
- 推荐逻辑全硬编码:温度<10℃就推羽绒服,结果广州 9℃+90%湿度,用户穿成“移动桑拿”。
- 把整套规则写进后端,后续想换模型、加缓存、做 AB 实验,都得重新发版,耦合到哭。
于是我把目光转向“AI 辅助开发”——让 LLM 当“策略编译器”,我只管定义边界和兜底,结果两周就调通 MVP,老师看完只改了两个标点。
技术选型:直接 LLM 黑盒 vs LangChain 白盒
| 方案 | 优点 | 缺点 | 结论 |
|---|---|---|---|
| 裸调 GPT-4 | 上手快,一句 prompt 就能出推荐 | 字段不稳定、幻觉多、难做单元测试 | 放弃 |
| 传统规则引擎 | 输出可控,零 token 成本 | 代码膨胀,新城市新风速都要改 | 放弃 |
| LangChain + Pydantic | 结构化输出、可缓存、可插拔 | 多引一个库,学习曲线略陡 | 采用 |
核心思路:把“天气 JSON”→“衣物对象”的映射交给 LLM,但用 Pydantic 把输出锁死成结构体,既保留自然语言灵活性,又保住工程严谨性。
核心实现三步走
1. 天气数据获取:OpenWeather One Call 3.0
官方 1000 次/天免费额度,毕设够用。返回字段里我们只用:
- current.temp
- current.weather[0].main (Rain/Snow/Clear...)
- current.humidity
- current.wind_speed
用 aiohttp 异步拉取,500ms 内返回,超时重试 2 次即可。
2. 衣物知识库:轻量级“向量+规则”混合
- 规则层:先查城市维度表,把“最冷月平均”写死,防止 LLM 把哈尔滨 0℃ 当“南方湿冷”。
- 向量层:用 88 条 Markdown 语料(羽绒服、风衣、短袖…)跑 SentenceTransformer,缓存到本地 Chroma,LLM 可动态检索 Top3 相似描述,补充推荐理由。
3. LLM 提示模板:三锁机制
- 锁格式:Pydantic 类,字段 enum 化(top_layer, bottom_layer, shoes, accessories, reason)。
- 锁上下文:把“城市、性别、活动场景”用 System Message 注入,减少歧义。
- 锁温度:temperature=0,保证幂等,同一请求 100 次结果字节级一致。
模板片段(已脱敏):
You are a context-aware stylist. Given the weather below, output valid JSON that satisfies the schema. Do not hallucinate items outside the enum. Weather: {weather_json} UserProfile: {profile}代码:端到端可运行
以下单文件可直接python main.py 广州 male campus,返回推荐 JSON。
#!/usr/bin/env python3 """ Weather-based outfit recommender Run: python main.py <city> <gender> <scene> """ import asyncio, os, json, time from typing import Literal from aiohttp import ClientSession from pydantic import BaseModel, Field from enum import Enum from langchain.chat_models import ChatOpenAI from langchain.output_parsers import PydanticOutputParser from langchain.schema import HumanMessage, SystemMessage # ------------------ 配置 ------------------ OPENWEATHER_KEY = os.getenv("OWM_KEY") OPENAI_API_KEY = os.getenv("OPENAI_KEY") UNITS = "metric" LANG = "zh_cn" CACHE_TTL = 600 # 秒 # ------------------ 模型 ------------------ class TopLayerEnum(str, Enum): down_jacket = "down_jacket" wool_coat = "wool_cooat" hoodie = "hoodie" shirt = "shirt" tshirt = "tshirt" class BottomLayerEnum(str, Enum): jeans = "jeans" shorts = "shorts" sweatpants = "sweatpants" class ShoesEnum(str, Enum): sneakers = "sneakers" boots = "boots" sandals = "sandals" class Outfit(BaseModel): top_layer : TopLayerEnum bottom_layer: BottomLayerEnum shoes : ShoesEnum accessories : list[str] = Field(default_factory=list, description="e.g. scarf") reason : str = Field(..., description="一句话推荐理由,不超过30字") parser = PydanticOutputParser(pydantic_object=Outfit) # ------------------ 天气客户端 ------------------ class WeatherClient: def __init__(self, key: str): self.key = key self._cache = {} async def get_current(self, city: str) -> dict: if city in self._cache and time.time() - self._cache[city]["ts"] < CACHE_TTL: return self._cache[city]["data"] url = ( f"https://api.openweathermap.org/data/2.5/weather" f"?q={city}&appid={self.key}&units={UNITS}&lang={LANG}" ) async with ClientSession() as session: async with session.get(url) as resp: resp.raise_for_status() data = await resp.json() slim = { "temp": data["main"]["temp"], "weather_main": data["weather"][0]["main"], "humidity": data["main"]["humidity"], "wind_speed": data["wind"]["speed"], } self._cache[city] = {"ts": time.time(), "data": slim} return slim # ------------------ 推荐服务 ------------------ class OutfitService: def __init__(self): self.llm = ChatOpenAI( model="gpt-3.5-turbo", temperature=0, openai_api_key=OPENAI_API_KEY, max_tokens=200, ) self.weather = WeatherClient(OPENWEATHER_KEY) async def recommend(self, city: str, gender: Literal["male", "female"], scene: str) -> Outfit: w = await self.weather.get_current(city) system = SystemMessage(content=( "你是穿衣助手。根据天气与用户档案返回 JSON,严格遵守字段枚举值,禁止超范围。" f"用户档案:性别={gender},场景={scene}。" )) human = HumanMessage(content=( f"天气情况:{json.dumps(w, ensure_ascii=False)}" )) chain = parser | self.llm result = await chain.ainvoke([system, human]) return result # ------------------ CLI ------------------ if __name__ == "__main__": import sys, argparse parser = argparse.ArgumentParser() parser.add_argument("city") parser.add_argument("gender", choices=["male", "female"]) parser.add_argument("scene") args = parser.parse_args() svc = OutfitService() outfit = asyncio.run(svc.recommend(args.city, args.gender, args.scene)) print(outfit.json(ensure_ascii=False, indent=2))Clean Code 要点:
- 每个类单一职责,方便单元测试。
- 所有 I/O 异步,阻塞耗时 < 200ms。
- Pydantic 自动校验,非法字段直接抛 ValidationError,避免脏数据落到前端。
性能与安全:别让免费额度一夜归零
- 限流:OpenWeather 60 次/分钟,用 asyncio.BoundedSemaphore(60) 兜底。
- 缓存:Redis 存 <city, ts, json>,TTL 10 分钟,命中率 85%,节省 5000 次/天。
- 输出幂等:temperature=0 + 固定 seed,同一输入缓存可直接返回,LLM 层零额外成本。
- 异常熔断:连续 3 次超时自动降级到规则引擎,保证用户始终有衣可穿。
生产环境踩坑实录
- 冷启动延迟:LLM 首包 800ms,把“系统消息”预生成模板放内存,减少 30% 延迟。
- 幻觉案例:模型把“Clear”理解成“需要防晒衣”,但枚举里没有 sunscreen_coat,结果抛 ValidationError,前端白屏。解决:在 prompt 末尾加“若匹配不到请选最接近的 top_layer=shirt”,兜底成功。
- 多语言混用:中文 prompt + 英文 enum,模型偶尔返回“卫衣”不在枚举,被 pydantic 拒绝。统一全英文枚举,前端再做 i18n 映射。
- 隐私合规:城市字段做 md5 加盐哈希落日志,避免用户轨迹被反推。
可泛化的上下文感知框架
把“Weather”换成任意实时信号——空气质量、日历事件、地理位置、音乐节奏——只要:
- 用 Pydantic 把输出锁成业务对象;
- 用 LangChain 把外部 API 封装成 Tool;
- 用规则引擎做兜底 + 缓存;
你就能在 1 天内把“根据 PM2.5 推口罩”、“根据心率推歌单”等场景快速落地。毕设只是起点,真正的价值是把 LLM 当“策略编译器”,让需求变更变成改 schema 而非改代码。
下一步,你会把这套架构搬到哪些上下文感知推荐里?