首页 / 文章 / 从第一性原理到三大核心智能体：RAG技术

rag embeddings bedrock langgraph retrieval aws

从第一性原理到三大核心智能体：RAG技术

为什么权重在私有数据上会失效，嵌入与余弦检索的原理是什么，以及从托管知识库到自主数据采集的三种渐进式智能体。

4464 词

冻结后的基础模型仅凭权重无法回答企业内部的业务问题。训练过程中从未接触过内部的人力资源政策、合同或操作手册；时间节点的冻结限制了信息更新；即便缺乏事实依据，生成器仍会输出看似合理的猜测。因此，要让模型掌握企业的知识，这其实是一个系统层面的问题，而非单纯增加检查点就能解决的。

为何仅靠权重行不通

三种结构性限制阻碍了企业级问答系统的应用：

内部数据从未被用于训练。 保留规则和工单模板并非公开的网络文本。
时间在冻结点后停止流动。 上个月修订的法律条款对已冻结的模型来说是不可见的。
沉默并非默认状态。 下一个词元的预测更倾向于选择看似合理的延续内容，而非诚实地表示未知。

三种注入知识的方法——以及哪种最有效

选项1：在语料库上微调

进一步训练虽可植入特定风格或技能，但会无依据地模糊事实信息，且由于所有人使用相同的权重，无法实现按用户显示不同内容。对于需要引用来源并控制访问权限的政策问答场景，微调并非合适的方法。

选项2：将所有内容直接放入系统指令中

在提问时直接提供文本，可将检索结果转化为阅读内容——这才是正确的机制。对于规模较小的稳定模型而言此方法有效，但当数据量超出处理范围、每次提问都会对每个词元计费、答案被埋在长提示语中导致质量下降，以及单一静态指令无法按角色进行过滤时，该方法就会失效。

选项3：仅粘贴关键内容

保留“提问时读取”的机制，同时解决规模问题。只需获取少量相关段落后再进行生成，这便是用一句话概括的检索增强生成技术。

RAG的完整工作流程

离线阶段：导入文件，将其分割成多个片段，嵌入这些片段，并存储向量（通常还包括词汇索引）。在线阶段：将问题嵌入其中，获取最相关的片段，将这些片段与问题一起构建成指令，再基于这些片段生成答案。

最困难的部分并非最后的生成步骤——而是如何实现精准、高效且具备权限意识的片段获取。

摒弃神秘色彩的向量与嵌入

这里的向量是一组数字，用于将文本片段定位到一个空间中，在该空间里语义相近的元素会彼此靠近。嵌入网络负责将文本映射到这个空间中。余弦相似度更注重方向而非长度，因此被广泛用于检索评分。而那些忽略归一化的距离度量方法，在片段长度不均匀的情况下表现会更差。

选择维度数和嵌入器属于产品层面的决策：更高的维度能更好地区分不同含义，但会增加存储成本和查询延迟；在索引阶段与查询阶段使用不同的嵌入器会无形中破坏最近邻搜索功能。

相似度计算

余弦值是单位向量的点积。即使关于密码重置的描述措辞不同，只要嵌入器能够捕捉到其意图，这两个内容仍会处于相近位置。关键词搜索无法发现这种关联，而密集检索可以——但除非结合混合词法搜索，否则可能无法识别出精确的SKU标识。

在托管云推理平台上构建三个完整的智能体

本指南的其余部分将这些理念具体化为 AWS Bedrock 风格托管模型上的三种渐进式智能体：精心构建的知识库、LangGraph 编排器以及自主管理的数据导入流程。每部分内容都是完整的智能体设计框架——应将其视为可执行的架构，而非零散代码片段。

智能体 1 — 托管知识库路径

"""
AGENT 1: Managed Knowledge Base. The 30-line version.
Bedrock does the chunking, embedding, indexing and searching. You wire it up.
"""
import boto3
from langchain_aws import AmazonKnowledgeBasesRetriever, ChatBedrockConverse
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

REGION = "eu-west-1"
KB_ID = "ABCD1234EF"

retriever = AmazonKnowledgeBasesRetriever(
    knowledge_base_id=KB_ID,
    region_name=REGION,
    retrieval_config={
        "vectorSearchConfiguration": {
            "numberOfResults": 5,
            "overrideSearchType": "HYBRID",
        }
    },
)

llm = ChatBedrockConverse(
    model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name=REGION,
    temperature=0,
    max_tokens=1024,
)

PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "Answer only from the numbered context. Cite every claim as [n]. "
     "If the context does not contain the answer, say so plainly."),
    ("human", "Context:\n{context}\n\nQuestion: {question}"),
])


def format_docs(docs) -> str:
    return "\n\n".join(
        f"[{i}] source={d.metadata.get('location', {}).get('s3Location', {}).get('uri', '?')}\n"
        f"{d.page_content}"
        for i, d in enumerate(docs, start=1)
    )


chain = (
    RunnableParallel(context=retriever | format_docs, question=RunnablePassthrough())
    | PROMPT
    | llm
    | StrOutputParser()
)

if __name__ == "__main__":
    print(chain.invoke("How fast must we submit a shortlist for a niche engineering role?"))

该路径依赖于托管式知识库：文档存储在相应位置，托管索引器负责生成嵌入向量，智能体则通过该服务返回的引用信息来给出答案。其优势在于无需编写自定义索引代码；缺点则是对数据分段边界及融合处理的控制力较弱。

智能体 2 — 基于 LangGraph 的工具编排

"""
AGENT 2: LangGraph agent that decides WHETHER and WHICH index to search.
Retrieval is a tool. Identity travels in typed context, never in the prompt.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any

import boto3
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
from langchain.tools import ToolRuntime
from langchain_aws import ChatBedrockConverse
from langchain_core.tools import tool

REGION = "eu-west-1"
POLICY_KB = "ABCD1234EF"
ENGINEERING_KB = "WXYZ5678GH"

agent_runtime = boto3.client("bedrock-agent-runtime", region_name=REGION)


@dataclass
class RequestContext:
    """Built from validated JWT claims. The model cannot read or forge it."""
    subject: str
    roles: tuple[str, ...]
    region: str


def _retrieve(kb_id: str, query: str, ctx: RequestContext, k: int = 5) -> str:
    """Bedrock Retrieve API with a server-side metadata filter for tenant + residency."""
    resp = agent_runtime.retrieve(
        knowledgeBaseId=kb_id,
        retrievalQuery={"text": query},
        retrievalConfiguration={
            "vectorSearchConfiguration": {
                "numberOfResults": k,
                "overrideSearchType": "HYBRID",
                "filter": {
                    "andAll": [
                        {"in": {"key": "acl_roles", "value": list(ctx.roles)}},
                        {"in": {"key": "region", "value": [ctx.region, "GLOBAL"]}},
                    ]
                },
            }
        },
    )
    results: list[dict[str, Any]] = resp.get("retrievalResults", [])
    if not results:
        return "NO_AUTHORISED_RESULTS: nothing in this index that you may read."
    return "\n\n".join(
        f"[{i}] score={r.get('score'):.3f} "
        f"source={r.get('location', {}).get('s3Location', {}).get('uri', '?')}\n"
        f"{r['content']['text']}"
        for i, r in enumerate(results, start=1)
    )


@tool
def search_policy_docs(query: str, runtime: ToolRuntime[RequestContext]) -> str:
    """Search HR, compliance and commercial policy.

    Use for shortlist SLAs and service credits, candidate data retention and consent,
    GDPR erasure, statement-of-work approval thresholds, background check rules and
    automated decision governance.
    """
    return _retrieve(POLICY_KB, query, runtime.context)


@tool
def search_engineering_docs(query: str, runtime: ToolRuntime[RequestContext]) -> str:
    """Search platform runbooks and integration documentation.

    Use for the requisition sync service, Workday and SuccessFactors webhooks, Kafka
    topics and event schemas, idempotency and retry behaviour, and vector index refresh.
    """
    return _retrieve(ENGINEERING_KB, query, runtime.context)


agent = create_agent(
    model=ChatBedrockConverse(
        model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
        region_name=REGION,
        temperature=0,
    ),
    tools=[search_policy_docs, search_engineering_docs],
    system_prompt=(
        "You are an enterprise knowledge assistant.\n"
        "- Answer only from tool results and cite the [n] markers they return.\n"
        "- If a tool returns NO_AUTHORISED_RESULTS, say you cannot access a source and "
        "name the document owner. Never fill the gap from memory.\n"
        "- For questions spanning policy and platform, call both tools before answering."
    ),
    context_schema=RequestContext,
    middleware=[
        ModelCallLimitMiddleware(thread_limit=8, run_limit=6, exit_behavior="end"),
        ToolCallLimitMiddleware(thread_limit=6, run_limit=5, exit_behavior="end"),
    ],
)

if __name__ == "__main__":
    claims = {"sub": "asha@example.com", "roles": ["recruiter"], "region": "EU"}
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "How is the vector index refreshed?"}]},
        context=RequestContext(
            subject=claims["sub"],
            roles=tuple(claims["roles"]),
            region=claims["region"],
        ),
    )
    print(result["messages"][-1].content)

在此模式下，控制流程表现为一个明确的图形结构：先对意图进行分类，调用检索工具，必要时询问人工，最后给出响应。通过检查点机制和中断功能，长时间运行的支持线程会被转化为持久化的会话，而非仅存在于内存中的字典结构。

Agent 3 — 自主管理的数据摄取与获取

"""
AGENT 3: Self-managed pipeline: your chunking, your index, your fusion, your rerank.
Chosen when Bedrock KB's fixed options are not enough (custom fusion, per-stage timing,
an abstention gate, or a chunking rule the managed service will not express).
"""
from __future__ import annotations
import time
from typing import Annotated, Any, TypedDict

import boto3
from langchain_aws import BedrockEmbeddings, ChatBedrockConverse
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
from langgraph.graph import END, START, StateGraph

REGION = "eu-west-1"
CONFIDENCE_FLOOR = 0.34
CANDIDATE_K, CONTEXT_K = 20, 4

embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v2:0",
    region_name=REGION,
    model_kwargs={"dimensions": 1024, "normalize": True},
)
llm = ChatBedrockConverse(
    model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name=REGION,
    temperature=0,
)
rerank_client = boto3.client("bedrock-agent-runtime", region_name=REGION)


# ---------------------------------------------------------------- ingest
def chunk_policy_document(markdown: str, source: str, acl: list[str], region: str):
    """Structure-aware first, size-capped second. Headings are the natural unit for
    policy text; the recursive splitter only intervenes on oversized sections."""
    by_heading = MarkdownHeaderTextSplitter(
        headers_to_split_on=[("#", "title"), ("##", "section"), ("###", "clause")]
    ).split_text(markdown)

    capper = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)
    chunks: list[Document] = []
    for part in capper.split_documents(by_heading):
        heading = " > ".join(
            v for k, v in part.metadata.items() if k in ("title", "section", "clause")
        )
        chunks.append(
            Document(
                # contextual retrieval: the chunk carries its own scope
                page_content=f"{heading}\n{part.page_content}",
                metadata={**part.metadata, "source": source, "acl": acl, "region": region},
            )
        )
    return chunks


# ---------------------------------------------------------------- state
class State(TypedDict, total=False):
    question: str
    roles: list[str]
    region: str
    candidates: list[Document]
    context_docs: list[Document]
    confidence: float
    answer: str
    timings: Annotated[dict[str, float], lambda a, b: {**(a or {}), **(b or {})}]


def _ms(name: str, t0: float) -> dict[str, float]:
    return {name: round((time.perf_counter() - t0) * 1000, 1)}


# ---------------------------------------------------------------- nodes
def retrieve(state: State, *, store) -> State:
    """Hybrid search with an ACL pre-filter pushed into the index query."""
    t0 = time.perf_counter()
    docs = store.similarity_search(
        state["question"],
        k=CANDIDATE_K,
        filter={"acl": {"$in": state["roles"]}, "region": {"$in": [state["region"], "GLOBAL"]}},
    )
    return {"candidates": docs, "timings": _ms("retrieve", t0)}


def rerank(state: State) -> State:
    """Bedrock Rerank: a cross-encoder scoring each (query, chunk) pair jointly."""
    t0 = time.perf_counter()
    cands = state["candidates"]
    if not cands:
        return {"context_docs": [], "confidence": 0.0, "timings": _ms("rerank", t0)}

    resp = rerank_client.rerank(
        queries=[{"type": "TEXT", "textQuery": {"text": state["question"]}}],
        sources=[
            {"type": "INLINE", "inlineDocumentSource": {
                "type": "TEXT", "textDocument": {"text": d.page_content}}}
            for d in cands
        ],
        rerankingConfiguration={
            "type": "BEDROCK_RERANKING_MODEL",
            "bedrockRerankingConfiguration": {
                "numberOfResults": CONTEXT_K,
                "modelConfiguration": {
                    "modelArn": f"arn:aws:bedrock:{REGION}::foundation-model/amazon.rerank-v1:0"
                },
            },
        },
    )
    ranked = resp["results"]
    top = [cands[r["index"]] for r in ranked]
    return {
        "context_docs": top,
        "confidence": round(float(ranked[0]["relevanceScore"]), 3),
        "timings": _ms("rerank", t0),
    }


def generate(state: State) -> State:
    """Abstain below the floor rather than answering from weak context."""
    t0 = time.perf_counter()
    if not state["context_docs"] or state["confidence"] < CONFIDENCE_FLOOR:
        return {
            "answer": "I cannot ground an answer for that in a source you are permitted "
                      "to read. Routing to the document owner.",
            "timings": _ms("generate", t0),
        }
    context = "\n\n".join(
        f"[{i}] {d.metadata.get('source')}\n{d.page_content}"
        for i, d in enumerate(state["context_docs"], start=1)
    )
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Answer only from the numbered context and cite each claim as [n]."),
        ("human", "Context:\n{context}\n\nQuestion: {question}"),
    ])
    reply = (prompt | llm).invoke({"context": context, "question": state["question"]})
    return {"answer": reply.content, "timings": _ms("generate", t0)}


def build_graph(store):
    g = StateGraph(State)
    g.add_node("retrieve", lambda s: retrieve(s, store=store))
    g.add_node("rerank", rerank)
    g.add_node("generate", generate)
    g.add_edge(START, "retrieve")
    g.add_edge("retrieve", "rerank")
    g.add_edge("rerank", "generate")
    g.add_edge("generate", END)
    return g.compile()

该版本负责数据分块、嵌入计算、向量存储以及提示词组装等工作。当分段策略、混合词汇+密集融合模式或租户过滤功能需要以代码形式实现而非仅通过控制台选项调整时，团队会选择此方案。

三种方案的选择

在需要快速演示且数据量较少的情况下，可从托管模式开始使用。当分支处理、人工审核以及多轮对话间的记忆保持成为核心需求时，应转而使用LangGraph。若对标识符、访问控制列表或领域分段的检索质量是产品的关键指标，则需自主管理整个处理流程。

真正能提升质量的评估方法

在冻结的题目集上测试@k召回率与@k精确度，以及答案与检索到的片段的匹配程度。将每次片段长度或嵌入器的变化与其前后得分对应起来。若最佳检索片段质量较差，则拒绝使用该结果——经过校准的沉默胜过生硬的编造。

何时不应使用RAG

对于基础模型已掌握的常识性内容、纯粹的创意任务、需要通过API处理的动态数据，以及无法承受检索延迟的超低延迟场景，均无需进行检索。有时，更清晰的指令或对格式的微调就足够了。

总结

RAG就像管道系统：需要干净的输入、可靠的检索、明确的生成指令以及可衡量的控制机制。这三种模型体现了同一理念在不同掌控程度下的应用。应根据实际控制需求选择合适的层级，而非仅看演示文稿中看似最先进的那个。

操作检查清单：(1) 用于索引和查询的嵌入器各一个；(2) 从第一天起就设置ACL相关的元数据及日期信息；(3) 当出现标识符时采用混合搜索方式；(4) 在精确度至关重要的情况下重新排序；(5) 在持续集成中设置关键问题；(6) 当数据源发生变化时触发重建。应将知识体系视为一款拥有负责人、服务等级指标及回滚机制的产品，而非无人监控的周末临时索引。随着智能体图谱的扩展，除了衡量答案质量外还需关注令牌成本；如果某种编排方式在保持内容准确性的前提下仍比简单对话的消耗高出15倍，那不过是伪装成架构设计的倒退行为。需为无结果响应、过时索引问题以及嵌入器匹配错误等情况准备操作手册，避免值班人员只能根据聊天记录临时应对。

为每个语料库来源指定负责人并明确更新频率；否则这些被忽视的存储桶就会逐渐沦为隐患。

应尽早采用片段重叠方式；只有在评估显示固定窗口无效时才使用语义分割器。

将获取的ID与对应答案一起记录，以便区分糟糕回复是由于数据获取失败还是生成偏差造成的。

尽可能将评估人员与回答模型分开，以减少自我偏好偏差。

将用于索引构建的API支出与交互式服务支出分开预算，以便财务部门能清晰看到两项开支。

在索引构建时同步提供版本提示模板；若不重新评估就直接调整提示，就会导致功能退化。

对于多租户系统，需测试是否存在通过声称具有管理员权限的提示注入来绕过过滤机制的情况。

当忠实度监控指标下降时，应设置紧急关闭机制，停止生成内容并仅返回引用信息。

每当产品代码在流量构成中占据主导地位时，都应重新审视混合词法权重设置。

培训支持人员，在提交“错误机器人”相关工单之前先阅读引用面板。

为每个语料库来源记录负责人及更新频率；无人管理的存储桶会逐渐失效。

尽量尽早实现段落重叠处理；只有在评估证明固定窗口方法无效时才采用语义分割方式。

将获取的ID与回复结果一同记录，以便区分是数据获取失败还是生成内容偏离问题。

尽可能将评估人员与回复系统分开，以减少自我偏好偏差。

将索引构建的API支出与交互式服务支出分开预算，以便财务部门清晰掌握各项开支。

在索引构建时同步更新提示词模板；若不重新进行评估就直接升级提示词，会导致功能退化。

对于多租户系统，需测试是否存在通过声称具有管理员权限的提示词注入来绕过过滤机制的情况。

设置一个紧急关闭机制，在忠实度监控指标下降时立即停止内容生成，仅返回引用信息。

每当产品代码在流量构成中占据主导地位时，就重新评估混合词频权重。

培训客服人员，在升级“错误机器人”相关工单之前先查看引用信息面板。

为每个语料库来源标注负责人及更新频率；否则这些未被管理的资源会逐渐失效。

尽量尽早实现文本片段的重叠处理；只有在评估表明固定窗口方法无效时才采用语义分割方式。

将获取编号与回复内容一同记录，以便区分糟糕的回复是由于数据获取失败还是生成偏差造成的。

在可能的情况下，将评估人员与回复系统分开，以减少自我偏好偏差。

将用于索引的API支出与交互式服务支出分开预算，以便财务部门能清晰看到两项开支情况。

在索引构建版本旁设置提示词模板；未经重新评估的提示词调整正是缺陷传播的方式。

对于多租户系统，需测试是否存在通过伪装成管理员角色来绕过过滤机制的情况。

设置一个紧急关闭开关，当忠实度监控指标下降时立即停止内容生成并仅返回引用信息。

每当产品代码在流量构成中占据主导地位时，就需重新审视混合词频权重设置。

培训支持人员，在提交“错误机器人”相关工单之前先查看引用信息面板。

将评估结果、拒绝率指标与延迟数据并列展示，以便在同一表格中直观了解成本与质量状况。

将评估结果与拒绝率指标与延迟数据放在一起，以便在同一张表格中查看成本和质量情况。

将评估结果与拒绝率指标与延迟值放在一起，以便在同一张表格中查看成本和质量情况。

将评估结果与拒绝率指标与延迟数据放在一起，以便在同一张表格中查看成本和质量情况。

将评估结果与拒绝率指标与延迟值放在一起，以便在同一张表格中查看成本和质量情况。

将评估结果与拒绝率指标与延迟数据放在一起，以便在同一张表格中查看成本和质量情况。

全部文章发送简报