Building Context-Aware Search in Python with LLM Embeddings + Metadata

pip install sentence-transformers numpy

from datetime import date

tickets=[
{"id":"T-101","team":"infrastructure","status":"open","priority":"high",
"created":date(2025,11,3),
"text":"Kubernetes pod keeps crashing with OOMKilled — memory limits on the ML inference container are set too low for the model it loads at runtime."},
{"id":"T-102","team":"infrastructure","status":"open","priority":"high",
"created":date(2025,11,8),
"text":"Nginx ingress returning 502 after rotating TLS certificate. Chain is valid per openssl verify but the backend handshake fails immediately."},
{"id":"T-103","team":"infrastructure","status":"resolved","priority":"medium",
"created":date(2025,10,14),
"text":"Terraform state file locked in S3 — a team member force-applied a plan without releasing the DynamoDB lock first."},
...
{"id":"T-401","team":"infrastructure","status":"open","priority":"medium",
"created":date(2025,11,11),
"text":"CI pipeline fails on ARM64 runners — base Docker image has no ARM variant, exec format error at build stage."},
{"id":"T-402","team":"infrastructure","status":"resolved","priority":"high",

"created": date(2025, 10, 9),

"text": "高峰时段 VPN 网关延迟激增 —— 两个对等点之间的 BGP 路由抖动导致私有子网中出现间歇性丢包。"

open_ct = sum(1 for t in tickets if t["status"] == "open")
resolved_ct = sum(1 for t in tickets if t["status"] == "resolved")
print(f"{len(tickets)} tickets | {open_ct} open | {resolved_ct} resolved")

20 tickets | 14 open | 6 resolved

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("all-MiniLM-L6-v2")
texts = [t["text"] for t in tickets]
embeddings = model.encode(texts, normalize_embeddings=True, show_progress_bar=True)
print(f"Shape: {embeddings.shape} | norm[0]: {np.linalg.norm(embeddings[0]):.4f}")

class ContextAwareIndex:
    def __init__(self, embeddings: np.ndarray, documents: list):
        self.embeddings = embeddings  # (N, D), L2-normalized
        self.documents = documents

    def search(
        self,
        query: str,
        top_k: int = 5,
        team: str = None,
        status: str = None,
        priority: str = None,
        after: "date" = None,
        before: "date" = None,
        min_score: float = 0.0,
    ) -> list[dict]:
        # 将查询转换成与文档相同的向量空间中的向量
        q_vec = model.encode([query], normalize_embeddings=True)[0]

        # 构建布尔掩码 —— 不满足过滤条件的文档标记为 False
        mask = np.ones(len(self.documents), dtype=bool)
        for i, doc in enumerate(self.documents):
            if team and doc["team"] != team:
                mask[i] = False
            if status and doc["status"] != status:
                mask[i] = False
            if priority and doc["priority"] != priority:
                mask[i] = False
            if after and doc["created"] < after:
                mask[i] = False
            if before and doc["created"] > before:
                mask[i] = False

        candidate_idx = np.where(mask)[0]
        if len(candidate_idx) == 0:
            return []

        # 对通过筛选的候选文档进行打分
        scores = self.embeddings[candidate_idx] @ q_vec

        # 去掉低于最小分数阈值的结果，排序后返回前 k 个
        valid = np.where(scores >= min_score)[0]
        if len(valid) == 0:
            return []

        top_local = np.argsort(scores[valid])[::-1][:top_k]
        top_global = candidate_idx[valid[top_local]]

        return [
            {**self.documents[i], "score": float(scores[valid[top_local[j]]])}
            for j, i in enumerate(top_global)
        ]

index = ContextAwareIndex(embeddings, tickets)

results = index.search("authentication token expiry and session management", top_k=4)
show("'authentication token expiry and session management' (no filters)", results)

Query: 'authentication token expiry and session management' (no filters)
[0.6133] T-207 backend open high 2025-11-03
登出后 Session Cookie 仍然存在 —— 中间件缺少令牌黑名单校验机制...
[0.4958] T-201 backend open high 2025-11-05
OAuth2 刷新令牌间歇性失败 —— 令牌缓存中存在竞态条件问题...
[0.3459] T-203 backend open medium 2025-11-01
JWT 签名验证间歇性失败 —— 鉴权服务器与应用服务时间偏差达 4 秒...
[0.1714] T-206 backend open high 2025-11-13
速率限制未按用户区分 —— 中间件使用的是共享 Redis 键而非用户唯一标识...

results = index.search(
    "authentication token expiry and session management",
    top_k=4,
    status="open",
    before=date(2025, 11, 10),
)

show("same query[status=open, before=2025-11-10]",results)

1
2
3
4
5
6
7
8
9
Query:same query[status=open,before=2025-11-10]

[0.6133]T-207 backend open high 2025-11-03

登出后会话 Cookie 仍然存在——中间件缺少令牌黑名单检查...

[0.4958]T-201 backend open high 2025-11-05

OAuth2 令牌刷新间歇性失败——令牌缓存中的竞态条件导致...

[0.3459]T-203 backend open medium 2025-11-01

JWT 签名验证间歇性失败——签名服务器与验证服务之间存在 4 秒的时钟偏差...

[0.1419]T-202 backend open high 2025-11-09

负载下数据库连接池耗尽——连接池限制为 20 个连接，但实际需求更高...

results=index.search(
    "resource exhaustion and memory pressure under load",
    top_k=2,
    status="open",
    priority="high",
)
show("'resource exhaustion and memory pressure'[status=open, priority=high]",results)

1
2
3
4
5
Query:'resource exhaustion and memory pressure'[status=open,priority=high]

[0.3877]T-202 backend open high 2025-11-09

Database connection pool exhausted under load—pool capped at 20 connections but the...

[0.2908]T-101 infrastructure open high 2025-11-03

Kubernetes pod keeps crashing with OOMKilled—memory limits on the ML inference cont...

import json

# 将嵌入矩阵和工单元数据写入磁盘
np.save("ticket_embeddings.npy",embeddings)
with open("ticket_metadata.json","w") as f:
    json.dump(
        [{**t,"created":t["created"].isoformat()} for t in tickets],
        f, indent=2,
    )

from datetime import date
import json
import numpy as np
from sentence_transformers import SentenceTransformer

# 加载嵌入矩阵，反序列化元数据，重建索引
embeddings_loaded = np.load("ticket_embeddings.npy")
with open("ticket_metadata.json") as f:
    tickets_loaded = json.load(f)
for t in tickets_loaded:
    t["created"] = date.fromisoformat(t["created"])

model = SentenceTransformer("all-MiniLM-L6-v2")
index = ContextAwareIndex(embeddings_loaded, tickets_loaded)
print(f"Reloaded: {embeddings_loaded.shape[0]} docs, {embeddings_loaded.shape[1]}D.")