上一篇我們畫了架構圖，描述一個場景：桌上架一台 Webcam 對著產品拍，使用者一邊看畫面一邊打字問「這是什麼型號？」、「規格是什麼？」，這篇就把它實作出來。

先快速回憶一下：為什麼要 VLM + CLIP 搭配著用

VLM（視覺語言模型）很會「看圖說故事」，你問它畫面裡有什麼，它答得頭頭是道。但如果你要它準確吐出一個型號，它常常會憑訓練時看過的知識瞎猜，因為它根本沒真的比對過你公司自己的產品庫。

所以這次的分工是：

• VLM 負責「懂」：理解使用者在問什麼、組織最後要講的話。

• CLIP 負責「認」：把鏡頭畫面轉成向量，拿去跟預先建好的產品照片向量庫做相似度比對，找出最像的那張，回推型號。

這其實就是視覺版的 RAG — 不是拿文字去向量資料庫找相似段落，而是拿「這張照片」去找「長得最像的產品照片」。認出型號後，再回頭查公司的 SPEC 資料，組成最終答案。

專案資料夾長怎樣

project/
├── client.py             # 前台 Flask 服務
├── server.py             # WebSocket Server
├── vlm.py                # VLM 主程式（整合 OpenCV + LangChain + Ollama）
├── mjpg-server.py        # Webcam 串流伺服器
├── config.ini            # 所有連線設定集中在這
├── requirements.txt
├── model/
│   ├── productModel2.py  # 公司產品 SPEC 資料（base64）
│   └── promptModel.py    # Prompt 樣板
├── module/
│   └── ClipSearch.py     # CLIP 圖搜圖引擎
├── template/
│   └── frontend.html     # 前台網頁（client.py 用 render_template 讀它）
├── data/
│   └── products/         # 產品參考圖庫，CLIP 建索引用
└── tmp/                  # process_frame() 暫存截圖用

先講一下 config.ini：這幾支程式幾乎每一支都會讀它，把 Redis、WebSocket、Ollama、攝影機參數全部集中管理，不寫死在程式碼裡。實際跑起來前，至少要準備好這些欄位：

[server]
WEBSOCKET_HOST = 0.0.0.0
WEBSOCKET_PORT = 8765
REDIS_HOST = localhost
REDIS_PORT = 6379

[client]
CLIENT_HOST = 0.0.0.0
CLIENT_PORT = 5000
CLIENT_DEBUG = False

[vlm]
MJPG_STREAMER_URL = http://127.0.0.1:8887/mjpeg
OLLAMA_MODEL = llama3.2-vision:latest
OLLAMA_BASE_URL = http://0.0.0.0:11434
OPENCV_CAP_PROP_FRAME_WIDTH = 1920
OPENCV_CAP_PROP_FRAME_HEIGHT = 1080
OPENCV_CAP_PROP_FPS = 30
SIMILARITY_THRESHOLD = 0.85

再來是套件清單，CLIP 是直接裝 OpenAI 官方 repo：

langchain
langchain-community
langchain_ollama
opencv-python
pillow
termcolor
websockets
redis
flask
jsonify
flask-cors

# CLIP
torch
numpy
tqdm
git+https://github.com/openai/CLIP.git

接下來，就照著資料流的方向（影像來源 → 資料層 → CLIP → VLM 大腦 → WebSocket → 前台），一支一支拆開來看。

mjpg-server.py：幫 Webcam 開一個「水龍頭」

import cv2
from mjpeg_streamer import MjpegServer, Stream

cap = cv2.VideoCapture(0)

# mjpeg 為圖片網址 / size 尺寸 / quality 品質 / fps 幀率
stream = Stream("mjpeg", size=(1920, 1080), quality=100, fps=60)

server = MjpegServer("127.0.0.1", 8887)
server.add_stream(stream)
server.start()

while True:
    ret, frame = cap.read()
    if not ret:
        break
    else:
        stream.set_frame(frame)

這支最單純，就是拿 OpenCV 讀 Webcam，然後用 mjpeg_streamer 把畫面用 MJPG 格式廣播出去，開在 127.0.0.1:8887。它的角色就是架構圖裡的「Web Streaming Server」：一份影像來源，兩邊拿去用 — 前台瀏覽器可以直接嵌 <img> 顯示即時畫面，vlm.py 也可以用 OpenCV 去讀這個串流網址拿「最新一張」。把「串流轉發」和「單張截圖給 AI 用」分開處理，是效能上的實際考量：串流本身要連續，但 AI 不需要每一幀都算。

productModel2.py：公司自己的產品規格庫

這支是架構圖裡「資料層」的其中一塊：公司產品 SPEC JSON 資料。內容其實就是一個大字典，key 是型號，value 裡放名稱、一張參考圖（base64 編碼）、還有規格表：

productData = {

    "XB860G2": {
        "name": "XB860G2",
        "image": "data:image/png;base64,iVBORw0KGgo...(略，實際檔案是完整的 base64 字串)",
        "specs": {
            "Model Name": "XB860G2",
            "Product Type": "XPC Slim",
            "CPU": "Intel® Arrow lake-S LGA1851 socket Core Ultra 200 series 65W processor",
            "GPU": "Intel® UHD Graphics series",
            "Memory": "2 x DDR5 5600 MT/s (Max. 96GB)",
            "Storage": "SATA 3.0 6Gb/s",
            "Expansion slots": "3 x M.2 2280, M Key , 1 x M.2 2230, E Key",
            "Video output": "2 x HDMI, 1 x DisplayPort, support 3 displays",
            "I/O interface": "2 x USB 2.0 , 4 x USB 3.2 Gen1 , 2 x USB 3.2 Gen2",
            "LAN": "1 x Intel® 1G LAN , 1 x Intel® 2.5G LAN",
            "Optional Accessory": "WLN-M, WWN04"
        }
    },

    # ... 其餘產品依此類推
}

這份資料本身沒有任何邏輯，純粹是「素材」— 它會被 client.py 用 from model.productModel2 import productData 整包 import 進去，直接丟給前端 render 成 JSON，讓前台在拿到型號之後，可以直接從瀏覽器端的 JS 查出對應規格顯示出來，完全不用再打一次後端 API。這也呼應了上一篇提到的設計：離線先把知識庫準備好，查詢時只做查表，不用臨時現算。

ClipSearch.py：CLIP 圖搜圖引擎

這支是整個系統「精準辨識」的核心，做的事情很簡單粗暴：把圖轉向量，去比對誰最像。

import os, torch, clip
from PIL import Image
import numpy as np
from tqdm import tqdm

class ImageSearch:
    def __init__(self, image_dir):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model, self.preprocess = clip.load("ViT-B/32", device=self.device)
        self.image_dir = image_dir
        self.image_paths = []
        self.image_features = []
        self.index_built = False

    def build_index(self):
        """建立圖片庫的特徵索引"""
        for root, _, files in os.walk(self.image_dir):
            for file in files:
                if file.lower().endswith(('.png', '.jpg', '.jpeg', '.webp', '.bmp')):
                    self.image_paths.append(os.path.join(root, file))

        with torch.no_grad():
            for image_path in tqdm(self.image_paths):
                image = self.preprocess(Image.open(image_path).convert("RGB")).unsqueeze(0).to(self.device)
                image_feature = self.model.encode_image(image)
                image_feature /= image_feature.norm(dim=-1, keepdim=True)  # 標準化
                self.image_features.append(image_feature.cpu().numpy())

        self.image_features = np.vstack(self.image_features)
        self.index_built = True

    def search(self, query_image_path, top_k=1):
        """搜索與輸入圖片最相似的圖片"""
        image = self.preprocess(Image.open(query_image_path).convert("RGB")).unsqueeze(0).to(self.device)

        with torch.no_grad():
            query_features = self.model.encode_image(image)
            query_features /= query_features.norm(dim=-1, keepdim=True)
            query_features = query_features.cpu().numpy()

        # 標準化向量做內積，就等於算 cosine similarity
        similarities = np.dot(query_features, self.image_features.T)[0]
        top_indices = similarities.argsort()[::-1][:top_k]

        return [(float(similarities[idx]), self.image_paths[idx]) for idx in top_indices]

重點在 build_index()：啟動時把 data/products/ 底下所有圖片，用 CLIP 的 ViT-B/32 模型轉成向量，順便做 L2 標準化（除以自己的 norm），這樣之後兩個向量做內積 np.dot(...)，結果就直接等於 cosine similarity，不用再另外算一次。這步是離線做一次就好，跟一般文字 RAG 建向量索引的邏輯一模一樣。

search() 則是查詢時的動作：拿到一張新的查詢圖片，轉成向量，跟索引庫裡所有向量做內積，分數排序後取最相似的 top_k 筆，回傳「相似度＋圖片路徑」。因為圖片路徑通常會是 data/products/型號/xxx.jpg 這種結構，所以路徑往前推就能拿到型號。

vlm.py：整個系統的大腦

這支負責把「抓畫面」、「讀提問」、「問 VLM」、「讀 CLIP 結果」、「組答案」全部串起來：

初始化 VLM

from langchain_ollama import ChatOllama
from langchain.schema import HumanMessage

llm = ChatOllama(
    model=config.get('vlm', 'OLLAMA_MODEL'),      # 例如 llama3.2-vision:latest
    temperature=0,
    disable_streaming=False,
    num_gpu=100,
    base_url=config.get('vlm', 'OLLAMA_BASE_URL'), # 地端 Ollama 服務位置
    top_k=64,
    top_p=0.95,
    min_p=0.0
)

就是透過 langchain_ollama 接上地端跑的 Ollama，VLM 可以換成 llama3.2-vision 系列或任何支援視覺的模型，base_url 指向你自己機器上跑 Ollama 的 IP。

影像來源：VideoSource

class VideoSource:
    def __init__(self, video_input: str):
        self.video_input = video_input
        self.eos = False

        if video_input.startswith('/dev/video'):
            self.cap = cv2.VideoCapture(int(video_input.replace('/dev/video', '')))
        elif video_input.startswith('http://') or video_input.startswith('https://'):
            # 接 mjpg-server.py 的串流網址
            self.cap = cv2.VideoCapture(str(video_input))
        elif '*' in video_input:
            # 也支援吃一批圖檔測試
            self.image_files = sorted(glob.glob(video_input))
        else:
            raise ValueError(f"Unsupported video input: {video_input}")

    def capture(self):
        # 依來源類型讀一張畫面回傳，讀不到就標記 self.eos = True
        ...

這個類別做了一個變通的彈性：video_input 可以是本機攝影機（/dev/video0）、也可以是 mjpg-server.py 開出來的網址（http://127.0.0.1:8887/mjpeg），甚至可以直接餵一批圖片路徑（*.jpg）方便離線測試，不用真的接攝影機。

核心：VisionChat.process_frame()

def process_frame(self, frame):
    # 暫存這一幀畫面成圖檔
    image_path = "tmp/temp.jpg"
    pil_image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
    pil_image.save(image_path)

    # 從 Redis 拿使用者最新提問
    self.prompt = redis_client.get('prompt').decode('utf-8')
    self.prompt = str(promptTemplate['llama']).replace('{prompt}', self.prompt)  # 套用 Prompt 樣板

    # 圖片轉 base64，跟文字一起包成訊息丟給 VLM
    with open(image_path, "rb") as f:
        image_b64 = base64.b64encode(f.read()).decode()

    message = HumanMessage(content=[
        {"type": "text", "text": self.prompt},
        {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}},
    ])
    response = self.llm.invoke([message])

    # 讀 CLIP 背景比對結果（相似度、型號），而不是同步呼叫 CLIP
    similarity = float(redis_client.get('clip_similarity').decode('utf-8'))
    content = str(redis_client.get('clip_content').decode('utf-8')).strip()

    # 相似度過門檻才把「精準辨識結果」蓋過 VLM 自己的猜測
    if similarity >= config.getfloat('vlm', 'SIMILARITY_THRESHOLD'):
        result = "Product|" + content + '|' + str(similarity) + '|' + f"This is our product {content}"
    else:
        result = str(response.content)
    # 寫回 Redis，交給 server.py 廣播出去
    redis_client.set('response', result)
    return frame

這段就是整個功能的核心：VLM 的回答（response.content）跟 CLIP 的比對結果（similarity / content）是兩條線，只有當 CLIP 的相似度過了門檻值，才會用「精準辨識出來的型號」去蓋掉 VLM 自己天馬行空的猜測。這正好對應到最開始講的分工：VLM 負責「懂使用者在問什麼、把話講好」，CLIP 負責「認出正確答案」，兩者各司其職，最後由相似度門檻決定聽誰的。

主程式：兩條執行緒各跑各的

def capture_process():
    # 持續讀取 Webcam / 串流最新畫面，更新到 shared_state.frame

def process_frames():
    while not shared_state.exit_flag:
        current_frame = shared_state.frame.copy()
        vision_chat.process_frame(current_frame)
        # 睡多久由 Redis 的 sleep_time 決定，加快反應
        time.sleep(float(redis_client.get('sleep_time')...))

capture_thread = threading.Thread(target=capture_process, daemon=True)
process_thread = threading.Thread(target=process_frames, daemon=True)

這裡用兩條 daemon thread 分工：一條專門「抓畫面」，一條專門「跑 VLM 推論」，中間用 shared_state（加鎖）傳最新一幀。這正好呼應了上一篇提到的「串流跟推論要分開拿資料」— 抓畫面要盡量快、不能被 VLM 推論卡住，所以拆成兩條執行緒平行處理。

server.py：WebSocket Server，Redis 的擴音器

class WebSocketRedisServer:
    async def redis_listener(self):
        pubsub = redis_client.pubsub()
        await pubsub.subscribe("__keyspace@0__:prompt", "__keyspace@0__:response")

        async for message in pubsub.listen():
            if message["type"] == "message":
                key = message["channel"].split(":")[-1]
                value = await redis_client.get(key)
                data = {"key": key, "value": value}

                if self.clients:
                    await asyncio.gather(*[c.send(json.dumps(data)) for c in self.clients])

    async def handler(self, websocket):
        await self.register(websocket)

        # 新連線先送一份目前的 prompt / response 當作初始畫面
        await websocket.send(json.dumps({
            "type": "initial_data",
            "data": {"prompt": await redis_client.get('prompt'),
                      "response": await redis_client.get('response')}
        }))
        async for message in websocket:
            pass  # 這條連線不接受前端傳訊息，純推播用

還有一段容易漏掉但很關鍵的設定：

async def configure_redis_notifications(host, port, db):
    redis_client = redis.Redis(host=host, port=port, db=db)
    await redis_client.config_set('notify-keyspace-events', 'KEA')

Redis 預設不會主動通知「某個 key 被改了」，要手動打開 notify-keyspace-events（KEA 代表監聽鍵空間事件 + 所有事件類型），這支程式一啟動就會先幫你設好。設好之後，prompt 或 response 這兩個 key 只要一被寫入，Redis 就會發布一個 __keyspace@0__:prompt 這樣的事件，redis_listener() 訂閱到之後立刻把最新值廣播給所有連線的前台。這就是架構圖裡說的「WebSocket Server 靠 WebSocket 做雙向推播，不用前端一直輪詢」的具體實作方式 — 本質上是拿 Redis 的 keyspace notification 當作觸發器。

client.py：前台的 Flask 服務

from model.productModel2 import productData

@app.route('/', methods=['GET', 'POST'])
def index():
    return render_template(
        'frontend.html',
        ws_host=config.get('server', 'WEBSOCKET_HOST'),
        ws_port=config.get('server', 'WEBSOCKET_PORT'),
        webcam_url=config.get('vlm', 'MJPG_STREAMER_URL'),
        product_data=json.dumps(productData, ensure_ascii=False)
    )

@app.route('/api/ask', methods=['POST'])
def ask():
    prompt = request.form.get("prompt")
    redis_client.set('prompt', prompt)   # 寫進 Redis，交給 vlm.py 去讀
    return jsonify({"result": True, "msg": "success", "data": {"prompt": prompt}})

這支的角色很單純：

• 首頁 / 把 WebSocket 位置、Webcam 串流網址、整包產品 SPEC 資料一次塞進網頁模板，讓前端 JS 有足夠資訊自己接 WebSocket、接 MJPG 串流、查規格表，後續完全不用再打 API 查資料。
• /api/ask 是使用者打字送出問題的入口，做的事情極簡 — 就是把 prompt 寫進 Redis 的 prompt 這個 key，然後直接回傳成功。

沒錯，這支完全沒有直接呼叫 VLM、也沒呼叫 CLIP。它只負責寫入 Redis，剩下的事情全部丟給 vlm.py 的背景執行緒去處理，處理完再透過 server.py 的 WebSocket 推回來。前台完全不需要知道 AI 處理花了多久、CLIP 有沒有比對成功，這就是上一篇強調的「cache + WebSocket 解耦前後端」，在程式碼裡真的落地的樣子。

小結：使用者問一句話，系統實際跑了什麼

對照上面六支程式，一次完整互動大概是這樣接力：

1. 使用者在前台打字送出「這是什麼型號？」→ 打到 client.py 的 /api/ask

2. client.py 把 prompt 寫進 Redis

3. vlm.py 的 process_frames 執行緒醒來，從 shared_state.frame 拿到目前最新一幀（來源是 mjpg-server.py 的串流，由 capture_process 執行緒持續更新）

4. 從 Redis 讀出剛剛那句 prompt，套上 Prompt 樣板，跟畫面一起丟給 Ollama 跑的 VLM

5. 同時從 Redis 讀出 CLIP 背景比對的 clip_similarity / clip_content（這兩個值是另一支跑 ClipSearch.py 的程式持續寫入的）

6. 相似度過門檻 → 採用 CLIP 精準辨識出來的型號；沒過門檻 → 採用 VLM 自己的回答

7. 結果寫回 Redis 的 response

8. server.py 偵測到 response 這個 key 被改了，透過 WebSocket 廣播給所有連線的前台

9. 前台收到後，即時更新畫面：目前影像、提問、VLM 回應，再依辨識出來的型號從 productData（productModel2.py）查出完整規格一起呈現