Skip to content

task: optimize vector indexes for brujula investigate index rebuild core 104 #690

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
irumvanselme merged 4 commits into from
Mar 10, 2026
Merged

task: optimize vector indexes for brujula investigate index rebuild core 104 #690

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b8f79b7
feat(server): add swagger UI parameter to collapse documentation sect…
irumvanselme Mar 10, 2026
1366934
feat(esco_search): optimize skill aggregation pipeline for improved p…
irumvanselme Mar 9, 2026
bf424f6
feat(index): create compound index on skillId and modelId for optimiz…
irumvanselme Mar 9, 2026
2943fc2
refactor(generate_taxonomy_embeddings): do not validate the model det…
irumvanselme Mar 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions backend/app/server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -314,6 +314,7 @@ async def lifespan(_app: FastAPI):
version=get_application_config().version_info.to_version_string(),
description=f"The {_global_product_name} API is used to interact with the {_global_product_name} conversation agent.",
redirect_slashes=False,
swagger_ui_parameters={"docExpansion": "none"},
servers=[
{
"url": backend_url or "/",
Expand Down
61 changes: 29 additions & 32 deletions backend/app/vector_search/esco_search_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -388,51 +388,48 @@ async def _find_skills_of_occupation(self, occupation: OccupationEntity):
if cached_result is not None:
return cached_result

skills = await self.relations_collection.aggregate([
pipeline = [
{"$match": {"modelId": self._model_id, "requiringOccupationId": ObjectId(occupation.id)}},
{"$project": {
"requiredSkillId": 1,
"modelId": 1,
"relationType": 1,
"signallingValueLabel": 1,
}},
{
"$lookup": {
"from": self.embedding_config.skill_collection_name,
"localField": "requiredSkillId",
"foreignField": "skillId",
"as": "skills",
"pipeline": [
{"$match": {"modelId": self._model_id}}
{"$match": {"modelId": self._model_id}},
# Limit 1 of the skills (Note: each skill has 3 corresponding documents, pick the first.
{"$limit": 1},
Comment on lines +407 to +408

Copilot AI Mar 10, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The $limit: 1 stage in the inner $lookup pipeline picks an arbitrary document (the first one returned by MongoDB) out of the potentially 3 copies of a skill document (one per embedded field: description, preferredLabel, altLabels). Because there is no $sort before the $limit, the document returned is non-deterministic and may vary between queries or after index changes. While the non-embedding fields should be identical across all 3 copies, it would be safer and clearer to add a $sort: {"embedded_field": 1} (or any stable sort) before $limit: 1 to ensure a deterministic and predictable result.

Copilot uses AI. Check for mistakes.
# Exclude large/irrelevant fields from the result to reduce network transfer and improve latency
{"$project": { "embedding": 0, "embedded_field": 0, "embedded_text": 0, }}
]
}
},
{"$unwind": "$skills"},
{"$group": {"_id": "$skills.skillId",
"modelId": {"$first": "$modelId"},
"skillId": {"$first": "$skills.skillId"},
"UUID": {"$first": "$skills.UUID"},
"preferredLabel": {"$first": "$skills.preferredLabel"},
"description": {"$first": "$skills.description"},
"scopeNote": {"$first": "$skills.scopeNote"},
"originUUID": {"$first": "$skills.originUUID"},
"UUIDHistory": {"$first": "$skills.UUIDHistory"},
"altLabels": {"$first": "$skills.altLabels"},
"skillType": {"$first": "$skills.skillType"},
"relationType": {"$first": "$relationType"},
"signallingValueLabel": {"$first": "$signallingValueLabel"},
}
}
]).to_list(length=None)
{"$unwind": "$skills"}
]

skills_relationships = await self.relations_collection.aggregate(pipeline).to_list(length=None)
result = [AssociatedSkillEntity(
id=str(skill.get("skillId", "")),
modelId=str(skill.get("modelId", "")),
UUID=skill.get("UUID", ""),
preferredLabel=skill.get("preferredLabel", ""),
description=skill.get("description", ""),
scopeNote=skill.get("scopeNote", ""),
altLabels=skill.get("altLabels", []),
skillType=skill.get("skillType", ""),
originUUID=skill.get("originUUID", ""),
UUIDHistory=skill.get("UUIDHistory", []),
relationType=skill.get("relationType", ""),
signallingValueLabel=skill.get("signallingValueLabel", ""),
id=str(skill_relationship.get("skills").get("skillId", "")),
modelId=str(skill_relationship.get("modelId", "")),
UUID=skill_relationship.get("skills").get("UUID", ""),
preferredLabel=skill_relationship.get("skills").get("preferredLabel", ""),
description=skill_relationship.get("skills").get("description", ""),
scopeNote=skill_relationship.get("skills").get("scopeNote", ""),
altLabels=skill_relationship.get("skills").get("altLabels", []),
skillType=skill_relationship.get("skills").get("skillType", ""),
originUUID=skill_relationship.get("skills").get("originUUID", ""),
UUIDHistory=skill_relationship.get("skills").get("UUIDHistory", []),
relationType=skill_relationship.get("relationType", ""),
signallingValueLabel=skill_relationship.get("signallingValueLabel", ""),
score=0.0
) for skill in skills]
) for skill_relationship in skills_relationships]
Comment on lines +422 to +432

@sentry sentry Bot Mar 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: The removal of the $group stage from the aggregation pipeline may lead to duplicate skills being returned if the underlying relations collection contains duplicate entries.
Severity: MEDIUM

Suggested Fix

Reintroduce a deduplication mechanism. The safest approach is to add a $group stage back into the MongoDB aggregation pipeline to group results by skillId, mirroring the previous implementation's defensive approach. Alternatively, perform deduplication in the application code after fetching the results.

Prompt for AI Agent 
Review the code at the location below. A potential bug has been identified by an AI
agent.
Verify if this is a real issue. If it is, propose a fix; if not, explain why it's not
valid.

Location: backend/app/vector_search/esco_search_service.py#L414-L432

Potential issue: The MongoDB aggregation pipeline was modified by removing a `$group`
stage that previously deduplicated results by `skillId`. The underlying `relations`
collection does not enforce uniqueness on the combination of `requiringOccupationId` and
`requiredSkillId`. If duplicate relation documents exist in the collection, this change
will cause the query to return duplicate `AssociatedSkillEntity` entries, whereas the
previous implementation would have returned a unique list. This is a functional
regression that relies on an unverified assumption that the source data contains no
duplicates.

Did we get this right? 👍 / 👎 to inform future reviews.

@irumvanselme irumvanselme Mar 10, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@sentry, the fact that I added limit 1 in the previous stage pipeline, won't that sovle the problem of removing duplicates?


# Cache the result
await _skills_of_occupation_cache.set(occupation.id, result)
Expand Down
23 changes: 23 additions & 0 deletions backend/scripts/embeddings/_common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,29 @@ async def create_skills_indexes(*, hot_run: bool,
id_field_name="skillId",
logger=logger,
)

# Create an index on (skillId, modelId).
# This index optimizes MongoDB $lookup operations where documents
# are searched by `skillId` within a specific `modelId`.
#
# The default compound index order (modelId, skillId) is not optimal
# for this query pattern because MongoDB can miss the index when the
# leading field is not used first. In some cases this causes the query
# planner to skip the index and perform a less efficient scan.
#
# Therefore, we explicitly create the index with `skillId` as the
# leading field to match the lookup access pattern.
await _upsert_index(
hot_run=hot_run,
collection=collection,
keys={
"skillId": 1,
"modelId": 1,
},
name="skill_id_model_id_index",
logger=logger,
)

await _create_vector_search_index(hot_run=hot_run,
collection=collection,
num_of_dimensions=num_of_dimensions,
Expand Down
26 changes: 13 additions & 13 deletions backend/scripts/embeddings/generate_taxonomy_embeddings.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -404,21 +404,21 @@ async def main(opts: Options):
logger.info("Starting the main function")
logger.info("Using options: " + opts.__str__())

if opts.delete_existing:
# Delete existing relations and model info collections
for collection in CompassEmbeddingsCollections:
await delete_existing(hot_run=opts.hot_run,
collection_name=collection.value,
model_id=SCRIPT_SETTINGS.tabiya_model_id)
# [1/5] Copy the model info and validate for existing state, if it is compatible with the new state.
if opts.generate_embeddings:
if opts.delete_existing:
# Delete existing relations and model info collections
for collection in CompassEmbeddingsCollections:
await delete_existing(hot_run=opts.hot_run,
collection_name=collection.value,
model_id=SCRIPT_SETTINGS.tabiya_model_id)

embeddings_service = await get_embeddings_service(service_name=SCRIPT_SETTINGS.embeddings_service_name,
model_name=SCRIPT_SETTINGS.embeddings_model_name)
embeddings_service = await get_embeddings_service(service_name=SCRIPT_SETTINGS.embeddings_service_name,
model_name=SCRIPT_SETTINGS.embeddings_model_name)

# [1/5] Copy the model info and validate for existing state, if it is compatible with the new state.
await _copy_model_info(hot_run=opts.hot_run, embeddings_service=embeddings_service)
await _copy_model_info(hot_run=opts.hot_run, embeddings_service=embeddings_service)

# run the three tasks in parallel
if opts.generate_embeddings:
# run the three tasks in parallel
await asyncio.gather(
# [2/5] Copy the relations collection
copy_relations_collection(hot_run=opts.hot_run),
Expand All @@ -438,8 +438,8 @@ async def main(opts: Options):
num_of_dimensions=opts.num_of_dimensions,
logger=logger)

logger.info("Script execution completed")

logger.info("Script execution completed")

if __name__ == "__main__":
try:
Expand Down
Loading