《实用笔记》：我如何将照片库转变为自主AI智能体
《实用笔记》操作指南：我如何将照片库转变为自主AI智能体——以及为采用该模式的团队提供的合同、检查清单和可直接插入的代码模块。
本指南将逐步讲解如何从原始材料构建出一个可运行的系统，内容涵盖《我如何将照片库转变为自主AI智能体——完整指南》。重点在于可操作的步骤、明确的检查点，以及可直接放入代码仓库的代码，无需猜测其用途。 在概览阶段，应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 建议使用小型、可测试的单元而非庞大的脚本。当某个步骤失败时，故障应能指向单一的责任模块，而非复杂的流程链。
引言
在完成入门阶段时，首先写下合同条款：所需的输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功检测标准，并杜绝无声的半完成状态。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
为什么传统的照片搜索会出问题
在处理“为何选择传统照片搜索”这一阶段时，首先需列出相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前明确成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次对同一次大语言模型调用收费。
相册式方法在现实中行不通
在运用“专辑式分解法”处理任务时，首先需明确相关约定：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持透明可追溯。 应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一处，以便操作人员无需查看整个系统结构即可进行审计。 在耗时较长的步骤之后设置检查点。当操作人员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型接口。 在运用“专辑式分解法”处理任务时，首先需明确相关约定：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持透明可追溯。 相较于庞大的脚本，更应优先使用小型且易于测试的单元。当某个步骤失败时，故障原因应能明确指向某个具体的功能模块，而非整个复杂的流程。
仅靠关键词搜索效果有限
将关键词搜索视为可度量的基准时，它在阶段性工作中效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默地部分完成任务。 保持图表状态简洁且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段，还会在任务中断后导致无法继续。
云图库搜索：有用，但需付出代价
将 Cloud gallery 搜索功能视为可度量的对象，其效果会最佳。在扩大搜索范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图表状态简洁且类型明确，嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在出现中断后导致数据恢复失败。
真正的差距：缺乏语义理解
将每个阶段视为可测量的对象才是最佳处理方式。在扩大范围之前，先记录一份完美的操作日志、一个故障案例以及回滚说明。 配置应与应用程序代码分开存放。环境文件、密钥存储和功能开关应集中于一处，以便操作人员无需查看整个系统结构即可进行审计。 保持系统状态的结构化与类型化。嵌套的数据结构会掩盖具体是哪个节点修改了哪个字段，还会导致中断后无法继续执行。 将每个阶段视为可测量的对象才是最佳处理方式。在扩大范围之前，先记录一份完美的操作日志、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障原因应能明确指向某个具体的责任模块，而非复杂的流程链。
概念解析：简单易懂的语义搜索
在“概念语义搜索”阶段，应在修改代码之前明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。
语言在其中起什么作用？
在“语言从何而来”这一阶段，应在修改代码之前明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境转向共享环境时出现意外费用。对于会耗费资金或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。
选择合适的工具（以及为何耗时数周）
在“选择合适工具”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审计。 在网关处进行身份验证，在数据层进行重新授权。仅凭承载令牌并不足以界定租户边界。 在“选择合适工具”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相较于庞大的脚本，应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向单一责任模块，而非复杂的流程链。
通过FastEmbed实现的CLIP ViT-B/32——系统的“眼睛”
在处理CLIP ViT-B 32相关流程时，首先需明确规范：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续代码修改的规范性。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，定义成功判定标准，杜绝无声的半完成状态。 在调整提示词之前，先在固定问题集上测试召回率。仅仅更换提示词很难改善较差的检索效果。
from fastembed import ImageEmbeddingModel, TextEmbeddingModel
# Loaded once, reused forever
image_model = ImageEmbeddingModel.from_pretrained("Qdrant/clip-ViT-B-32-vision")
text_model = TextEmbeddingModel.from_pretrained("Qdrant/clip-ViT-B-32-text")
# After embedding:
image_vector = embed_image("sunset_photo.jpg") # shape: (512,)
query_vector = embed_text("beautiful sunset") # shape: (512,)
# Cosine similarity — just a dot product on normalised vectors
similarity = np.dot(image_vector, query_vector)
# If image is a sunset → similarity ≈ 0.85 (strong match)
# If image is food → similarity ≈ 0.15 (weak match)
Qdrant Edge——内存存储
在处理 Qdrant Edge 的内存阶段时，首先需记录下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改不会出错。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时的操作之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的 LLM 调用费用。
from qdrant_edge import(
Distance,
EdgeConfig,
EdgeShard,
EdgeVectorParams,
)
config = EdgeConfig(
vectors={
VECTOR_NAME: EdgeVectorParams(
size=EMBED_DIM,
distance=Distance.Cosine
)
}
)
_shard = EdgeShard.create(path=str(SHARD_DIR), config=config)
整合应用
在“整合所有组件”阶段，首先需明确合同规范：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 配置信息应与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置，以便操作员无需查看整个流程即可进行审计。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。 在“整合所有组件”阶段，首先需明确合同规范：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 相较于庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障应能指向单一的责任模块，而非复杂的流程链。
OpenClaw —— 核心引擎
将OpenClaw的脑力阶段视为可测量的表面来处理时，其效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的、不完整的处理结果。 保持图结构的状态简洁且具有明确类型。嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在出现中断后导致无法继续处理。
环境配置
将“环境搭建”阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份成功的测试案例、一个失败案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图表状态简洁且类型明确，嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在出现中断后导致无法继续处理。
步骤1：克隆仓库并搭建虚拟环境
将“第一步：克隆阶段”视为可度量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份理想状态下的日志、一个故障案例以及回滚说明。 应将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个架构就能进行审计。 要保持架构状态的简洁性与类型化。嵌套的数据结构会掩盖是哪个节点修改了哪个字段，还会导致在出现中断后无法继续执行。 将“第一步：克隆阶段”视为可度量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份理想状态下的日志、一个故障案例以及回滚说明。 相比庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向某个具体的责任模块，而非复杂的流程链。
# Clone the repo
git clone https://github.com/vatsala-singh/AI-Powered-Photo-Search-and-Tagging-Agent.git
cd AI-Powered-Photo-Search-and-Tagging-Agent
# Create and activate virtual environment
python -m venv .venv
source .venv/bin/activate # Mac/Linux
# .venv\Scripts\activate # Windows
第二步：安装依赖项
在第二步安装阶段时，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功检测标准，并拒绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。
pip install -r requirements.txt
第三步：了解项目结构
在第三步“理解阶段”中，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解这些成本可以避免在从演示环境过渡到共享环境时出现意外费用。对于那些会消耗资金或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。
AI-Powered-Photo-Search-and-Tagging-Agent/
│
├── main.py # FastAPI app + OpenClaw agent entry point
├── config.py # All configurable parameters in one place
├── requirements.txt
│
├── pipeline/
│ ├── embedder.py # CLIP embedding logic (image + text)
│ └── indexer.py # Batch photo processing and indexing
│
├── store/
│ └── qdrant_client.py # Qdrant Edge setup and collection management
│
├── tools/
│ ├── search.py # Semantic search tool
│ ├── tag.py # Zero-shot auto-tagging tool
│ ├── duplicates.py # Near-duplicate detection tool
│ └── albums.py # Smart album grouping tool
│
├── test/
│ ├── embedder_test.py
│ ├── indexer_test.py
│ ├── search_test.py
│ └── qdrant_edge_client_test.py
│
└── qdrant-edge-data/ # Auto-created at runtime
├── storage/ # Qdrant's internal shard data
├── models/ # Cached CLIP model weights
└── photos/ # Collection data
第四步：查看config.py
在第四步的“概览”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务流程的完整性。 在第四步的“概览”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 相较于庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向某个具体的责任模块，而非整个复杂的流程。
# config.py
CLIP_IMAGE_MODEL = "Qdrant/clip-ViT-B-32-vision"
CLIP_TEXT_MODEL = "Qdrant/clip-ViT-B-32-text"
EMBEDDING_DIM = 512 # CLIP ViT-B/32 output dimension
COLLECTION_NAME = "photos"
QDRANT_PATH = "./qdrant-edge-data"
BATCH_SIZE = 32 # Images per indexing batch
TOP_K = 10 # Default search results returned
TAG_THRESHOLD = 0.20 # Min similarity score for a tag to apply
DUPLICATE_THRESHOLD = 0.97 # Min similarity to flag as duplicate
第5步：启动服务器
在执行“启动服务器”这一步骤时，首先需列出相关约定：所需输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改始终符合约定。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝默许部分完成的情况。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
uvicorn main:app --reload --port 8000
第6步：验证设置
在执行第6步“验证阶段”时，首先记下相关要求：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。
# Quick sanity check - should return {"status": "ok"}
curl http://localhost:8000/health
curl http://localhost:8000/api/status
构建图像嵌入流程
在完成“构建图像嵌入”阶段时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，以便操作人员无需查看整个系统结构即可进行审计。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。 在完成“构建图像嵌入”阶段时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤出现故障时，故障原因应能指向单一责任模块，而非复杂的流程链。
嵌入器
将嵌入阶段视为可测量的界面时，其效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声无息的半完成状态。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。
# pipeline/embedder.py
from fastembed import ImageEmbeddingModel, TextEmbeddingModel
from PIL import Image
import numpy as np
from config import CLIP_IMAGE_MODEL, CLIP_TEXT_MODEL
# Load once, reuse for the lifetime of the process
# Models are large (~150MB each) — we never want to reload them per request
_image_model = ImageEmbeddingModel.from_pretrained(CLIP_IMAGE_MODEL)
_text_model = TextEmbeddingModel.from_pretrained(CLIP_TEXT_MODEL)
def embed_image(image_path: str) -> np.ndarray:
"""Convert an image file to a 512-d CLIP embedding."""
image = Image.open(image_path).convert("RGB")
embeddings = list(_image_model.embed([image]))
return np.array(embeddings[0]) # shape: (512,)
def embed_text(query: str) -> np.ndarray:
"""Convert a text string to a 512-d CLIP embedding."""
embeddings = list(_text_model.embed([query]))
return np.array(embeddings[0]) # shape: (512,)
索引器
将索引阶段视为可测量的对象时，其效果最佳。在扩大范围之前，先记录一个成功的示例、一个失败案例以及回滚说明。在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。需保持图结构简洁且类型明确，嵌套的数据块会掩盖具体是哪个节点修改了哪个字段，还会在出现中断后导致流程无法继续。
# pipeline/indexer.py
import os
import uuid
from pathlib import Path
from datetime import datetime
from PIL import Image
from pipeline.embedder import embed_image
from store.qdrant_client import get_shard
from tools.tag import generate_tags
from config import BATCH_SIZE
SUPPORTED_FORMATS = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tiff"}
def index_folder(folder_path: str) -> dict:
"""
Recursively index all images in a folder into Qdrant Edge.
Returns a summary: total found, indexed, skipped.
"""
folder = Path(folder_path)
shard = get_shard()
image_paths = [
p for p in folder.rglob("*")
if p.suffix.lower() in SUPPORTED_FORMATS
]
total = len(image_paths)
indexed = 0
skipped = 0
batch = []
for i, path in enumerate(image_paths):
try:
vector = embed_image(str(path))
tags = generate_tags(str(path))
img = Image.open(path)
point = {
"id": str(uuid.uuid4()),
"vector": vector.tolist(),
"payload": {
"filename": path.name,
"filepath": str(path.absolute()),
"tags": tags,
"timestamp": int(path.stat().st_mtime),
"width": img.width,
"height": img.height,
}
}
batch.append(point)
indexed += 1
except Exception as e:
print(f"Skipping {path.name}: {e}")
skipped += 1
# Flush every BATCH_SIZE images
if len(batch) >= BATCH_SIZE:
shard.upsert(points=batch)
batch = []
print(f" Progress: {i+1}/{total} images indexed...")
# Flush remaining
if batch:
shard.upsert(points=batch)
return {"total": total, "indexed": indexed, "skipped": skipped}
将图像索引到 Qdrant Edge 中
将图像索引到 Qdrant 的流程，若视为可度量的结构来处理会更为高效。在扩大范围之前，需记录一份最佳实践案例、一个故障实例以及回滚说明。 配置应与应用程序代码分开存放。环境文件、密钥存储和功能标志应集中于一处，以便操作人员无需查看整个流程即可进行审计。 保持图结构简洁且类型明确。嵌套的数据块会掩盖具体是哪个节点修改了哪个字段，还会导致中断后无法继续处理。 将图像索引到 Qdrant 的流程，若视为可度量的结构来处理会更为高效。在扩大范围之前，需记录一份最佳实践案例、一个故障实例以及回滚说明。 相较于复杂的脚本，应优先使用小型且可测试的单元。当某一步骤出现故障时，故障原因应能明确指向某个特定职责，而非整个混乱的流程。
Qdrant Edge 在此处的作用
在“Qdrant Edge的工作原理”阶段，应在修改代码之前明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功检测标准，并拒绝默许的半完成状态。 对于会耗费资金或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。
设置分片
在设置分片阶段时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况可以避免在从演示环境过渡到共享环境时出现意外费用。对于会消耗资金或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。
def get_shard() -> EdgeShard:
"""
Return the singleton EdgeShard, creating it on first call.
- If SHARD_DIR does not exist → create a brand-new shard.
- If SHARD_DIR already contains data → reopen it (no config needed).
EdgeShard runs entirely in-process. No binary, no port, no network.
"""
global _shard
if _shard is not None:
return _shard
SHARD_DIR.mkdir(parents=True, exist_ok=True)
# Detect whether this is a fresh shard or an existing one.
# EdgeShard.create() fails if data already exists on disk.
shard_has_data = any(SHARD_DIR.iterdir())
if shard_has_data:
print(f"[qdrant_client] Reopening existing shard at '{SHARD_DIR}'")
_shard = EdgeShard.load(path=SHARD_DIR)
else:
print(f"[qdrant_client] Creating new shard at '{SHARD_DIR}'")
config = EdgeConfig(
vectors={
VECTOR_NAME: EdgeVectorParams(
size=EMBED_DIM,
distance=Distance.Cosine
)
}
)
_shard = EdgeShard.create(path=str(SHARD_DIR), config=config)
print(f"[store] Shard ready — vector: '{VECTOR_NAME}', dim: {EMBED_DIM}")
return _shard
创建与加载的分工
在“创建”与“加载”阶段，修改代码之前应先明确输入参数、该步骤的负责人以及终止标准。操作人员应当能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。仅靠编译时的连接方式并不足以确保业务流程的完整性。 在“创建”与“加载”阶段，修改代码之前应先明确输入参数、该步骤的负责人以及终止标准。操作人员应当能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 相较于庞大的脚本，应优先使用小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体的责任模块，而非整个复杂的流程。
单例模式
在处理单例模式阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关组件命名，明确成功判定标准，杜绝无声的半完成状态。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
@app.on_event("shutdown")
def on_shutdown():
close_shard()
有效载荷架构
在处理“有效载荷架构”阶段时，首先需明确相关约定：必需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续代码修改的规范性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次对同一次大型语言模型调用收费。
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class PhotoPayload:
filename:str
path:str
tags: list[str] = field(default_factory=list)
timestamp: Optional[str] = None
width: Optional[int] = None
height: Optional[int] = None
def to_dict(self) -> dict:
return {
"filename": self.filename,
"path": self.path,
"tags": self.tags,
"timestamp": self.timestamp,
"width": self.width,
"height": self.height
}
将数据写入分片
在规划“从编写到上线的流程”时，首先需明确合同条款：所需的输入参数、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持透明可追溯。 应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一处，以便操作人员无需查看整个系统结构即可进行审计。 在成本较高的步骤之后设置检查点。当操作人员重新执行后续节点时，恢复流程不应再次调用相同的大型语言模型。
from qdrant_edge import PointStruct
from store.qdrant_client import get_shard
from schema import PhotoPayload
shard = get_shard()
payload = PhotoPayload(
filename = "beach_sunset.jpg",
path = "/Users/me/Pictures/2024/Goa/beach_sunset.jpg",
tags = ["sunset", "beach", "outdoor"],
timestamp = "2024-05-01T18:42:00",
width = 4032,
height = 3024
)
point = PointStruct(
id = "3f7a2b1c-8e4d-4f9a-b2c1-7d8e9f0a1b2c",
vector = {"image": vector.tolist()}, # named vector matching VECTOR_NAME
payload = payload.to_dict()
)
shard.upsert(points=[point])
在规划“从编写到上线的流程”时，首先需明确合同条款：所需的输入参数、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持透明可追溯。 相比庞大的脚本，更应优先使用小型且易于测试的单元。当某个步骤失败时，故障原因应能明确指向某个具体的功能模块，而非整个复杂的流程体系。
自动照片标注
将“自动照片标注”阶段视为可度量的工作面时，其效果最佳。在扩大范围之前，需记录一个成功案例、一个失败案例以及回滚说明。 应将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的半完成状态。 需保持图结构的状态简洁且类型明确。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，且在中断后会导致无法继续处理。
核心理念：零样本分类
将零样本分类阶段视为可度量的模型表面时，其效果最佳。在扩大应用范围之前，先记录一个成功的示例、一个失败案例以及回滚说明。在功能结果旁还需记录处理时间以及令牌或查询成本。提前了解这些成本，就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图结构的层次简单且类型明确，嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在出现中断后导致流程无法继续。
TAG_VOCABULARY = [
"sunset", "sunrise", "beach", "ocean", "mountain", "forest", "city",
"night", "snow", "rain", "fog", "sunny", "cloudy",
"dog", "cat", "bird", "people", "crowd", "portrait", "selfie",
"food", "coffee", "restaurant", "travel", "architecture",
"car", "road", "nature", "flowers", "trees",
"indoor", "outdoor", "party", "celebration", "sport",
"screenshot", "document", "text", "map",
]
# tools/tag.py
from pipeline.embedder import embed_image, embed_text
from config import TAG_THRESHOLD
import numpy as np
TAG_LABELS = [...] # full list as above
# Pre-compute label embeddings once at module load —
# no point re-embedding the same 50 words on every photo
_label_vectors = {
label: embed_text(label)
for label in TAG_LABELS
}
def generate_tags(image_path: str) -> list[str]:
"""
Run zero-shot classification on an image.
Returns a list of tags whose similarity to the image
exceeds TAG_THRESHOLD (default: 0.20).
"""
image_vector = embed_image(image_path)
tags = []
for label, label_vector in _label_vectors.items():
similarity = np.dot(image_vector, label_vector) # cosine sim on normalised vectors
if similarity >= TAG_THRESHOLD:
tags.append(label)
return tags
索引时与查询时的标签
在索引阶段，将标签视为可度量的对象使用效果最佳。在扩大范围之前，先记录一份理想的日志、一个故障案例以及回滚说明。 应将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一处，以便操作人员无需查看整个结构即可进行审计。 要保持图结构的扁平化与类型化。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会导致中断后无法继续处理。 在索引阶段，将标签视为可度量的对象使用效果最佳。在扩大范围之前，先记录一份理想的日志、一个故障案例以及回滚说明。 相比庞大的脚本，更应选择小型且可测试的单元。当某个步骤出现故障时，故障点应指向单一的责任模块，而非复杂的流程链。
def generate_tags_from_vector(img_vec: np.ndarray, threshold: float = 0.20, max_tags: int = 6) -> list[str]:
"""
Generate tags for an image vector using zero-shot CLIP classification.
Tags with cosine similarity above threshold are included (up to max_tags).
This is a utility function used for generating tags during indexing
or when you already have an image vector.
"""
tag_vecs = _get_tag_vectors()
scores = {
tag: float(np.dot(img_vec, vec)) # both normalized → cosine similarity
for tag, vec in tag_vecs.items()
}
tags = sorted(
[t for t, s in scores.items() if s >= threshold],
key=lambda t: scores[t],
reverse=True,
)[:max_tags]
return tags
将标签用作过滤器
在“将标签用作过滤器”阶段，应在修改代码之前明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功检测标准，并拒绝默许的半完成状态。 对于涉及资金支出或更改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。
def search_photos(query: str, top_k: int = TOP_K, tags: list[str] = None) -> list[dict]:
#search photo library with a Natural language query
#takes in query, no of results to be displayed, and a list of tags
# returns list of dicts with photo metadata and relevance score
print(f"[search] Received query='{query}' with tags={tags} and top_k={top_k}")
shard = get_shard()
query_vector = embed_text(query)
# Over-fetch when tag filtering is requested to have enough candidates
# after post-filtering by tags
over_fetch_multiplier = 5 if tags else 1
fetch_limit = top_k * over_fetch_multiplier
results = shard.query(
QueryRequest(
query=Query.Nearest(query_vector.tolist(), using=VECTOR_NAME),
limit=fetch_limit,
with_vector=False,
with_payload=True,
)
)
print(f"[search] Found {len(results)} initial hits for query='{query}' with tags={tags}")
hits = []
untagged_hits = [] # Fallback results for images without tags
for point in results:
payload = point.payload or {}
point_tags = payload.get("tags", [])
result_dict = {
"path": payload.get("path"),
"filename": payload.get("filename"),
"tags": point_tags,
"timestamp": payload.get("timestamp"),
"score": round(point.score, 4)
}
# Post-filter by tags if specified
# (EdgeShard doesn't support complex filters, so we filter in Python
# after over-fetching more results than needed)
if tags:
if point_tags and any(t in point_tags for t in tags):
# Has tags and matches the filter
hits.append(result_dict)
elif not point_tags:
# No tags yet (images not auto-tagged), save as fallback
untagged_hits.append(result_dict)
else:
# No tag filter specified, include all results
hits.append(result_dict)
# Stop if we have enough tagged results
if len(hits) >= top_k:
break
# If we don't have enough tagged results, include untagged ones that match the query
if tags and len(hits) < top_k:
hits.extend(untagged_hits[:top_k - len(hits)])
return hits[:top_k]
构建搜索代理
在构建搜索代理阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前显示成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。对于会消耗资金或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。
curl --location 'http://localhost:8000/search' \
--header 'Content-Type: application/json' \
--data '{"query": "eiffel tower from rooftop","tags":[], "top_k": 1}'
{
"query": "eiffel tower from rooftop",
"results": [
{
"path": "/Users/vatsalasingh/Documents/Datasets/tag_phot/photo-1638051017225-0d9fcca18cf4.jpg",
"filename": "photo-1638051017225-0d9fcca18cf4.jpg",
"tags": [
"cloudy",
"city",
"rain",
"screenshot",
"architecture",
"travel"
],
"timestamp": "2021-12-09T17:27:56",
"score": 0.2864
}
]
}
优雅地处理边缘情况
为优雅地处理边缘情况，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。 为优雅地处理边缘情况，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一的责任主体，而非……
一个错综复杂的流程管道。使用 OpenClaw 统筹一切
在“统筹一切”阶段工作时，首先需明确合同条款：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功检查标准，并杜绝无声的半完成状态。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
OpenClaw 的工作原理
在研究 OpenClaw 的工作原理时，首先需列出相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的 LLM 调用费用。
@app.post("/chat")
def chat(req: ChatRequest):
"""
Conversational endpoint. Accepts user message and conversation history,
returns agent's reply after processing with tools.
"""
# Define tool functions that the agent can call
def search_tool(query: str, top_k: int = 10, tag_filter: list = None):
"""Search photos by natural language query"""
return search_photos(query=query, top_k=top_k, tags=tag_filter)
def duplicates_tool(threshold: float = 0.97):
"""Find duplicate or near-duplicate photos"""
return find_duplicates(threshold=threshold)
def tag_tool(image_path: str):
"""Generate and update tags for a specific photo"""
return generate_tags_from_vector(image_path=image_path)
# Create the agent with tools
agent = Agent(
tools=[search_tool, duplicates_tool, tag_tool],
system_prompt="""
You are a personal photo assistant. You help users search, organize,
and understand their local photo library. You have access to tools
for semantic search, duplicate detection, and tagging.
When helping users:
- Use the search tool to find photos by describing their content
- Use duplicates tool to find and clean up duplicate shots
- Use tag tool to inspect or update tags for specific photos
Always be concise and helpful. When returning photo results,
format them clearly with filenames, similarity scores, and tags.
Use emojis sparingly but helpfully.
"""
)
# Run the agent conversation
response = agent.chat(
message=req.message,
history=req.history
)
return {"response": response}
实际交互流程
在处理实际交互流程阶段时，首先需明确相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中管理，以便操作人员无需查看整个流程就能进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。 在处理实际交互流程阶段时，首先需明确相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 相较于庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障应能指向具体的责任模块，而非整个复杂的流程。
为何代理层如此重要
将代理层阶段视为可度量的界面时，其效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的、不完整的处理。 保持图结构的状态简洁且具有类型约束。嵌套的数据块会掩盖哪个节点修改了哪个字段，且在中断后会导致处理无法继续。
在本地运行所有内容——及其重要性
将“在本地运行一切”及“stage”模式视为可度量的对象来处理时，效果最佳。在扩大范围之前，先记录一份理想的测试结果、一个故障案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图表状态简洁且类型明确，嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在中断后导致无法继续运行。
没有任何数据会离开您的设备。就是这样。
The Nothing leaves your device stage工作最佳的方式是将其视为可测量的表面。在扩大范围之前，先记录一份理想状态下的日志、一个故障案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个流程就能进行审计。 保持流程状态的简洁性与类型化。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，还会导致中断后无法继续执行。 The Nothing leaves your device stage工作最佳的方式是将其视为可测量的表面。在扩大范围之前，先记录一份理想状态下的日志、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出错时，故障应指向单一的责任主体，而非复杂的流程链。
实际运行此系统需要什么
在“实际所需条件”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。
下一步——扩展系统
在“下一步扩展”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解这些成本可以避免在从演示环境过渡到共享环境时出现意外费用。对于那些会消耗资金或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。
人脸聚类
在面部聚类阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审核。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。 在面部聚类阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 相较于庞大的脚本，应优先使用小型且易于测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体的责任模块，而非整个复杂的流程。
截图与文档的OCR处理
在处理截图和文档的OCR任务时，首先需明确合同条款：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续代码修改的透明度。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，定义成功判定标准，并杜绝无声的半完成状态。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的LLM接口。
视频帧索引
在处理视频帧索引阶段时，首先列出相关约定：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前明确成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。
混合搜索：向量 + 关键词 + 元数据
在处理混合搜索向量关键词阶段时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。 在处理混合搜索向量关键词阶段时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 相较于庞大的脚本，应优先使用小型且可测试的单元。当某个步骤失败时，故障应能指向具体的责任模块，而非整个复杂的处理流程。
考虑时间和位置的检索
将“考虑时间和位置”的阶段视为可度量的对象，其效果会更好。在扩大范围之前，先记录一份理想的转录结果、一个失败案例以及回滚说明。 把这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声无息的半完成状态。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。
总结
“最终思考”阶段若被视为可度量的对象，效果会最佳。在扩大范围之前，先记录一份优秀的测试案例、一个失败案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图表状态简洁且类型明确，嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在出现中断后导致流程无法继续。
参考资料与延伸阅读
将“参考资料与延伸阅读”阶段视为可度量的对象来处理，效果最佳。在扩大范围之前，需记录一份理想的运行日志、一个故障案例以及回滚说明。 应将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一处，以便操作人员无需查看整个系统结构即可进行审计。 需保持系统状态的简洁性与类型化。嵌套的数据结构会掩盖具体是哪个节点修改了哪个字段，还会导致在出现中断后无法继续执行。 将“参考资料与延伸阅读”阶段视为可度量的对象来处理，效果最佳。在扩大范围之前，需记录一份理想的运行日志、一个故障案例以及回滚说明。 相比庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障应指向单一的责任模块，而非复杂的流程链。
操作检查清单
将操作检查清单视为可度量的基准，这样效果最佳。在扩大范围之前，先记录一份完美的操作流程、一个故障案例以及回滚说明。
同时记录正常流程和恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的内容。
保持图结构扁平且具有类型约束。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，还会导致中断后无法继续执行。
在预算允许的情况下，使用测试桩而非真实的付费 API，在持续集成过程中添加能够检测关键路径的冒烟测试。
优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向具体的责任模块，而非复杂的流程链。
保持图结构扁平且具有类型约束。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，还会导致中断后无法继续执行。
在推广该技术栈之前，应先冻结版本，为关键流程记录标准输出日志，并明确回滚步骤。共享环境需要设置速率限制、租户验证机制，以及负责密钥轮换的明确责任人。与其展示花哨的一次性演示，不如注重扎实的可靠性。
b7b9768f8acd 的批量处理说明：不要将提供商密钥放入代码仓库，为每个会话设置令牌使用上限，并将日志存储在评估用示例文件旁边，以便后续模型更换时仍能保持对比性。