How to Add Web Data to an Amazon Bedrock Knowledge Base

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "bedrock:IngestKnowledgeBaseDocuments",
        "bedrock:DeleteKnowledgeBaseDocuments",
        "bedrock:ListKnowledgeBaseDocuments",
        "bedrock:StartIngestionJob",
        "bedrock:Retrieve"
      ],
      "Resource": "arn:aws:bedrock:us-east-1:ACCOUNT_ID:knowledge-base/KB_ID"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:ListBucket"],
      "Resource": ["arn:aws:s3:::web-to-bedrock-ACCOUNT_ID", "arn:aws:s3:::web-to-bedrock-ACCOUNT_ID/*"]
    },
    {
      "Effect": "Allow",
      "Action": "lambda:InvokeFunction",
      "Resource": "arn:aws:lambda:us-east-1:ACCOUNT_ID:function:web-to-bedrock"
    },
    {
      "Effect": "Allow",
      "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
      "Resource": "*"
    }
  ]
}

"""Crawl a website section with Decodo's Web Data API and load it into an Amazon Bedrock knowledge base.

Local commands:
  python web_to_bedrock.py map   https://kubernetes.io/docs/concepts/
  python web_to_bedrock.py crawl https://kubernetes.io/docs/concepts/ https://<function-url>
  python web_to_bedrock.py crawl https://kubernetes.io/docs/concepts/   # no webhook: waits, then processes here
  python web_to_bedrock.py ask   "How does maxSkew work in topology spread constraints?" [max_age_days]
  python web_to_bedrock.py fill  "How do I safely drain a node before maintenance?"
  python web_to_bedrock.py process <crawl_id>   # reprocess a finished crawl, for example after a function error

Deployed as an AWS Lambda function with a function URL, the same file receives
the crawl webhook, cleans the pages, and pushes them into the knowledge base.
A scheduler can start each re-crawl by invoking the function with
{"crawl": "https://kubernetes.io/docs/concepts/", "callback_url": "https://<function-url>"},
and your app can run fill by invoking it with {"fill": "<question>"}.
"""
import collections
import hashlib
import hmac
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor

import boto3
from botocore.config import Config

API = "https://data.decodo.com/v1"
DECODO_API_KEY = os.environ["DECODO_API_KEY"]
KB_ID = os.environ["KB_ID"]
DS_ID = os.environ["DS_ID"]                    # CUSTOM data source, or the S3 data source when ROUTE=s3
BUCKET = os.environ["BUCKET"]                  # holds the page manifest, and the pages when ROUTE=s3
WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"]  # sent as passthrough, checked when the webhook arrives
ROUTE = os.environ.get("ROUTE", "custom")      # "custom" (direct ingestion) or "s3" (sync job)
CONTACT_EMAIL = os.environ.get("CONTACT_EMAIL", "")  # some sites, such as sec.gov, refuse robots.txt requests without one
ALLOWED_SITES = [s.strip().lower() for s in os.environ.get("ALLOWED_SITES", "").split(",") if s.strip()]  # hosts fill may use
DELETE_MISSING = os.environ.get("DELETE_MISSING", "true") == "true"  # "false" for news, where pages leave the listings but stay online

CRAWL_LIMIT = 2000      # pages per crawl; a crawl that reaches it is treated as partial
ROBOTS_AGENTS = {"*", "chatgpt-user", "oai-searchbot", "claude-user", "claude-searchbot",
                 "perplexity-user", "perplexitybot"}  # AI search and assistant crawlers, which many news sites block by name
EXCLUDE = ["/_print"]   # print views repeat whole sections, so the same text would be indexed several times
MAX_EXCLUDE_CHARS = 700 # size of the combined robots.txt pattern; above it, crawl a narrower path
MIN_CHARS = 200         # skip near-empty pages after cleaning
BLOCK_PAGE = re.compile(  # phrases that mark a site's error or bot-check page rather than content
    r"captcha|robot or human|not a robot|confirm that you.re human|humans only|verify(ing)? (your )?(connection|browser)"
    r"|checking your browser|additional verification|unusual traffic|just a moment|pardon our interruption"
    r"|access denied|access to this page has been denied|you have been blocked|please enable (js|javascript|cookies)"
    r"|enable javascript and|taking longer than usual|currently unavailable", re.IGNORECASE)
MAX_BLOCK_PAGE_CHARS = 3000  # only short pages are checked, so a docs page about access control isn't skipped
ALERT_SHARE = 0.10      # alert when more than 10% of a crawl's pages are skipped
BATCH = 10              # managed knowledge bases rejected batches over 10 in testing
MAX_REQUEST_BYTES = 5_000_000  # AWS rejects ingestion requests of 6 MB or more
WORKERS = 4             # parallel ingestion calls, each about 2.5 s for 10 pages in testing
MAX_DELETE_SHARE = 0.15 # skip deletes if one crawl would remove more than 15% of indexed pages
MIN_TEMPLATE_PAGES = 20 # a crawl this size learns the template; smaller crawls and fill reuse the saved one
MANIFEST_KEY = "state/manifest.json"   # pages from the scheduled crawl
GAPS_KEY = "state/gaps.json"           # pages added by fill, kept out of the crawl's delete logic
TEMPLATE_KEY = "state/template.json"

agent = boto3.client("bedrock-agent", config=Config(retries={"mode": "adaptive", "max_attempts": 10}))
runtime = boto3.client("bedrock-agent-runtime")
s3 = boto3.client("s3")


def decodo(url, body=None, attempts=4):
    req = urllib.request.Request(
        url if url.startswith("http") else API + url,
        data=json.dumps(body).encode() if body is not None else None,
        method="POST" if body is not None else "GET",
        headers={"Authorization": f"Bearer {DECODO_API_KEY}",
                 "Content-Type": "application/json", "Accept": "application/json"},
    )
    retry = (429, 500, 502, 503, 504, 524, 613) if body is None else (429,)  # a repeated POST /crawl could start a 2nd job
    for attempt in range(attempts):
        try:
            with urllib.request.urlopen(req, timeout=120) as resp:
                return json.load(resp)
        except urllib.error.HTTPError as e:
            if e.code not in retry or attempt == attempts - 1:
                raise
            time.sleep(2 ** attempt)  # transient errors, safe to retry; 613 is a Decodo server-side code, like 500 and 524


def map_site(url):
    links = decodo("/map", {"url": url, "sitemap": "include"})["results"][0]["content"]["links"]
    paths = collections.Counter("/".join(l["url"].split("/")[3:6]) for l in links)
    for path, n in paths.most_common(20):
        print(f"{n:5}  /{path}")
    print(f"{len(links)} URLs")


def start_crawl(url, callback_url=None):
    path = urllib.parse.urlparse(url).path.rstrip("/")
    rules = robots_excludes(url)
    if disallowed(url, rules):  # exclude_paths filters discovered pages, so check the start URL here
        print("ALERT robots.txt disallows", url)
        return None
    rules = in_scope(rules, path)
    robots = ["^(?:" + "|".join(rule[1:] for rule in rules) + ")"] if rules else []  # all in-scope rules as one pattern
    if robots and len(robots[0]) > MAX_EXCLUDE_CHARS:
        print(f"ALERT robots.txt has {len(rules)} rules under {path or '/'}, so crawl a narrower path")
        return None
    body = {
        "url": url,
        "limit": CRAWL_LIMIT,
        "select_paths": [f"^{re.escape(path)}"],
        "exclude_paths": EXCLUDE + robots,
        "sitemap": "include",
        "output": "markdown",
        "JS_retry": True,
    }
    if callback_url:
        body.update(callback_url=callback_url, passthrough=WEBHOOK_SECRET)
    job = decodo("/crawl", body)
    print(job["crawl_id"], job["status"])
    if callback_url:
        return job["crawl_id"]
    # No webhook: wait, then process here. In testing, "pending" was the only status while a crawl ran.
    while decodo(f"/crawl/{job['crawl_id']}/status")["status"] == "pending":
        time.sleep(15)
    return process(job["crawl_id"])


def robots_excludes(url):
    """Paths that robots.txt disallows for ROBOTS_AGENTS, as exclude_paths patterns. Allow lines are ignored, which excludes more."""
    root = "{0.scheme}://{0.netloc}".format(urllib.parse.urlparse(url))
    agent_header = {"User-Agent": f"web-to-bedrock/1.0 ({CONTACT_EMAIL})" if CONTACT_EMAIL else "web-to-bedrock/1.0"}
    try:
        with urllib.request.urlopen(urllib.request.Request(root + "/robots.txt", headers=agent_header), timeout=30) as resp:
            lines = resp.read().decode("utf-8-sig", errors="ignore").splitlines()
    except urllib.error.HTTPError as e:
        if e.code >= 500 or e.code in (401, 403, 429):
            print("ALERT robots.txt request refused with HTTP", e.code, "for", root)
            raise  # rules unknown, so crawl nothing; a 403 can mean the site wants CONTACT_EMAIL or refuses your network
        return []  # no robots.txt, no restrictions
    rules, agents, in_rules = [], set(), False
    for line in lines:
        field, _, value = line.split("#")[0].partition(":")
        field, value = field.strip().lower(), value.strip()
        if field == "user-agent":
            if in_rules:  # a User-agent line after rules starts a new group
                agents, in_rules = set(), False
            agents.add(value.lower())
        elif field in ("allow", "disallow"):
            in_rules = True
            if field == "disallow" and value and agents & ROBOTS_AGENTS:
                pattern = re.escape(value).replace(r"\*", ".*").replace(r"\$", "$")
                rules.append("^" + re.sub(r"/\$?$", "(/|$)", pattern))  # exclude_paths patterns match paths without a trailing slash
    return rules


def in_scope(rules, path):
    """Rules that can match a URL under path. select_paths keeps the crawl inside it, so other rules can't apply."""
    kept = []
    for rule in rules:
        literal = re.sub(r"\\(.)", r"\1", re.match(r"\^((?:\\.|[^\\.*+?()\[\]{}|$])*)", rule).group(1))
        if literal.startswith(path) or path.startswith(literal):
            kept.append(rule)
    return kept


def disallowed(url, rules=None):
    parts = urllib.parse.urlparse(url)
    path = (parts.path or "/") + ("?" + parts.query if parts.query else "")
    return any(re.match(rule, path) for rule in (robots_excludes(url) if rules is None else rules))


def looks_blocked(text):
    return len(text) < MAX_BLOCK_PAGE_CHARS and BLOCK_PAGE.search(text) is not None


def fetch_pages(crawl_id):
    pages, url = [], f"/crawl/{crawl_id}?type=content"
    while url:
        batch = decodo(url)
        pages += batch["url"]  # Decodo lists the pages of this response under "url"
        url = batch.get("next")
    return pages


# The site template (menus, docs tree, footer) repeats on every page. A line that
# appears on at least half of the pages is template, whatever its link target.
LINK = re.compile(r"\]\([^)]*\)")
TABLE_RULE = re.compile(r"^\s*\|?[\s:|-]+\|?\s*$")
FENCE = "`" * 3


def line_key(line):
    return LINK.sub("]()", line.strip())


def template_lines(texts, share=0.5):
    counts = collections.Counter()
    for text in texts:
        counts.update({line_key(l) for l in text.splitlines() if l.strip()})
    return {k for k, n in counts.items() if n >= share * len(texts)}


def strip_template(text, template):
    kept, in_code, title = [], False, None
    for line in text.splitlines():
        if line.strip().startswith(FENCE):
            in_code = not in_code
            kept.append(line)
        elif in_code or not line.strip() or TABLE_RULE.match(line) or line_key(line) not in template:
            if title is None and not in_code and line.startswith("# "):
                title = len(kept)  # per-page navigation, such as a list of translations, sits above the title
            kept.append(line)
    return re.sub(r"\n{3,}", "\n\n", "\n".join(kept[title or 0:])).strip()


def load_json(key, default):
    try:
        return json.loads(s3.get_object(Bucket=BUCKET, Key=key)["Body"].read())
    except s3.exceptions.NoSuchKey:
        return default


def save_json(key, value):
    s3.put_object(Bucket=BUCKET, Key=key, Body=json.dumps(value).encode())


def attributes(url, changed_at, digest):
    # Unchanged pages are not pushed again, so this is the time a crawl last found new content
    return {"source_url": url, "site": url.split("/")[2], "changed_at": changed_at, "content_sha256": digest}


def custom_doc(url, text, attrs):
    inline = [{"key": k, "value": {"type": "NUMBER", "numberValue": v} if isinstance(v, int)
               else {"type": "STRING", "stringValue": v}} for k, v in attrs.items()]
    return {
        "content": {"dataSourceType": "CUSTOM", "custom": {
            "customDocumentIdentifier": {"id": url}, "sourceType": "IN_LINE",
            "inlineContent": {"type": "TEXT", "textContent": {"data": text}}}},
        "metadata": {"type": "IN_LINE_ATTRIBUTE", "inlineAttributes": inline},
    }


def s3_key(url):
    return "pages/" + hashlib.sha256(url.encode()).hexdigest()[:24] + ".md"


def batches(docs):
    """Up to BATCH documents per request, and under MAX_REQUEST_BYTES in total."""
    batch, size = [], 0
    for doc in docs:
        n = len(json.dumps(doc).encode())
        if batch and (len(batch) == BATCH or size + n > MAX_REQUEST_BYTES):
            yield batch
            batch, size = [], 0
        batch.append(doc)
        size += n
    if batch:
        yield batch


def push_custom(changed, removed):
    docs = [custom_doc(*c) for c in changed]
    with ThreadPoolExecutor(WORKERS) as pool:
        list(pool.map(lambda b: agent.ingest_knowledge_base_documents(
            knowledgeBaseId=KB_ID, dataSourceId=DS_ID, documents=b), batches(docs)))
    for i in range(0, len(removed), BATCH):
        agent.delete_knowledge_base_documents(knowledgeBaseId=KB_ID, dataSourceId=DS_ID, documentIdentifiers=[
            {"dataSourceType": "CUSTOM", "custom": {"id": u}} for u in removed[i:i + BATCH]])


def put_page(url, text, attrs):
    meta = {"metadataAttributes": {k: {"value": {"type": "NUMBER", "numberValue": v} if isinstance(v, int)
                                       else {"type": "STRING", "stringValue": v}} for k, v in attrs.items()}}
    s3.put_object(Bucket=BUCKET, Key=s3_key(url), Body=text.encode(), ContentType="text/markdown")
    s3.put_object(Bucket=BUCKET, Key=s3_key(url) + ".metadata.json", Body=json.dumps(meta).encode())


def push_s3(changed, removed):
    with ThreadPoolExecutor(WORKERS) as pool:
        list(pool.map(lambda c: put_page(*c), changed))
    for url in removed:
        s3.delete_objects(Bucket=BUCKET, Delete={"Objects": [{"Key": s3_key(url)}, {"Key": s3_key(url) + ".metadata.json"}]})
    if changed or removed:
        job = agent.start_ingestion_job(knowledgeBaseId=KB_ID, dataSourceId=DS_ID)["ingestionJob"]
        print("ingestion job", job["ingestionJobId"])


def failed_urls(manifest):
    """Pages in the manifest whose last ingestion failed, so this run sends them again."""
    by_uri = {f"s3://{BUCKET}/{s3_key(u)}": u for u in manifest}
    failed, kw = set(), {}
    while True:
        resp = agent.list_knowledge_base_documents(knowledgeBaseId=KB_ID, dataSourceId=DS_ID, maxResults=100, **kw)
        for d in resp["documentDetails"]:
            ident = d["identifier"]
            url = ident["custom"]["id"] if "custom" in ident else by_uri.get(ident.get("s3", {}).get("uri"))
            if "FAILED" in d["status"] and url in manifest:
                failed.add(url)
        if not resp.get("nextToken"):
            return failed
        kw = {"nextToken": resp["nextToken"]}


def process(crawl_id):
    status = decodo(f"/crawl/{crawl_id}/status")["status"]
    pages = fetch_pages(crawl_id)
    seen = {p["url"] for p in pages}
    done = [p for p in pages if p["status"] == "done" and p.get("content")]
    if len(done) >= MIN_TEMPLATE_PAGES:  # learned per crawl, so keep the scope fixed between runs
        template = template_lines([p["content"] for p in done])
        save_json(TEMPLATE_KEY, sorted(template))
    else:  # too few pages to tell the template from content
        template = set(load_json(TEMPLATE_KEY, []))
    manifest, now = load_json(MANIFEST_KEY, {}), int(time.time())
    failed = failed_urls(manifest)
    for u in failed:
        del manifest[u]  # treated as a new page below

    changed, current, suspect, alerts = [], {}, [], []
    if failed:
        alerts.append(f"{len(failed)} pages failed to index in the previous run, sent again")
    for p in done:
        text = strip_template(p["content"], template)
        if len(text) < MIN_CHARS or looks_blocked(text):
            suspect.append(p["url"])  # keeps the indexed copy, like a failed page
            continue
        if ROUTE != "s3" and len(text.encode()) >= MAX_REQUEST_BYTES:
            alerts.append(f"{p['url']} is too large for 1 ingestion request")  # left out of the manifest, so the next crawl tries again
            continue
        digest = hashlib.sha256(text.encode()).hexdigest()
        current[p["url"]] = digest
        if manifest.get(p["url"]) != digest:
            changed.append((p["url"], text, attributes(p["url"], now, digest)))

    # A URL that a finished crawl no longer discovers is gone. A page that failed this time stays indexed.
    removed = [u for u in manifest if u not in seen] if DELETE_MISSING else []
    if status != "done" or len(pages) >= CRAWL_LIMIT:  # an incomplete crawl didn't reach every page, so missing isn't gone
        alerts.append(f"crawl {status} with {len(pages)} pages, skipped {len(removed)} deletes")
        removed = []
    elif len(removed) > MAX_DELETE_SHARE * max(1, len(manifest)):
        alerts.append(f"skipped {len(removed)} deletes, over the {MAX_DELETE_SHARE:.0%} guard")
        removed = []
    bad = len(pages) - len(done) + len(suspect)
    if bad > ALERT_SHARE * max(1, len(pages)):
        alerts.append(f"{bad} of {len(pages)} pages failed, were too short, or matched BLOCK_PAGE")
    for u in manifest:
        if u not in current and u not in removed:  # failed this time, or held back by the guard
            current[u] = manifest[u]

    (push_s3 if ROUTE == "s3" else push_custom)(changed, removed)
    save_json(MANIFEST_KEY, current)
    for alert in alerts:
        print("ALERT", alert)  # a CloudWatch metric filter on ALERT can notify someone
    summary = {"pages": len(pages), "indexed": len(current), "changed": len(changed), "removed": len(removed)}
    print(json.dumps(summary))
    return summary


def handler(event, context):
    if "crawl" in event:     # scheduled run, for example from Amazon EventBridge Scheduler
        return start_crawl(event["crawl"], event["callback_url"])
    if "fill" in event:      # a question that users of your app marked as unanswered
        return fill(event["fill"])
    if "crawl_id" in event:  # second, asynchronous invocation does the slow work
        return process(event["crawl_id"])
    raw = event.get("body") or "{}"
    if event.get("isBase64Encoded"):
        import base64
        raw = base64.b64decode(raw).decode()
    body = json.loads(raw)
    if not hmac.compare_digest(str(body.get("passthrough", "")), WEBHOOK_SECRET):
        return {"statusCode": 403}
    if body.get("status") in ("done", "truncated"):
        boto3.client("lambda").invoke(FunctionName=context.function_name, InvocationType="Event",
                                      Payload=json.dumps({"crawl_id": body["crawl_id"]}))
    else:
        print("ALERT crawl", body.get("crawl_id"), body.get("status"))
    return {"statusCode": 202}  # answer the webhook at once, before the pages are processed


def ask(question, max_age_days=None):
    search = {"numberOfResults": 3}
    if max_age_days:  # only pages a crawl found new or changed in the last N days
        since = int(time.time()) - int(max_age_days) * 86400
        search["filter"] = {"greaterThan": {"key": "changed_at", "value": since}}
    results = runtime.retrieve(
        knowledgeBaseId=KB_ID,
        retrievalQuery={"text": question},
        retrievalConfiguration={"managedSearchConfiguration": search},  # vectorSearchConfiguration is for customer-managed KBs
    )["retrievalResults"]
    for r in results:
        meta = r.get("metadata", {})  # other data sources in the same knowledge base use _source_uri
        print(meta.get("source_url") or meta.get("_source_uri"), round(r.get("score", 0), 3))
        print(r["content"]["text"][:300].replace("\n", " "), "\n")


def fill(question):
    """Add the page that search ranks first for a question, if the knowledge base doesn't have it yet."""
    if not ALLOWED_SITES:  # without a site limit, the question would go to search as is
        print("set ALLOWED_SITES first")
        return None
    sites = " OR ".join(f"site:{s}" for s in ALLOWED_SITES)
    organic = decodo("/search", {"query": f"{sites} {question}"})["organic"]
    links = (urllib.parse.urldefrag(r["link"])[0].rstrip("/") for r in organic)
    url = next((l for l in links if urllib.parse.urlparse(l).hostname in ALLOWED_SITES), None)
    gaps, manifest = load_json(GAPS_KEY, {}), load_json(MANIFEST_KEY, {})
    if url is None or url in gaps or url in manifest:
        print("no gap:", url)
        return None
    if disallowed(url):
        print("robots.txt disallows:", url)
        return None
    job = decodo("/crawl", {"url": url, "limit": 1, "sitemap": "skip", "output": "markdown", "JS_retry": True})
    deadline = time.time() + 120
    while decodo(f"/crawl/{job['crawl_id']}/status")["status"] == "pending" and time.time() < deadline:
        time.sleep(1)
    pages = fetch_pages(job["crawl_id"])  # fetched the same way as the full crawl, so the saved template matches
    page = pages[0] if pages else {}
    if page.get("status") != "done" or not page.get("content") or page["url"] in gaps or page["url"] in manifest:
        print("not added:", url)
        return None
    text = strip_template(page["content"], set(load_json(TEMPLATE_KEY, [])))
    if len(text) < MIN_CHARS or looks_blocked(text):
        print("too short or matched BLOCK_PAGE:", page["url"])
        return None
    digest = hashlib.sha256(text.encode()).hexdigest()
    (push_s3 if ROUTE == "s3" else push_custom)([(page["url"], text, attributes(page["url"], int(time.time()), digest))], [])
    gaps[page["url"]] = digest
    save_json(GAPS_KEY, gaps)
    print("added:", page["url"])
    return page["url"]


if __name__ == "__main__":
    command, *args = sys.argv[1:]
    {"map": map_site, "crawl": start_crawl, "process": process, "ask": ask, "fill": fill}[command](*args)

#!/bin/bash
# Creates the IAM roles, a managed knowledge base, a CUSTOM data source, and the webhook Lambda.
# Run it from the folder that holds web_to_bedrock.py and lambda-policy.json, with DECODO_API_KEY exported.
set -euo pipefail
REGION=us-east-1
export AWS_DEFAULT_REGION=$REGION        # also covers the IAM and STS calls
ACCOUNT=$(aws sts get-caller-identity --query Account --output text)
BUCKET=web-to-bedrock-$ACCOUNT          # holds the page manifest
SECRET=$(openssl rand -hex 24)

KB_ROLE_ARN=$(aws iam create-role --role-name WebDocsKBRole --query Role.Arn --output text \
  --assume-role-policy-document "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Service\":\"bedrock.amazonaws.com\"},\"Action\":\"sts:AssumeRole\",\"Condition\":{\"StringEquals\":{\"aws:SourceAccount\":\"$ACCOUNT\"}}}]}")
LAMBDA_ROLE_ARN=$(aws iam create-role --role-name WebToBedrockLambdaRole --query Role.Arn --output text \
  --assume-role-policy-document '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}]}')
sleep 15                                # new IAM roles take a few seconds to become usable

aws s3 mb "s3://$BUCKET" --region $REGION >/dev/null

KB_ID=$(aws bedrock-agent create-knowledge-base --region $REGION \
  --name web-docs --role-arn "$KB_ROLE_ARN" \
  --knowledge-base-configuration '{"type":"MANAGED","managedKnowledgeBaseConfiguration":{"embeddingModelType":"MANAGED"}}' \
  --query knowledgeBase.knowledgeBaseId --output text)
while [ "$(aws bedrock-agent get-knowledge-base --region $REGION --knowledge-base-id "$KB_ID" \
  --query knowledgeBase.status --output text)" = CREATING ]; do
  sleep 5                               # wait until the knowledge base can take a data source
done

DS_ID=$(aws bedrock-agent create-data-source --region $REGION \
  --knowledge-base-id "$KB_ID" --name web-pages \
  --data-source-configuration '{"type":"MANAGED_KNOWLEDGE_BASE_CONNECTOR","managedKnowledgeBaseConnectorConfiguration":{"connectorParameters":{"type":"CUSTOM","version":"1","aclEnabled":false}}}' \
  --query dataSource.dataSourceId --output text)
aws iam put-role-policy --role-name WebToBedrockLambdaRole --policy-name pipeline \
  --policy-document "$(sed -e "s/ACCOUNT_ID/$ACCOUNT/g" -e "s/KB_ID/$KB_ID/g" lambda-policy.json)"

zip -q function.zip web_to_bedrock.py
aws lambda create-function --region $REGION --function-name web-to-bedrock \
  --runtime python3.13 --handler web_to_bedrock.handler --role "$LAMBDA_ROLE_ARN" \
  --timeout 900 --memory-size 1024 --zip-file fileb://function.zip \
  --environment "Variables={DECODO_API_KEY=$DECODO_API_KEY,KB_ID=$KB_ID,DS_ID=$DS_ID,BUCKET=$BUCKET,WEBHOOK_SECRET=$SECRET}" >/dev/null
aws lambda wait function-active-v2 --region $REGION --function-name web-to-bedrock

URL=$(aws lambda create-function-url-config --region $REGION --function-name web-to-bedrock \
  --auth-type NONE --query FunctionUrl --output text)
aws lambda add-permission --region $REGION --function-name web-to-bedrock --statement-id url \
  --action lambda:InvokeFunctionUrl --principal "*" --function-url-auth-type NONE >/dev/null
# Function URLs need both permissions. --invoked-via-function-url limits this one to calls through the URL.
aws lambda add-permission --region $REGION --function-name web-to-bedrock --statement-id url-invoke \
  --action lambda:InvokeFunction --principal "*" --invoked-via-function-url >/dev/null

echo "export AWS_DEFAULT_REGION=$REGION KB_ID=$KB_ID DS_ID=$DS_ID BUCKET=$BUCKET WEBHOOK_SECRET=$SECRET FUNCTION_URL=$URL"

export DECODO_API_KEY="your-api-key"
python3 -m venv .venv && source .venv/bin/activate
pip install boto3
bash setup.sh > env.sh && source env.sh
python web_to_bedrock.py map https://kubernetes.io/docs/concepts/
python web_to_bedrock.py crawl https://kubernetes.io/docs/concepts/ "$FUNCTION_URL"
aws bedrock-agent list-knowledge-base-documents --knowledge-base-id "$KB_ID" --data-source-id "$DS_ID" --query "length(documentDetails[?status=='INDEXED'])"
python web_to_bedrock.py ask "How does maxSkew work in topology spread constraints?"
export ALLOWED_SITES=kubernetes.io
python web_to_bedrock.py ask "How do I safely drain a node before maintenance?"
python web_to_bedrock.py fill "How do I safely drain a node before maintenance?"
python web_to_bedrock.py ask "How do I safely drain a node before maintenance?"

{"pages": 191, "indexed": 190, "changed": 190, "removed": 0}

https://kubernetes.io/docs/concepts/scheduling-eviction/topology-spread-constraints 0.72
You can read more about this field by running `kubectl explain Pod.spec.topologySpreadConstraints` or refer to the [scheduling](/docs/reference/kubernetes-api/workload-resources/pod-v1/#scheduling) section of the API reference for Pod.  ### Spread constraint definition  You can define one or multipl

# 1st ask, first result
https://kubernetes.io/docs/concepts/cluster-administration/dra 0.831
# fill
added: https://kubernetes.io/docs/tasks/administer-cluster/safely-drain-node
# 2nd ask, first result
https://kubernetes.io/docs/tasks/administer-cluster/safely-drain-node 0.89