Laisser un agent IA choisir mes articles connexes

Je continue d'étendre la pipeline de mon blog avec de petites automatisations. J'ai corrigé mes articles avec Amazon Nova, et je les traduis automatiquement, et chaque étape fonctionne sans serveur et est déclenchée par un événement. Cette fois, j'ai visé quelque chose à quoi je pense depuis longtemps.

J'écris un article et quelqu'un le trouve, le lit jusqu'à la fin, puis part. Cette dernière partie est ce à quoi j'ai réfléchi. Un lecteur qui vient de terminer un article sur Lambda est probablement intéressé par lire mes autres articles sur le serverless. Je n'ai rien fait avec ce moment, pas de "lisez ceci ensuite". Rien. La lecture s'est terminée exactement au point où elle aurait dû continuer.

J'ai donc construit une pipeline qui corrige exactement cela, un agent IA qui lit chaque nouvel article, recherche tous mes articles de blog par signification et non par mots-clés, puis choisit trois articles connexes, et explique chaque choix en une phrase. La configuration est entièrement sans serveur, et entièrement déclenchée par des événements, à quoi vous attendiez-vous ? J'ai combiné Lambda Durable Functions, Strands Agents, Bedrock AgentCore Gateway, et MongoDB Atlas Vector Search en une seule pipeline.

Tout le code pour cet article peut être trouvé sur le manuel serverless comme d'habitude

Pourquoi ne pas simplement utiliser des étiquettes ?

Mon blog est du HTML statique et une solution facile serait d'utiliser des étiquettes, comme tant d'autres sites statiques. J'ai fait un test, et ce n'était tout simplement pas suffisant. J'utilise des étiquettes comme serverless qui sont plutôt larges, mais j'ai aussi quelques étiquettes plus étroites comme lambda. Mais ce n'est pas parce qu'un article est étiqueté serverless ou lambda qu'il ne partage rien avec d'autres articles. Les étiquettes répondent à "avec quels mots ai-je étiqueté ceci ?", pas à "de quoi cet article parle réellement ?"

L'étape suivante serait d'utiliser la similarité de contenu. Ici, je peux créer des embeddings vectoriels pour chaque article et retourner les trois voisins les plus proches, c'est fait ! Eh bien, oui, c'est l'épine dorsale de la pipeline, mais un top trois brut des voisins les plus proches donne souvent des articles presque identiques couvrant exactement le même sujet. Un correspondance pure des voisins les plus proches peut me dire que deux articles sont similaires, mais elle ne me dit pas si un lecteur qui a terminé l'un trouverait quelque chose d'intéressant dans l'autre.

J'ai donc étendu la conception, et le cœur de tout cela est que la recherche vectorielle récupère des données, et l'agent IA décide de ce qui est le plus pertinent. Je commence par réduire le nombre d'articles à l'aide de la recherche vectorielle, puis un agent IA, jouant le rôle de mon éditeur, lit les articles et choisit les trois qu'il pense que les lecteurs apprécieraient le plus. Il crée une justification en une phrase pour expliquer pourquoi il a choisi cet article. Ainsi, je peux examiner et affiner l'agent IA au fil du temps.

Aperçu de l'architecture

Lorsqu'un article est publié, la nouvelle pipeline de blog envoie un événement PostPublished à un bus EventBridge.

{
  "Source": "....",
  "DetailType": "PostPublished",
  "Detail": {
    "slug": "lambda-durable-functions-101",
    "language": "en",
    "branch": "main",
    "commit_sha": "f8a2c1d3..."
  }
}

Cela va maintenant déclencher une Lambda Durable Function, qui va coordonner l'ensemble du flux. Comme la coordination consiste principalement à exécuter des étapes de code, j'ai décidé d'essayer les Durable Functions au lieu des Step Functions, qui ont été mon choix habituel dans le passé. Avec les Durable Functions, j'obtiens un outil supplémentaire dans ma boîte à outils.

La pipeline se présente ainsi : elle récupère le markdown de l'article depuis GitHub, appelle Bedrock pour créer un embedding à l'aide de Amazon Titan Text Embeddings V2, et l'insère dans une collection MongoDB Atlas. Ensuite, l'agent IA est appelé pour choisir les trois articles connexes, en commençant par une recherche vectorielle dans MongoDB Atlas à l'aide d'un outil MCP hébergé sur AgentCore Gateway. Ensuite, les références des articles connexes sont ajoutées à DSQL, qui fait partie de mon système CMS maison, et enfin, cela met à jour le frontmatter de l'article dans GitHub et ouvre une PR.

J'ai également dû m'assurer que les embeddings étaient créés pour tous les articles actuels ; sinon, cela ne fonctionnerait pas pour les nouveaux articles. J'ai donc créé une petite configuration de rembourrage unique.

C'est beaucoup de pièces mobiles pour une seule fonction. Je reviendrai sur la raison pour laquelle cela fonctionne.

Du Markdown aux embeddings

La pipeline entière repose sur une transformation, transformer l'article de blog en embeddings afin que nous puissions trouver de manière programmable des articles similaires.

Un modèle d'embedding mappe du texte à un vecteur fixe de flottants. Dans mon cas, 1024 flottants de Titan Text Embeddings v2 de Bedrock. Le modèle a été entraîné pour que les textes traitant des mêmes concepts se trouvent proches les uns des autres dans cet espace de 1024 dimensions, même s'ils ne partagent aucun vocabulaire. Mon article sur les Durable Functions et mon article sur les Step Functions finissent par être des voisins parce qu'ils traitent tous deux de l'orchestration de workflows sur AWS.

Ce qui entre dans l'embedding compte beaucoup, tout comme le modèle. Je n'embed pas l'intégralité du markdown brut ; je commence par une petite étape de normalisation qui supprime le frontmatter YAML, supprime chaque exemple de code, et supprime les images, puis crée une nouvelle structure qui peut être alimentée dans le modèle.

TITLE: Lambda Durable Functions 101
TAGS: aws, lambda, serverless
SUMMARY: Un guide pratique pour construire un workflow durable sur AWS Lambda.
BODY: Le corps nettoyé

Alors pourquoi supprimer tout le code ? Eh bien, les blocs de code ne sont que du "bruit". Deux articles sur le même sujet peuvent avoir des exemples de code complètement différents, donc si j'incluais le code, ces articles se retrouveraient maintenant plus éloignés les uns des autres dans l'espace vectoriel. Le code ne dit rien sur la similarité, donc je n'inclus que le contenu réel du blog.

L'appel à Bedrock pour créer l'embedding est assez petit.

_MODEL_ID = "amazon.titan-embed-text-v2:0"

def embed_text(text: str, dimensions: int = 1024) -> list[float]:
    response = _bedrock_client().invoke_model(
        modelId=_MODEL_ID,
        contentType="application/json",
        accept="application/json",
        body=json.dumps({
            "inputText": text,
            "dimensions": dimensions,
            "normalize": True,
        }),
    )
    payload = json.loads(response["body"].read())
    return [float(x) for x in payload["embedding"]]

Un indicateur que nous devons examiner de plus près est normalize: True. Considérez chaque embedding comme une flèche pointant quelque part dans l'espace. Les articles sur le même sujet pointent dans à peu près la même direction. Lors de la comparaison de deux articles, c'est la direction qui compte, pas la longueur des flèches. La recette standard pour cela est la similarité cosinus : multipliez les deux vecteurs ensemble, puis divisez par la longueur de chaque flèche. Cette division n'est là que pour annuler les longueurs. Avec normalize: True, Titan ajuste chaque vecteur à une longueur exactement de 1 avant de le retourner. Et diviser par 1 ne change rien. Ainsi, la comparaison se réduit à simplement l'étape de multiplication, un produit scalaire simple. Même résultat, moins de mathématiques, et cela s'additionne lorsque chaque nouvel article est comparé à l'intégralité du catalogue arrière.

L'entrée de l'embed est limitée à 30 000 caractères avant cet appel, donc les articles très longs sont tronqués au lieu de échouer (Titan v2 accepte environ 8 000 jetons).

En même temps, un hachage de contenu SHA-256 est créé et stocké avec le vecteur, avec cela je peux vérifier si les anciens articles ont changé et doivent être mis à jour.

new_hash = compute_hash(markdown)  # sha256 sur le corps normalisé
existing = context.step(
    lambda _: find_by_id(slug, language),
    name="find_by_id",
)
if not force_recompute and existing and existing.get("content_hash") == new_hash:
    return {"skipped": True, "reason": "content_hash_unchanged"}

Stocker les vecteurs dans MongoDB Atlas

Les vecteurs créés doivent être stockés quelque part où ils peuvent être interrogés correctement. J'ai exploré quelques options différentes comme OpenSearch serverless et Aurora avec pgvector. Mais finalement j'ai décidé d'exécuter cela avec MongoDB Atlas, ce qui semblait être le meilleur choix pour moi. Et comme je peux exécuter un cluster M0 sous l'offre gratuite de MongoDB, c'était un bonus supplémentaire.

MongoDB Atlas Vector Search exécute HNSW sous les hoods, le même algorithme de voisin le plus proche approché derrière la plupart des bases de données vectorielles de production. La requête est une étape d'agrégation $vectorSearch, ce qui signifie que l'index vectoriel et les documents vivent dans un seul système. Une requête retourne les voisins avec leurs titres, résumés et étiquettes. Exactement les données dont l'agent a besoin, pas de deuxième recherche, pas de base de données vectorielle séparée à synchroniser.

pipeline = [
    {"$vectorSearch": {
        "index": "posts_vector_idx",
        "path": "embedding",
        "queryVector": embedding,           # 1024 dims, from Titan
        "numCandidates": max(100, k * 10),
        "limit": k,
        "filter": {"language": language},   # évalué PENDANT la traversée
    }},
    {"$match": {"slug": {"$nin": exclude_slugs}}},
    {"$project": {
        "_id": 0,
        "slug": 1, "language": 1, "title": 1,
        "summary": 1, "tags": 1, "category": 1,
        "score": {"$meta": "vectorSearchScore"},
    }},
]
return list(_collection().aggregate(pipeline))

Cette ligne filter est très importante pour moi et mes blogs. Mes articles existent en plusieurs langues, et le voisin le plus proche de n'importe quel article est de manière fiable sa propre traduction. Atlas me permet de déclarer language comme un champ de filtre à l'intérieur de l'index vectoriel, donc la traversée HNSW ne considère jamais que les vecteurs de la même langue.

Pour l'authentification, j'utilise AWS IAM, pas de mots de passe nulle part. MongoDB Atlas accepte les principes AWS IAM comme utilisateurs de base de données via le mécanisme MONGODB-AWS de pymongo. Mes fonctions Lambda assument un rôle IAM dédié via STS, un rôle qui a été fédéré avec Atlas exactement une fois, et les identifiants temporaires vont directement dans la connexion MongoDB.

def _assume_atlas_role() -> dict:
    sts = boto3.client("sts")
    response = sts.assume_role(
        RoleArn=os.environ["ATLAS_ROLE_ARN"],
        RoleSessionName="related-posts-atlas",
        DurationSeconds=3600,
    )
    return response["Credentials"]


def _client() -> MongoClient:
    creds = _assume_atlas_role()
    return MongoClient(
        _connection_config()["srvUri"],   # ...authMechanism=MONGODB-AWS
        username=creds["AccessKeyId"],
        password=creds["SecretAccessKey"],
        authMechanismProperties={"AWS_SESSION_TOKEN": creds["SessionToken"]},
    )

Pourquoi utiliser un rôle dédié au lieu du rôle d'exécution propre à Lambda ? C'est essentiellement pour deux raisons : tout d'abord, je veux pouvoir gérer la configuration de MongoDB Atlas en un seul flux. Je ne veux pas configurer une partie d'Atlas, puis devoir déployer des fonctions Lambda, récupérer les ARNs du rôle, et mettre à jour Atlas. En séparant cela, je peux également créer un rôle que j'autorise vers Atlas, puis plusieurs fonctions Lambda peuvent assumer le même rôle et je peux contrôler cet accès correctement avec les autorisations IAM.

Si vous voulez aller plus loin sur l'authentification sans mot de passe vers Atlas, j'ai écrit sur Fédération d'identité sortante avec MongoDB récemment, ce qui est une technique différente de celle que j'utilise ici.

Éditeur de contenu IA avec des outils

C'est ici que la partie IA s'aiguise. L'agent n'est pas "appeler un LLM avec une invite". C'est une boucle. Claude Sonnet 4.6 sur Bedrock est appelé avec une invite système pour trouver des articles connexes, deux outils, et il décide quoi appeler, quoi lire, et quand il a terminé.

Les deux outils sont de simples fonctions Lambda gérées comme des outils MCP dans AgentCore Gateway. vector_search retourne les top-k candidats avec titres, résumés, étiquettes et scores de similarité. read_post_excerpt retourne les premiers 1000 caractères du corps d'un article, et l'agent l'appelle uniquement lorsque deux candidats semblent interchangeables à partir de leurs résumés et qu'il veut rompre le lien en lisant réellement.

Un détail de conception délibéré dans vector_search est que l'agent passe le slug de l'article, jamais un embedding. La fonction Lambda de l'outil résout le vecteur stocké depuis Atlas lui-même, donc le modèle ne peut pas halluciner un tableau de 1024 flottants malformé dans la limite de l'index.

Comme mentionné, les outils sont devant Bedrock AgentCore Gateway. La passerelle transforme mes fonctions Lambda en un catalogue d'outils que tout agent conscient MCP peut découvrir et appeler, gère le protocole afin que les Lambdas restent sans protocole, et authentifie les appels entrants avec IAM SigV4 simple, pas de serveur OAuth à exécuter. Pourrais-je avoir sauté la passerelle et passé les outils comme des fonctions Python inline ? Bien sûr. Mais le prochain agent que je construirai (et il y en aura un) pourra réutiliser le même catalogue d'outils sans que je doive copier du code.

La passerelle et ses cibles sont de simples CloudFormation. Chaque cible mappe une Lambda à un nom d'outil plus un schéma d'entrée, et ce schéma est ce que le modèle voit lorsqu'il décide comment appeler l'outil.

RelatedPostsGateway:
  Type: AWS::BedrockAgentCore::Gateway
  Properties:
    Name: !Sub ${Application}-related-posts-gw
    ProtocolType: MCP
    RoleArn: !GetAtt GatewayExecutionRole.Arn
    AuthorizerType: AWS_IAM

VectorSearchTarget:
  Type: AWS::BedrockAgentCore::GatewayTarget
  Properties:
    GatewayIdentifier: !Ref RelatedPostsGateway
    Name: vector-search-target
    CredentialProviderConfigurations:
      - CredentialProviderType: GATEWAY_IAM_ROLE
    TargetConfiguration:
      Mcp:
        Lambda:
          LambdaArn: !GetAtt VectorSearchToolFunction.Arn
          ToolSchema:
            InlinePayload:
              - Name: vector_search
                Description: >
                  Recherche vectorielle sur le corpus des articles de blog. Résout l'embedding
                  stocké de l'article source depuis Atlas par source_slug, puis retourne
                  jusqu'à k candidats par similarité sémantique.
                InputSchema:
                  Type: object
                  Properties:
                    source_slug:
                      Type: string
                    language:
                      Type: string
                    k:
                      Type: integer
                    exclude_slugs:
                      Type: array
                      Items:
                        Type: string
                  Required:
                    - source_slug

Et la fonction Lambda de l'outil ne sait rien de MCP. La passerelle passe l'entrée de l'outil comme l'événement, et la valeur de retour devient la sortie de l'outil.

def handler(event: dict, context) -> dict:
    source_slug = event["source_slug"]
    language = event.get("language") or "en"

    embedding = find_embedding(source_slug, language)
    candidates = vector_search_neighbors(
        embedding=embedding,
        k=int(event.get("k", 20)),
        language=language,
        exclude_slugs=event.get("exclude_slugs"),
    )
    return {"candidates": candidates, "count": len(candidates)}

Strands Agents est le cadre d'agent open-source d'AWS, il pilote la boucle agentique afin que je n'aie pas à l'implémenter moi-même.

from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client
from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp.mcp_client import MCPClient

mcp_client = MCPClient(
    lambda: aws_iam_streamablehttp_client(
        endpoint=os.environ["AGENTCORE_GATEWAY_URL"],
        aws_service="bedrock-agentcore",
    )
)
model = BedrockModel(
    model_id="eu.anthropic.claude-sonnet-4-6",  # Profil d'inférence cross-région EU
    temperature=0.0,
    streaming=False,
)

with mcp_client:
    tools = mcp_client.list_tools_sync()
    agent = Agent(
        model=model,
        tools=tools,
        system_prompt=_SYSTEM_PROMPT,
        callback_handler=_BoundedToolCallHandler(max_calls=8),
    )
    result = agent(
        f"source_slug: {source_slug}\n"
        f"source_title: {source_title}\n"
        f"source_summary: {source_summary}\n"
        f"language: {language}\n\n"
        "Choisis 3 articles connexes."
    )

Le comportement du modèle est façonné presque entièrement par l'invite système. Persona, conseils anti-schémas, et un contrat strict.

Vous êtes l'éditeur de jimmydqv.com, un blog technique sur AWS, serverless, et IA.
Votre travail est de choisir exactement 3 articles connexes qu'un lecteur apprécierait le plus après
avoir terminé l'article actuel.
Privilégiez la profondeur thématique plutôt que la superposition de mots-clés de surface.
Chaque choix nécessite une justification en une phrase qui nomme la connexion spécifique.

Flux de travail :
1. Appelez vector_search avec source_slug=, language=, k=20,
   exclude_slugs=[]. L'outil résoudra l'embedding depuis Atlas
   internement, ne tentez PAS de construire un embedding vous-même.
2. Si deux candidats semblent interchangeables, appelez éventuellement read_post_excerpt sur l'un
   pour rompre le lien.
3. Retournez STRICTEMENT cette forme JSON et rien d'autre :
   {"picks": [{"slug": "...", "rationale": "..."}, ... exactement 3 éléments ...]}

parsed = json.loads(_strip_fences(raw))
picks_raw = parsed.get("picks", [])
if len(picks_raw) != 3:
    raise ValueError(f"L'agent doit retourner exactement 3 choix, obtenu {len(picks_raw)}")

Alors, pourquoi ai-je opté pour le modèle Sonnet, quelque peu plus cher, et pas le plus mince Haiku ou même Nova 2 ? J'ai testé plusieurs modèles différents. Haiku et Nova 2 Light sont tous deux plus rapides et moins chers mais ne retournaient pas de manière fiable trois bons choix ; parfois j'obtenais deux, parfois quatre, parfois vraiment de mauvais choix. Sonnet produisait les choix le plus consistamment, et un appel prend deux à trois secondes.

Une exécution typique : un vector_search, occasionnellement une lecture d'extrait, puis trois choix avec des justifications comme "Les deux parcourent la boucle des outils d'agent sur Bedrock ; cet article se concentre sur l'orchestration durable autour de celui-ci."

Orchestration avec Lambda Durable Functions

La pipeline prend 30 à 90 secondes pour s'exécuter, elle fait environ six appels LLM, parle à quatre systèmes externes. La réponse classique est Step Functions, et j'ai utilisé Step Functions pour l'orchestration de nombreuses fois. Cette fois, j'ai opté pour Lambda Durable Functions à la place, et pour un travail d'agent, je ferais le même choix à nouveau.

Le modèle est simple. L'intégralité du flux de travail est un Python ordinaire dans un seul gestionnaire, et chaque effet secondaire est enveloppé dans context.step(...). Le résultat de chaque étape est vérifié. Si l'invocation plante, Bedrock limite, time-out, quoi que ce soit, une nouvelle invocation re-exécute le gestionnaire depuis le début, mais les étapes terminées rejouent depuis leurs vérifications au lieu de ré-exécuter.

Transformer une Lambda régulière en fonction durable est une propriété SAM. L'orchestrateur est déclaré comme ceci.

RelatedPostsOrchestrator:
  Type: AWS::Serverless::Function
  Properties:
    CodeUri: lambda/orchestrator
    Handler: handler.handler
    Runtime: python3.13
    Timeout: 900
    MemorySize: 1024
    AutoPublishAlias: live
    DurableConfig:
      ExecutionTimeout: 3600
      RetentionPeriodInDays: 7

Le gestionnaire lui-même ressemble à une liste d'étapes.

@durable_execution
def handler(event: dict, context: DurableContext) -> dict:
    slug = event["slug"]
    language = event.get("language", "en")

    # run_id est non déterministe ; calculez-le à l'intérieur d'une étape
    # afin que les rejoues réutilisent la même valeur.
    run_id = context.step(lambda _: str(uuid.uuid4()), name="generate_run_id")

    file_path = context.step(
        lambda _: lookup_post_file_path(slug, language),
        name="lookup_file_path",
    )
    fetched = context.step(
        lambda _: invoke_fetch_source(file_path, branch, git_ref),
        name="fetch_source",
    )
    # ... contournement du hachage de contenu ...
    embedding = context.step(
        lambda _: embed_text(embed_input),
        name="embed_post",
    )
    context.step(lambda _: upsert_post(doc), name="upsert_post")

    primary_picks = context.run_in_child_context(
        lambda child_ctx: child_ctx.step(
            lambda _: _picks_for_one_post(slug, doc["title"], doc["summary"], language),
            name="agent_pick_primary",
        ),
        name="primary_agent_picks",
    )
    if len(primary_picks) != 3:
        raise ExecutionError(
            f"L'agent a retourné {len(primary_picks)} choix, il en faut exactement 3"
        )
    # ... fan-out, persistance, PR ...

Remarquez le run_id. Même un uuid.uuid4() doit vivre à l'intérieur d'une étape, car une rejoue générerait autrement un identifiant différent et dévierait de la trajectoire de vérification. L'appel à l'agent s'exécute dans un contexte enfant puisqu'il orchestre plusieurs invocations d'outils MCP en arrière-plan, et le contexte enfant garde l'arbre d'étapes ordonné.

Ce que le modèle de rejoue apporte ici est concret : lorsqu'un appel à l'agent est limité, la réessai ne récupère pas depuis GitHub, ne ré-embede pas, ou ne ré-insère pas. Le travail coûteux précédent rejoue gratuitement. Et la logique de l'agent elle-même, branchement, analyse JSON, "re-exécuter pour cinq voisins", est exactement le genre de code qui vous combat dans une machine d'état JSON et se lit naturellement en Python. Le fan-out est quelques lignes.

batch = context.parallel(
    [_make_backlink_fn(n) for n in backlink_targets],
    name="backlink_fan_out",
    config=ParallelConfig(max_concurrency=5),
)
# Une seule jambe échouée ne devrait pas échouer la pipeline ; gardez ce qui a réussi.
for item in batch.succeeded():
    result = item.result
    if result and len(result.get("picks") or []) == 3:
        backlink_results.append(result)

Les liens d'articles connexes en arrière

C'est une partie intéressante de la solution et de la pipeline.

Trouver les articles connexes pour un nouvel article est la partie facile et très directe. La partie plus difficile est : que se passe-t-il si j'ajoute un nouvel article qui est maintenant un meilleur choix comme article connexe pour un ancien article ? Cela signifierait que je devrais mettre à jour les anciens articles également lorsqu'un nouveau est ajouté, en le liant en arrière. Parce que si je ne le fais pas, un article que j'ai écrit il y a un an resterait "gelé" pour toujours.

Donc après avoir embeddé le nouvel article, l'orchestrateur demande à Atlas ses dix voisins les plus proches et réexécute l'agent pour les cinq premiers, chacun en tant qu'étape durable indépendante et parallèle. L'exécution de l'éditeur de chaque voisin pose la question complète depuis le début : donné le blog tel qu'il existe maintenant, quels sont les trois meilleurs ? Parfois le nouvel article déplace un ancien choix. Parfois rien ne change.

Pas de travail par lot nocturne, pas de cron. Le travail se produit exactement lorsque l'article de blog change, limité au voisinage qui a changé.

Atterrir les choix

La sortie de l'agent doit aboutir à deux endroits, car le blog est un site statique 11ty.

Amazon DSQL est la source de vérité. Une ligne par choix, écrite comme DELETE puis INSERT dans une seule transaction, donc les réexécutions sont idempotentes et une écriture échouée laisse les anciens choix intacts. J'utilise la même configuration DSQL que dans ma série AI Bartender, authentification IAM et tout.

CREATE TABLE cms_content.related_posts (
    source_slug      TEXT NOT NULL,
    language         TEXT NOT NULL,
    position         SMALLINT NOT NULL,
    related_slug     TEXT NOT NULL,
    rationale        TEXT NOT NULL,
    similarity_score DOUBLE PRECISION,
    run_id           UUID NOT NULL,
    generated_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (source_slug, language, position),
    CHECK (position BETWEEN 1 AND 3)
);

L'écriture s'exécute sur une seule connexion, et le DELETE se rétablit ensemble avec une INSERT échouée, donc la table n'est jamais à moitié écrite.

conn = connector.get_connection(token, user)
try:
    with conn.cursor() as cur:
        cur.execute(
            "DELETE FROM cms_content.related_posts "
            "WHERE source_slug = %(s)s AND language = %(l)s",
            {"s": source_slug, "l": language},
        )
        for position, pick in enumerate(picks, start=1):
            cur.execute(
                "INSERT INTO cms_content.related_posts "
                "(source_slug, language, position, related_slug, "
                " rationale, similarity_score, run_id) "
                "VALUES (%(src)s, %(lang)s, %(pos)s, %(rel)s, "
                "        %(rat)s, %(sim)s, %(run)s)",
                {...},
            )
    conn.commit()
except Exception:
    conn.rollback()
    raise

Mon tableau de bord CMS lit cette table et affiche les choix, avec les justifications, à côté de chaque article. La dernière étape de l'orchestrateur émet un événement terminé qui passe par EventBridge vers AppSync Events, donc le tableau de bord se met à jour en temps réel lorsqu'une exécution se termine. Pas de sondage nulle part.

Une PR GitHub le rend visible sur le blog. La pipeline injecte un bloc related_posts: dans le frontmatter des articles, jusqu'à six fichiers dans une seule PR (le nouvel article plus les voisins actualisés). La prochaine construction du site rend la section.

related_posts:
  - slug: "step-functions-vs-durable-functions"
    rationale: "Comparaison directe de l'approche d'orchestration utilisée ici contre l'alternative SFN."

Pourquoi une PR au lieu de commuter directement sur main ? C'est ainsi que toutes mes différentes pipelines d'amélioration le font, cela me donne en tant qu'humain une dernière vérification avant qu'il ne soit en production.

Résultat final

À la fin, la pipeline ajoutera les articles connexes dans le frontmatter, et le processus de construction 11ty les récupérera, cherchera l'image de couverture et la description, et injectera une section dans l'article construit.

Dernières paroles

Je suis parti pour garder les lecteurs en lecture, et j'ai fini par quelque chose que je trouve très intéressant, un modèle. La recherche vectorielle fait le rappel, bon marché, rapide, mathématiquement honnête sur la similarité. Un agent fait le jugement, plus lent, mais capable de lire deux candidats et de décider lequel un humain voudrait réellement ensuite, et de dire pourquoi. L'exécution durable enveloppe le tout ; un flux de travail de six appels LLM peut être traité aussi facilement qu'une seule fonction. Et MongoDB Atlas fait silencieusement ce dont chaque pipeline IA a besoin : il garde les vecteurs, les métadonnées, et le filtrage en un seul endroit interrogable, gratuitement, avec l'identité IAM au lieu des mots de passe.

Découvrez mes autres articles sur jimmydqv.com et suivez-moi sur LinkedIn et X pour plus de contenu serverless.

Tout le code pour cet article peut être trouvé sur le manuel serverless comme d'habitude

Maintenant, allez construire !