基于LangChain的RAG智能体

前几天，学习了langchain官方文档里的第一个例子“语义搜索”，这是官网的第二个示例，使用LangChain构建一个RAG智能体。链接在这里：https://docs.langchain.com/oss/python/langchain/rag

目标

这个示例，将读取一个博客网站，将文本进行分块处理，保存在向量库。用户发起对话之后，先从向量库中检索，再用大模型回答，这就是一个简单的RAG（检索增强生成）智能体。

获取网页

先安装bs4模块。

pip install bs4

这里使用WebBaseLoader加载网页，使用BeautifulSoup提取关心的内容，得到LangChain里的Document列表。这个Loader与以前的PdfLoader有相同的基类BaseLoader。

import bs4
from langchain_community.document_loaders import WebBaseLoader

# 从HTML中只保留我们关心的主要内容
bs4_strainer = bs4.SoupStrainer(class_=("post-title", "post-header", "post-content"))
loader = WebBaseLoader(
    web_paths=["http://shenlb.me/gtd/"],
    bs_kwargs={"parse_only": bs4_strainer},
)

# 读网页，生成List[Document]
docs = loader.load()

assert len(docs) == 1
print(f"文章字符数: {len(docs[0].page_content)}")
print(docs[0].page_content[:200])

文本分隔

如果文章很长，将所有内容直接送给大模型，大模型消化不了，所以需要分块处理。

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=5000, chunk_overlap=500, add_start_index=True
)
all_splits = text_splitter.split_documents(docs)
print("chunks: ", len(all_splits))

向量嵌入

这里访问阿里的文本嵌入模型，查询每一个文本块，得到一个1024维的浮点数向量，存入到chroma向量数据库中，以备后面的检索之用。

# 阿里百炼 Embeddings 适配器
class DashScopeEmbeddings(Embeddings):
    def __init__(
        self,
        api_key: str,
        model: str = "text-embedding-v4",
        dimensions: int = 1024,
    ):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        )
        self.model = model
        self.dimensions = dimensions

    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        response = self.client.embeddings.create(
            model=self.model,
            input=texts,
            dimensions=self.dimensions,
        )
        return [item.embedding for item in response.data]

    def embed_query(self, text: str) -> list[float]:
        response = self.client.embeddings.create(
            model=self.model,
            input=[text],
            dimensions=self.dimensions,
        )
        return response.data[0].embedding

# 阿里的文本嵌入模型
embeddings = DashScopeEmbeddings(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    model="text-embedding-v4",
    dimensions=1024,
)

vector_store = Chroma(
    collection_name="gtd",
    embedding_function=embeddings,
    persist_directory="./gtd_chroma_db",
)

if vector_store._collection.count() == 0:
    print("首次运行，写入网页向量库...")
    batch_size = 10  # 百炼 embedding 限制
    for i in range(0, len(splits), batch_size):
        vector_store.add_documents(splits[i:i + batch_size])
else:
    print("向量库已存在，直接使用")

RAG智能体

先定义一个工具，智能体可以调用这个工具，得到一些信息片段。

@tool 修饰符把一个普通的 Python 函数，声明成了“可以被大模型调用的工具”。

下面的@tool声明之后，用来告诉大模型：有一个工具，名字叫 retrieve_context，你在需要的时候可以用。

“content_and_artifact"意思是说有对话文本内容，还有原始素材。

docstring注释也是必须的，告诉大模型这个工具的用途。否则，会报ValueError: Function must have a docstring if description not provided.错误。

@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """检索向量库，改善回答的准确性和可靠性。"""
    docs = vector_store.similarity_search(query, k=3)

    combined = "\n\n".join(
        f"来源: {doc.metadata}\n内容: {doc.page_content}"
        for doc in docs
    )
    return combined, docs

下面可以定义一个智能体，智能体会自动调用这个工具，完成回答任务。

llm = ChatOpenAI(
    model="qwen-plus",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

from langchain.agents import create_agent

tools = [retrieve_context]
prompt = (
    "你是AI智能体，可以访问一些工具，获取某个博客文章的内容。"
    "辅助问答用户的问题。"
    "请严格基于原始素材，不要编造，并将答案控制在500字以内。"
)
agent = create_agent(llm, tools, system_prompt=prompt)

query = "什么是GTD？"
result = agent.invoke(
    {"messages": [{"role": "user", "content": query}]}
)
print(result["messages"][-1].content)

上面智能体的内部是一个黑盒，我们不知道中间工具的调用情况，可以使用流式执行(Streaming)，观察中间的一些运行细节。

智能体每往前走一步，会输出一些中间状态发给你。event里会不断地追加消息，所以event["messages"][-1]就是最新的一条消息。

for event in agent.stream(
    {"messages": [{"role": "user", "content": query}]},
    stream_mode="values",
):
    event["messages"][-1].pretty_print()

可以看到如下的输出：

================================ Human Message =================================

什么是GTD？
================================== Ai Message ==================================
Tool Calls:
  retrieve_context (call_9a847bb3974b45eca3a046)
 Call ID: call_9a847bb3974b45eca3a046
  Args:
    query: GTD 时间管理方法
================================= Tool Message =================================
Name: retrieve_context

Source: {'source': 'http://shenlb.me/gtd/'}
Content: GTD是英文Getting Things Done的缩写，是一种行为管理的方法，也是David Allen写的一本书的书名。……
================================== Ai Message ==================================

GTD（Getting Things Done）是大卫·艾伦（David Allen）提出的一种时间与自我管理系统，核心理念是“把 头脑中的任务清空到外部系统中”，从而减少焦虑、提升专注力与执行力。……

根据用户的问题，智能体可以在内部多次调用工具，来得到更好的答案。比如把问题改为：

query = (
    "什么是GTD？"
    "当你得到答案之后，再进行第二次查询，详细解释一下这个概念里的每一个步骤。"
)

事件里就可以看到多个 ToolMessage。

代码里其实还有个地方需要完善，每次都查询网页其实是没必要的，chroma向量库里已经有相关信息了。