Bedrock Knowledge Base Metadata: Filter Retrieval by Language and Repo

// s3://code-kb/payments/src/getOrder.ts.metadata.json
{
 "metadataAttributes": {
 "repo": "acme/backend",
 "service": "payments",
 "language": "typescript",
 "owner": "payments-platform",
 "path": "services/payments/src/getOrder.ts",
 "visibility": "internal"
 }
}

# sync_chunk.py — write chunk + metadata for Bedrock KB S3 data source
def write_chunk(bucket: str, key: str, text: str, meta: dict):
 s3.put_object(Bucket=bucket, Key=key, Body=text.encode("utf-8"))
 s3.put_object(
 Bucket=bucket,
 Key=f"{key}.metadata.json",
 Body=json.dumps({"metadataAttributes": meta}).encode("utf-8"),
 ContentType="application/json",
 )

import boto3
bedrock_agent = boto3.client("bedrock-agent-runtime")

def retrieve_for_ide(query: str, *, repo: str, language: str | None, service: str | None):
 and_all = [{"equals": {"key": "repo", "value": repo}}]
 if language:
 and_all.append({"equals": {"key": "language", "value": language}})
 if service:
 and_all.append({"equals": {"key": "service", "value": service}})

 return bedrock_agent.retrieve(
 knowledgeBaseId=KB_ID,
 retrievalQuery={"text": query},
 retrievalConfiguration={
 "vectorSearchConfiguration": {
 "numberOfResults": 8,
 "filter": {"andAll": and_all},
 }
 },
 )

function filtersFromEditor(ctx: IdeContext) {
 return {
 repo: ctx.gitRemoteSlug, // acme/backend
 language: mapVscodeLang(ctx.languageId),
 service: detectService(ctx.filePath), // services/payments/...
 };
}