用Cursor和LangChain搞了个计算机视觉实战项目，踩坑全记录
作者背景：美团外卖4年Java后端，刚跳槽到一家做AI的创业公司两个月。日常靠VSCode续命，插件装得比代码还多。最近被领导安排搞计算机视觉相关的需求，硬着头皮啃了两个月，算是有点心得，写出来跟大家唠唠。
事情是这样的
上周五晚上十点半，我正准备关电脑走人，leader老张突然走过来拍了拍我肩膀："小陈啊，有个新需求想跟你聊聊。"
听到这话，我后背一凉。在美团干了四年，我太知道这种"聊聊"意味着什么了。果然，老张说公司新接了一个餐饮商户的AI巡检项目，需要用计算机视觉技术对商户后厨的监控画面做实时分析——识别厨师有没有戴厨师帽、有没有戴口罩、后厨有没有老鼠之类的。
"这不是CV的活儿吗？我写Java的。"我下意识想拒绝。
老张笑了笑："现在AI应用层开发哪分得那么细，你先把Demo搭出来，算法那边小刘会配合你。"
行吧，来新公司才两个月，总不能说不行。
技术选型：为什么选LangChain + Function Calling
说实话，刚接到需求的时候我是懵的。计算机视觉？我上一次接触还是大学毕设用OpenCV做了个人脸检测，那都是七八年前的事了。
但好在现在AI应用开发的工具链已经非常成熟了。我花了一个周末调研了一圈，最终确定了技术路线：
| 技术栈 | 选型 | 理由 |
|---|---|---|
| 视觉模型 | YOLOv8 | 轻量、速度快，适合实时检测场景 |
| 应用框架 | LangChain | 生态好，文档全，Function Calling支持完善 |
| 开发工具 | Cursor | AI辅助编程，写Python效率直接翻倍 |
| 大模型 | GPT-4o | 多模态能力强，可以直接分析图片 |
| 后端服务 | FastAPI | 轻量高性能，跟Python生态无缝衔接 |
为什么不用Spring Boot？因为CV这块Python生态太强了，YOLO、OpenCV、各种模型推理框架全是Python优先。我一个Java开发，硬着头皮也得学。
说到Cursor，这工具真的是我最近发现的生产力神器。我之前在美团一直用IDEA，但写Python还是Cursor舒服。装了GitHub Copilot、Python、Pylance、Ruff一堆插件，写代码的体验直接拉满。关键是它的Composer功能，可以直接用自然语言描述需求，帮你生成整个模块的代码，对于我这种Python不太熟的人来说简直是救星。
项目架构设计
整体架构其实不复杂，核心思路是：
摄像头视频流 -> 抽帧 -> YOLOv8目标检测 -> 检测结果结构化
-> LangChain Agent + Function Calling -> GPT-4o分析 -> 输出巡检报告
简单说就是：先用YOLO把画面里的目标检测出来（人、帽子、口罩、老鼠等），然后把检测结果通过Function Calling喂给大模型，让大模型做最终的合规判断和报告生成。
为什么不直接让GPT-4o看图片？因为实时性要求高，视频流每秒要处理25帧，直接调多模态API延迟扛不住，而且成本也太高。YOLO做初筛，大模型做决策，各司其职。
核心代码实现
第一步：YOLOv8目标检测
from ultralytics import YOLO
import cv2
import numpy as np
class KitchenDetector:
def __init__(self, model_path="yolov8n.pt"):
# 加载预训练模型，实际项目需要用自己的数据集fine-tune
self.model = YOLO(model_path)
# 定义我们关心的目标类别
self.target_classes = {
0: "person",
1: "chef_hat",
2: "mask",
3: "glove",
4: "rat"
}
def detect(self, frame: np.ndarray) -> list:
"""
对单帧图像进行检测
返回检测结果列表
"""
results = self.model(frame, conf=0.5, verbose=False)
detections = []
for result in results:
boxes = result.boxes
for box in boxes:
cls_id = int(box.cls[0])
conf = float(box.conf[0])
x1, y1, x2, y2 = box.xyxy[0].tolist()
detections.append({
"class": self.target_classes.get(cls_id, "unknown"),
"confidence": round(conf, 3),
"bbox": [round(x1, 1), round(y1, 1), round(x2, 1), round(y2, 1)]
})
return detections
def analyze_compliance(self, detections: list) -> dict:
"""
根据检测结果做初步合规分析
"""
persons = [d for d in detections if d["class"] == "person"]
chef_hats = [d for d in detections if d["class"] == "chef_hat"]
masks = [d for d in detections if d["class"] == "mask"]
rats = [d for d in detections if d["class"] == "rat"]
return {
"person_count": len(persons),
"chef_hat_count": len(chef_hats),
"mask_count": len(masks),
"rat_detected": len(rats) > 0,
"hat_compliance_rate": round(len(chef_hats) / max(len(persons), 1), 2),
"mask_compliance_rate": round(len(masks) / max(len(persons), 1), 2),
"timestamp": cv2.getTickCount()
}
这里有个坑要提一下。YOLOv8默认的检测类别是COCO数据集的80类，里面并没有"厨师帽"和"口罩"这些类别。我们需要自己标注数据集然后fine-tune。这个活儿是算法同事小刘干的，他用了大概2000张后厨场景的图片做标注，训练了大概20个epoch，效果就挺不错了。
说到模型训练，分享几个心得：
- 数据质量比数量重要。一开始我们标了5000张图，但里面很多是重复场景，后来精简到2000张精选数据，效果反而更好。
- 数据增强很关键。后厨的光线变化很大，我们做了亮度、对比度、旋转等增强，模型的泛化能力提升明显。
- 小目标检测是个难题。老鼠在画面里占比很小，一开始漏检率很高。后来用了SAHI（Slicing Aided Hyper Inference）做切片推理，效果好了很多。
第二步：LangChain + Function Calling
这是整个项目最核心的部分。我们需要把YOLO的检测结构交给大模型做最终判断。这里用到了LangChain的Function Calling能力。
import json
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from typing import List, Optional
# 定义Function Calling的工具函数
@tool
def check_health_compliance(
person_count: int,
chef_hat_count: int,
mask_count: int,
rat_detected: bool,
hat_compliance_rate: float,
mask_compliance_rate: float
) -> str:
"""
检查后厨卫生合规情况。
根据检测到的厨师数量、戴帽数量、戴口罩数量、是否检测到老鼠等数据，
判断后厨是否符合卫生规范。
Args:
person_count: 检测到的厨师/工作人员数量
chef_hat_count: 检测到戴厨师帽的数量
mask_count: 检测到戴口罩的数量
rat_detected: 是否检测到老鼠
hat_compliance_rate: 戴帽合规率 (0-1)
mask_compliance_rate: 戴口罩合规率 (0-1)
"""
issues = []
if hat_compliance_rate < 0.8:
issues.append(f"厨师帽佩戴合规率仅{hat_compliance_rate*100:.0f}%，低于80%标准")
if mask_compliance_rate < 0.9:
issues.append(f"口罩佩戴合规率仅{mask_compliance_rate*100:.0f}%，低于90%标准")
if rat_detected:
issues.append("检测到老鼠！严重卫生隐患！")
if person_count > 0 and chef_hat_count == 0:
issues.append("所有工作人员均未佩戴厨师帽")
if not issues:
return "合规检查通过，后厨卫生状况良好。"
return f"发现{len(issues)}项违规：\n" + "\n".join([f"- {issue}" for issue in issues])
@tool
def generate_incident_report(
violation_type: str,
severity: str,
description: str,
suggestion: str
) -> str:
"""
生成违规事件报告。当发现严重违规时调用此工具。
Args:
violation_type: 违规类型，如"未戴厨师帽"、"发现老鼠"等
severity: 严重程度，可选值：low/medium/high/critical
description: 违规情况描述
suggestion: 整改建议
"""
report = {
"violation_type": violation_type,
"severity": severity,
"description": description,
"suggestion": suggestion,
"status": "pending_review"
}
# 实际项目中这里会写入数据库
return json.dumps(report, ensure_ascii=False)
# 构建Agent
def build_compliance_agent():
llm = ChatOpenAI(
model="gpt-4o",
temperature=0.1, # 低温度，保证输出稳定性
max_tokens=1000
)
tools = [check_health_compliance, generate_incident_report]
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个后厨卫生巡检AI助手。你的职责是：
1. 根据目标检测的结果数据，判断后厨是否符合卫生规范
2. 对违规行为进行分类和严重程度评估
3. 必要时生成违规事件报告
4. 给出专业、客观的巡检结论
规范要求：
- 厨师帽佩戴率 >= 80% 为合格
- 口罩佩戴率 >= 90% 为合格
- 检测到老鼠为严重违规
- 检测到人但未检测到任何防护装备为中度违规"""),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
handle_parsing_errors=True,
max_iterations=3
)
return agent_executor
第三步：串联整个Pipeline
from fastapi import FastAPI, WebSocket
from fastapi.responses import JSONResponse
import asyncio
import base64
app = FastAPI(title="后厨AI巡检系统")
agent_executor = build_compliance_agent()
detector = KitchenDetector(model_path="runs/detect/train/weights/best.pt")
@app.post("/api/v1/inspect")
async def inspect_frame(image_base64: str):
"""
接收单帧图像，返回巡检结果
"""
# 解码图像
img_bytes = base64.b64decode(image_base64)
nparr = np.frombuffer(img_bytes, np.uint8)
frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
# YOLO检测
detections = detector.detect(frame)
compliance = detector.analyze_compliance(detections)
# 构造prompt给Agent
input_text = f"""当前画面检测结果如下：
- 检测到工作人员：{compliance['person_count']}人
- 戴厨师帽：{compliance['chef_hat_count']}人
- 戴口罩：{compliance['mask_count']}人
- 戴帽合规率：{compliance['hat_compliance_rate']*100:.0f}%
- 口罩合规率：{compliance['mask_compliance_rate']*100:.0f}%
- 是否检测到老鼠：{'是' if compliance['rat_detected'] else '否'}
请进行合规检查，如有违规请生成事件报告。"""
# 调用Agent
result = agent_executor.invoke({"input": input_text})
return JSONResponse(content={
"detections": detections,
"compliance": compliance,
"ai_analysis": result["output"]
})
@app.websocket("/ws/realtime")
async def realtime_inspect(websocket: WebSocket):
"""
WebSocket实时巡检，处理视频流
"""
await websocket.accept()
frame_count = 0
try:
while True:
data = await websocket.receive_bytes()
frame = cv2.imdecode(np.frombuffer(data, np.uint8), cv2.IMREAD_COLOR)
# 每5帧分析一次，降低计算压力
frame_count += 1
if frame_count % 5 != 0:
continue
detections = detector.detect(frame)
compliance = detector.analyze_compliance(detections)
# 只在有违规时才调用大模型，节省成本
if compliance['hat_compliance_rate'] < 0.8 or \
compliance['mask_compliance_rate'] < 0.9 or \
compliance['rat_detected']:
input_text = f"检测结果：{json.dumps(compliance, ensure_ascii=False)}，请分析并生成报告。"
result = agent_executor.invoke({"input": input_text})
await websocket.send_json({
"compliance": compliance,
"ai_analysis": result["output"],
"alert": True
})
else:
await websocket.send_json({
"compliance": compliance,
"alert": False
})
except Exception as e:
print(f"WebSocket error: {e}")
finally:
await websocket.close()
踩过的坑，说多了都是泪
坑一：Function Calling的参数格式问题
刚开始写的时候，大模型调用check_health_compliance老是传错参数。比如把hat_compliance_rate传成了字符串"85%"，而不是浮点数0.85。排查了半天，最后发现是prompt里给的示例数据格式不统一。
解决方案是在tool的docstring里把参数类型和格式写得非常明确，并且在system prompt里强调数据格式要求。LangChain这块的文档说实话写得不够细，很多细节得自己试。
# 改进后的tool定义，参数描述更精确
@tool
def check_health_compliance(
person_count: int, # 整数，工作人员数量，如：3
chef_hat_count: int, # 整数，戴帽数量，如：2
mask_count: int, # 整数，戴口罩数量，如：3
rat_detected: bool, # 布尔值，是否有老鼠，如：false
hat_compliance_rate: float, # 浮点数，0到1之间，如：0.67
mask_compliance_rate: float # 浮点数，0到1之间，如：1.0
) -> str:
坑二：YOLO检测的误检问题
后厨场景太复杂了。一开始模型把白色的塑料袋识别成了厨师帽，把黑色的调料包识别成了老鼠。当时看到检测结果我人都傻了，这要上线了不得被商户骂死。
解决办法：
- 增加负样本训练，专门收集一些容易被误检的物品图片
- 提高置信度阈值，从0.5调到0.65
- 加了一个后处理逻辑，根据bbox的面积和位置做过滤（比如老鼠不可能出现在天花板上）
def post_process(self, detections: list, frame_shape: tuple) -> list:
"""后处理过滤明显不合理的检测结果"""
h, w = frame_shape[:2]
filtered = []
for det in detections:
x1, y1, x2, y2 = det["bbox"]
area = (x2 - x1) * (y2 - y1)
relative_area = area / (w * h)
# 老鼠不可能出现在画面最上方（天花板区域）
if det["class"] == "rat" and y1 < h * 0.3:
continue
# 太小的检测框大概率是误检
if relative_area < 0.001:
continue
filtered.append(det)
return filtered
坑三：大模型调用的延迟和成本
这是最头疼的问题。GPT-4o的API调用一次大概要1-2秒，如果每帧都调，别说实时性了，光API费用就能把公司搞破产。
我的优化策略：
- 抽帧处理：25fps的视频流，只每5帧做一次YOLO检测，相当于5fps
- 按需调用大模型：只有YOLO检测到疑似违规时才调用Agent，合规的画面直接跳过
- 结果缓存：对同一个摄像头的连续检测结果做滑动窗口平均，避免单帧误检触发告警
- 批量处理：把多帧的检测结果合并后一次性给大模型分析
class ResultBuffer:
"""滑动窗口结果缓冲，避免单帧误判"""
def __init__(self, window_size=10):
self.buffer = []
self.window_size = window_size
def add(self, compliance: dict):
self.buffer.append(compliance)
if len(self.buffer) > self.window_size:
self.buffer.pop(0)
def get_smoothed_result(self) -> dict:
if not self.buffer:
return {}
avg_hat_rate = sum(c['hat_compliance_rate'] for c in self.buffer) / len(self.buffer)
avg_mask_rate = sum(c['mask_compliance_rate'] for c in self.buffer) / len(self.buffer)
has_rat = any(c['rat_detected'] for c in self.buffer)
return {
"hat_compliance_rate": round(avg_hat_rate, 2),
"mask_compliance_rate": round(avg_mask_rate, 2),
"rat_detected": has_rat,
"sample_count": len(self.buffer)
}
def should_alert(self) -> bool:
"""是否需要触发告警（调用大模型）"""
result = self.get_smoothed_result()
if not result:
return False
return (result['hat_compliance_rate'] < 0.8 or
result['mask_compliance_rate'] < 0.9 or
result['rat_detected'])
优化之后，大模型的调用频率降低了大概80%，API费用从预估的每月3万降到了5000左右，延迟也从平均2秒降到了只在违规时才有感知。
坑四：Cursor用得太爽导致代码质量下降
这个得自我批评一下。Cursor的Composer功能太好用了，我经常一句话就让它生成一整个模块的代码。结果就是有些生成的代码我没仔细review，上线前测试的时候发现了一个空指针的bug——Python里是NoneType error。
教训是：AI辅助编程确实能提效，但生成的代码一定要认真review。特别是边界条件、异常处理这些地方，AI经常会忽略。我现在养成了一个习惯，Cursor生成的代码，每一行都要过一遍，尤其是try-catch和None检查。
模型训练的一些心得
虽然模型训练主要是算法同事小刘在做，但我也跟着学了不少。分享几个关键点：
数据集准备
我们最终用了约2000张图片，分布如下：
| 场景 | 图片数量 | 标注目标 |
|---|---|---|
| 正常后厨（合规） | 600 | person, chef_hat, mask |
| 违规后厨（未戴帽） | 400 | person |
| 违规后厨（未戴口罩） | 300 | person |
| 有老鼠的场景 | 200 | rat |
| 复杂背景/干扰物 | 300 | 各种负样本 |
| 不同光照条件 | 200 | 混合 |
标注工具用的是LabelImg，说实话这工具界面挺古老的，但胜在稳定。小刘一个人标了将近一周，眼睛都快瞎了。后来我们引入了半自动标注——先用预训练模型跑一遍，人工再修正，效率提升了大概3倍。
训练参数
# train_config.yaml
task: detect
mode: train
model: yolov8n.pt # 使用nano版本，追求推理速度
data: kitchen_dataset.yaml
epochs: 30
imgsz: 640
batch: 16
device: 0 # 单卡A100
# 数据增强
hsv_h: 0.015 # 色调增强
hsv_s: 0.7 # 饱和度增强
hsv_v: 0.4 # 亮度增强
degrees: 10 # 旋转增强
translate: 0.1 # 平移增强
scale: 0.5 # 缩放增强
fliplr: 0.5 # 水平翻转
mosaic: 1.0 # Mosaic增强
# 优化器
optimizer: AdamW
lr0: 0.001
lrf: 0.01
最终模型在测试集上的表现：
| 指标 | 数值 |
|---|---|
| mAP@0.5 | 0.89 |
| mAP@0.5:0.95 | 0.62 |
| 推理速度（A100） | 3.2ms/帧 |
| 推理速度（T4） | 8.5ms/帧 |
| 模型大小 | 6.2MB |
mAP@0.5:0.95只有0.62，说实话不算特别高，主要是老鼠这个类别的检测精度拉低了均值。但实际业务场景中，我们对精度的要求没有这么苛刻，宁可漏检也不能误报太多（误报会导致商户投诉），所以这个指标是可接受的。
上线后的效果
项目上线两周了，目前接入了大概50家商户的摄像头。说实话效果超出预期：
- 日均处理视频帧数：约200万帧
- 违规检出率：92%（人工抽检对比）
- 误报率：约8%（还在持续优化）
- 平均响应延迟：280ms（从抽帧到返回结果）
- 日均API调用次数：约3000次（优化后）
最让我开心的是，上线第一周就抓到了一家商户后厨有老鼠。商户老板一开始还不信，看了我们抓拍的截图之后沉默了。据说当天就请了专业的消杀公司。
一些思考
关于AI应用开发
做了这个项目之后，我对AI应用开发有了一些新的理解。跟传统的后端开发相比，AI应用开发有几个显著的不同：
- 不确定性是常态。传统代码是确定性的，输入A一定输出B。但AI模型的输出有概率性，你必须设计好兜底策略。
- 数据比算法重要。我们花了很多时间在数据清洗和标注上，这比调模型参数带来的收益大得多。
- 工程化能力是壁垒。模型大家都能用，但怎么把模型稳定、高效地部署到生产环境，这才是真正的技术含量。
关于Cursor和AI辅助编程
Cursor确实好用，但我发现它更适合写"胶水代码"——比如API对接、数据转换、CRUD这些模式化的代码。对于核心业务逻辑，还是得自己写，因为AI不理解你的业务上下文。
另外，Cursor生成的代码风格有时候不太统一，同一个项目里可能会出现几种不同的代码风格。我后来在.cursorrules文件里加了一些规范约束，效果好了很多：
# .cursorrules
- 所有函数必须有docstring和类型注解
- 异常处理不能吞掉异常，至少要log
- 使用pathlib而不是os.path
- 配置项统一放在config.py，不要硬编码
- 所有API接口必须有参数校验
关于Java开发转AI方向
作为一个写了四年Java的人，转做AI应用开发确实有不适应的地方。Python的生态虽然丰富，但工程化方面跟Java比还是差了不少。没有强类型、没有完善的依赖管理（pip真的不如Maven）、没有好用的调试工具……
但换个角度想，这也是一种成长。现在AI应用开发越来越火，懂业务、懂工程、又懂AI的人才其实很稀缺。Java的功底让我在系统设计、性能优化、高并发处理这些方面有明显优势。比如这个项目的WebSocket服务，一开始Python原生的实现扛不住并发，我直接用Java的思路重写了连接池和消息队列，性能提升了5倍。
写在最后
来新公司两个月，这个项目算是我交出的第一份答卷。虽然还有很多需要优化的地方，但总算跑通了。
最大的感受是：技术这东西，真的不能给自己设限。我以前觉得自己就是写Java的，CV、AI这些离我很远。但真正做起来发现，底层的方法论是相通的——系统设计、性能优化、问题排查，这些能力在任何领域都有用。
好了，不说了，老张又过来了，估计又有新需求。希望这次不是"聊聊"。
如果你也在做AI应用开发相关的事情，欢迎交流。特别是LangChain的Function Calling这块，坑真的不少，有机会再单独写一篇。
P.S. 我的VSCode插件列表，有兴趣的可以看看，真的很好用：Python、Pylance、Ruff、GitHub Copilot、GitLens、Docker、Remote-SSH、Thunder Client、Error Lens、Todo Tree。不多，就十几个而已。
评论 0