AI Agent，即人工智能代理（Artificial Intelligence Agent），是一种能够感知环境、进行自主理解、决策并执行动作的智能体。它通常基于机器学习和人工智能技术，具备自主性和自适应性，在特定任务或领域中能够自主地进行学习和改进。 AI Agent的核心在于其“智能”，即通过算法模拟人类或其他生物的智能行为，以自动化解决复杂问题。

【实战】使用RAG构建问答智能体

本章首先介绍实战的整体架构，然后介绍如何实现索引和检索，生成回答，最后介绍如何实现溯源、流式输出，以及结构化数据的检索和生成。

15.1.1 整体架构之项目介绍

虽然LLM能推理广泛的主题，但其知识受限于训练时所使用的公开数据。如果需要构建能处理私有数据或新数据的AI应用，则需要通过RAG技术引入相关信息以增强模型知识。简言之，RAG就是向指令中融入适当信息的过程。

LangChain设计了一系列组件，旨在辅助构建问答应用、RAG应用等。

源代码见本书配套资源中的“/Chapter15/RAG.ipynb”。

15.1.2 核心组件

典型的RAG应用主要包括以下两个核心组件。

— 索引（Indexing）：负责从数据源中提取数据并构建索引的管道，通常在线下完成。

— 检索与生成（Retrieval and Generation）：实际的RAG流程，在运行时接收用户查询，从已建立的索引中检索相关信息，并且传递给模型进行处理。

1．索引的过程

（1）数据加载（Data Loading）：通过DocumentLoader加载所需数据。

（2）文本拆分（Text Splitting）：利用文本拆分器将大型文档拆分成小块文本。这是因为小块文本更便于索引和模型处理，大型文档不仅搜索难度大，而且不适合模型的有限上下文窗口。

（3）数据存储与索引（Data Storage and Indexing）：需要一个存储和索引拆分后的数据块的地方，以便将来能够进行高效搜索。这通常借助向量数据库和嵌入模型来完成。

2．检索和生成过程

（1）数据检索（Data Retrieval）：根据用户输入，利用检索器从已存储的数据中精准检索出拆分后的相关数据块。

（2）答案生成（Answer Generation）：模型结合用户的问题和检索到的数据，通过特定的指令来生成准确、相关的答案。

3．结构化数据的检索和生成过程

结构化数据是指具有明确结构和格式的数据，它们通常存储在关系数据库（如MySQL和Oracle等）中。相对而言，非结构化数据，如文本、图片和视频等，缺乏固定的结构和格式，因此在处理时更为复杂。

结构化数据的检索和生成过程如下。

（1）提交问题。

（2）LLM将问题转换为SQL语句。

（3）数据库执行查询。

（4）LLM获取查询结果，并且转换为最终答案。

15.2 实现索引和检索

本节介绍如何实现索引和检索。

15.2.1 实现索引

（1）安装依赖，代码如下。

pip install --upgrade --quiet  langchain langchain-community langchain-chroma 

（2）导入相关依赖，代码如下。

import langchain

from langchain.text_splitter import CharacterTextSplitter

from langchain.document_loaders import TextLoader

（3）加载文档，代码如下。

loader = TextLoader("../example_data/Elon Musk's Speech at WAIC 2023.
txt",encoding='utf-8')

documents = loader.load()

（4）将加载的文档拆分成块，代码如下。

text_splitter = CharacterTextSplitter(chunk_size=500, chunk_
overlap=0)

docs = text_splitter.split_documents(documents)

（5）使用模型创建向量，代码如下。

from langchain.vectorstores import Chroma

from langchain_community.embeddings import OllamaEmbeddings

embeddings_model = OllamaEmbeddings(model="nomic-embed-text")

（6）将向量加载到Chroma中，代码如下。

db = Chroma.from_documents(docs, embeddings_model, persist_directory=
"chroma_db")

15.2.2 实现检索

实现检索的代码如下。

retriever = vectorstore.as_retriever(search_type="similarity", search_
kwargs={"k": 6})

retrieved_docs = retriever.invoke(query)

len(retrieved_docs)

print(retrieved_docs[0].page_content)

输出以下信息。

我认为人工智能在未来人类社会的演进中将发挥重要作用，并且对文明产生深远的影响。

…//省略部分内容

因此，我们需要小心确保最终结果对人类有益。

15.3 生成回答

接下来将使用LLM来生成回答。这要求构建一条流程链，该链能够接收用户问题，检索相关文档，构建并传递指令给模型，最后解析并输出答案。

15.3.1 创建指令模板

创建指令模板，代码如下。

from langchain_core.prompts import PromptTemplate

template = """使用以下内容回答问题。

如果你不知道答案，就说你不知道，不要试图编造答案。

最多使用三句话，并尽可能简明扼要。

总是在回答的最后说“谢谢你的提问！”。

{context}

问题: {question}

有用的回答:"""

custom_rag_prompt = PromptTemplate.from_template(template)

15.3.2 定义链

下面将使用Runnable接口来定义链，旨在实现以下目标：透明地组合各组件和功能，在LangSmith中自动追踪链的执行，以及实现流式、异步和批量调用。

定义链，代码如下。

…//部分代码省略，详见本书配套资源

def format_docs(docs):

    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (

 {"context": retriever | format_docs, "question": RunnablePassthrough()}

    | custom_rag_prompt

    | model

    | StrOutputParser()

)

从数据流中逐块读取数据，并且立即输出生成的答案，代码如下。

for chunk in rag_chain.stream(query):

print(chunk, end="", flush=True)

输出以下信息。

 …//省略部分内容

因此，可以得出结论：随着人工智能技术的发展，人工智能在人类社会文化传统和心理健康方面将产生深远影响。

15.4 实现溯源

使用LCEL，可以轻松地返回检索到的文档，代码如下。

…//部分代码省略，详见本书配套资源

rag_chain_from_docs = (

    RunnablePassthrough.assign(context=(lambda x: format_docs(x
["context"])))

    | prompt

    | model

    | StrOutputParser()

)

rag_chain_with_source = RunnableParallel(

    {"context": retriever, "question": RunnablePassthrough()}

).assign(answer=rag_chain_from_docs)

rag_chain_with_source.invoke(query)

输出以下信息。

{'context': [Document(page_content='我认为人工智能在未来人类社会的演进中将发挥重要作用，并且对文明产生深远的影响。

 …//省略部分内容

', metadata={'source': "../example_data/Elon Musk's Speech at WAIC 2023.txt"}),

  Document(page_content='我认为人工智能在未来人类社会的演进中将发挥重要作用，并且对文明产生深远的影响。

 …//省略部分内容

 Document(page_content='特斯拉认为我们已经非常接近完全无人干预的全自动驾驶状态了。

 …//省略部分内容

', metadata={'source': "../example_data/Elon Musk's Speech at WAIC 2023.txt"})],

 'question': '马斯克认为人工智能将对人类文明产生什么影响',

 'answer': '全球机器人的数量预计将超过人类的数量，这将是一个具有挑战性的问题，因为全自动驾驶汽车是人工智能领域的一个重大
突破，而这种突破在很大程度上依赖人工智能技术的发展和应用。\n\n然而，尽管全自动驾驶汽车可能会在未来实现，但这种有限的人工
智能与通用人工智能是完全不同的，通用人工智能很难定义。通用人工智能是一种超越人类在任何领域的智能的一种类型。特斯拉并没有
在这方面进行研究，其他公司正在研究AGI。但我认为这是现在我们需要考虑的重要问题。\n'}

15.5 实现流式传输最终输出

使用LCEL，可以便捷地实现流式传输最终输出，代码如下。

for chunk in rag_chain_with_source.stream(query):

print(chunk)

输出以下信息。

{'question': '马斯克认为人工智能将对人类文明产生什么影响'}

{'answer': '全球'}

{'answer': '机器'}

{'answer': '人的'}

{'answer': '数量'}

{'answer': '将'}

 …//省略部分内容

如果需要流式传输链的最终输出，以及某些中间步骤，则可以使用astream_log()方法。此方法具备异步特性，能执行JSONPatch操作，实现数据的流式记录与传输。

15.6 实现结构化数据的检索和生成

结构化数据的检索和生成无须进行向量化处理。下面展示如何实现结构化数据的检索和生成。

源代码见本书配套资源中的“/Chapter15/RAGSQL.ipynb”。

15.6.1 连接数据库

连接数据库需要使用SQLAlchemy驱动的SQLDatabase类与数据库建立接口连接，代码如下。

from langchain_community.utilities import SQLDatabase

…//部分代码省略，详见本书配套资源

db = SQLDatabase.from_uri("sqlite:///my.db")

print(db.dialect)

print(db.get_usable_table_names())

db.run("SELECT * FROM Artist LIMIT 10;")

输出以下信息。

sqlite

['Album', 'Artist', 'Customer', 'Employee', 'Genre', 'Invoice', 'InvoiceLine', 'MediaType', 'Playlist', 'PlaylistTrack', 'Track']

"[(1, 'AC/DC'), (2, 'Accept'), (3, 'Aerosmith'), (4, 'Alanis Morissette'), 
(5, 'Alice In Chains'), (6, 'Ant^onio Carlos Jobim'), (7, 'Apocalyptica'), (8, 'Audioslave'),
 (9, 'BackBeat'), (10, 'Billy Cobham')]"

接下来将构建一条链，该链将接收问题，通过LLM将问题转换为SQL查询语句，执行查询操作，并且基于查询结果回答原始问题。

15.6.2 将问题转换为SQL查询语句

SQL链或代理的第一步是接收用户输入并将其转换为SQL查询语句。LangChain为此提供了一条内置链：create_sql_query_chain。

使用create_sql_query_chain将问题转换为SQL查询语句的方法如下。

from langchain.chains import create_sql_query_chain

…//部分代码省略，详见本书配套资源

chain = create_sql_query_chain(model, db)

response = chain.invoke({"question": "有多少名员工?"})

response

输出以下信息。

'SELECT COUNT(*) FROM Employee'

15.6.3 执行SQL查询

成功生成SQL查询语句之后，接下来就是执行它。然而，这是构建SQL链中风险较高的环节。因此，在执行自动查询前必须深思熟虑，确保数据的安全性。

提示：为降低风险，建议尽可能降低数据库连接的权限，并且在执行查询前增设人工审批环节。

使用QuerySQLDataBaseTool可以轻松地将查询执行功能整合至链中，代码如下。

from langchain_community.tools.sql_database.tool import QuerySQLDataBaseTool

execute_query = QuerySQLDataBaseTool(db=db)

write_query = create_sql_query_chain(model, db)

chain = write_query | execute_query

chain.invoke({"question": "有多少名员工?"})

输出以下信息。

'[(8,)]'

15.6.4 生成最终答案

实现了自动生成SQL查询语句和执行SQL查询之后，接下来只需要将原始问题与SQL查询结果相结合，即可生成最终答案。为此，可以将问题和结果再次传递给LLM进行处理，代码如下。

…//部分代码省略，详见本书配套资源

answer_prompt = PromptTemplate.from_template(

    """给定以下用户问题、相应的SQL查询和SQL查询结果，回答用户问题。

Question: {question}

SQL Query: {query}

SQL Result: {result}

Answer: """

)

answer = answer_prompt | model | StrOutputParser()

chain = (

    RunnablePassthrough.assign(query=write_query).assign(

        result=itemgetter("query") | execute_query

    )

    | answer

)

chain.invoke({"question": "有多少名员工?"})

输出以下信息。

有8名员工。

以上内容摘自《LangChain实战派：大语言模型+LangChain+向量数据库》

添加图片注释，不超过 140 字（可选）

本书采用“知识点+实战”的编写方式，共包含28个基础实战和1个综合性实战，旨在深入解析大语言模型应用开发的核心知识。

每个知识点的介绍均遵循清晰的逻辑脉络：介绍概念、阐述应用原理、说明使用方法、探讨选择该方法的理由、提供优化建议，并且分享实践案例。

本书适合对LangChain感兴趣的读者阅读。