在过去的几个月里，许多由 LLM驱动的应用都汇聚到了人工智能代理的框架中，人工智能代理可以被定义为专门用于处理用户查询和执行任务的实体。更具体地说，人工智能代理具有以下特征：

• 它由 LLM（或 SML）驱动，LLM 充当应用程序的 “大脑”。这意味着，人工智能代理能够规划一系列行动，以完成用户的查询。
• 它可以访问一系列工具或插件，以便在周围的生态系统中执行操作（例如，查询 ERP 中的数据、发送带有生成的时事通讯的电子邮件、在 Linkedin 上发布图片......）。
• 随着多模态技术的出现，人工智能代理现在可以感知周围环境的所有元素。例如，人工智能代理可以看到因事故造成的汽车损坏，大声告诉车主并向保险公司提出索赔。
• 它有能力记住自己过去的互动和行为，并在未来加以改进。这个部分被称为记忆（更确切地说，是短期记忆）。

事实上，在人工智能代理方面，我们可能需要存储三种主要数据：

• 对话交互 → 我们希望为代理保留一个上下文窗口，以便记住某些事情，无论是与用户的完整对话，还是可能反复出现的特定问答。在这种情况下，最佳做法是使用内存数据库。
• 外部知识库 → 人工智能代理的一个关键组成部分是所谓的非参数知识，它是基于 RAG 的应用的核心。在这种情况下，近两年兴起了革命性的矢量搜索和矢量数据库方法。市场上有很多数据库可供选择，有些是向量原生数据库，如 Qdrand 或 Weaviate，有些则具有向量功能，如 Azure SQL 数据库。

• 跟踪/日志和对话历史（非缓存）→生产就绪的应用程序（包括人工智能代理）需要考虑日志机制和日志存储。在这种情况下，关系数据库就非常适合。此外，它们还可用于存储那些超出预期上下文窗口的对话，或者当我们不想缓存它们时。

现在，拥有多个独立数据库并不理想，因为这会影响人工智能代理的性能。此外，它还会造成额外的超负荷维护。最后但并非最不重要的一点是，尽管在代理之间共享内存通常很有用（也许两个或更多代理可能会合作解决一个问题），但允许每个代理管理自己的内存仍然很重要，因为这能体现其全部专业知识和个性。

什么是 Azure Cosmos DB 以及如何将其与AI代理一起使用

Azure Cosmos DB 是一种完全托管的 NoSQL、关系型和矢量数据库。它的响应时间仅为个位数毫秒，具有自动和即时可扩展性，在任何规模下都能保证速度。从地理复制分布式缓存到备份和矢量索引，它为现代应用提供了基础架构。此外，它还具有 SLA 支持的可用性和企业级安全性，可确保业务连续性。

CosmosDB 提供多种数据库应用程序接口：

• NoSQL → 它以文档格式（如 JSON）存储数据
• MongoDB → 它实现了 MongoDB 线协议，非常适合具有复杂嵌套数据结构的文档数据。
• PostgreSQL → 它支持关系数据和类似 SQL 的查询。
• Cassandra → 它与 Apache Cassandra 兼容，专为宽列存储而设计。
• Gremlin → 支持基于图形的数据建模。
• Table → 专为存储键值数据模型而设计。

NoSQL、PostgreSQL 和 MongoDB 也提供了具有多种搜索算法的向量存储功能。在本文中，我们将看到一个使用 MongoDB API 的例子，利用它既可以为 RAG 存储知识库，又可以进行内存缓存以获得更好的性能。

用嵌入式内容填充数据库

第一步是在你的 Azure 订阅上创建 CosmosDB 实例。更具体地说，我们需要为具有 vCore 架构的 MongoDB 创建一个 CosmosDB，该架构支持原生向量集成。

你可以从配置免费层开始，免费层可提供 32GiB 的存储空间。

实例启动并运行后，就可以开始填充数据了。你可以决定直接使用 Azure 门户中的 Mongo Shell：

或者通过 Python 中的 pymongo 库与数据库交互（通过 pip install pymongo 可以轻松安装）。我们将使用后一种方法将嵌入式内容上传到数据库。

为此，我们首先需要通过客户端创建一个与数据库的连接，并需要我们的连接字符串。你可以在 Azure CosmosDB 实例的连接字符串选项卡下找到它（你必须在其中填写部署实例时选择的用户名和密码）：

然后，可以按如下方式初始化客户端：

from pymongo import MongoClient
CONNECTION_STRING = "your-conection-string"
client: MongoClient = MongoClient(CONNECTION_STRING)

很好，现在我们有了客户端，可以用嵌入式内容填充数据库了。为此，我们需要：

• 数据库和要用嵌入式素材填充的集合名称：

INDEX_NAME = "vaalt-test-index""vaalt-test-index"
NAMESPACE = "vaalt_test_db.vaalt_test_collection"
DB_NAME, COLLECTION_NAME = NAMESPACE.split(".")

• 嵌入模型（我将使用 Azure OpenAI ada-002 模型）：

from langchain_openai import AzureOpenAIEmbeddings
os.environ["AZURE_OPENAI_API_KEY"] = "xxx"
os.environ["AZURE_OPENAI_ENDPOINT"] = "xxx"
embeddings = AzureOpenAIEmbeddings(
    azure_deployment="text-embedding-ada-002",
    openai_api_version="2023-05-15",
)

• 嵌入并保存到数据库中的知识库。在我们的案例中，我们将使用 Humza Naveed 等人撰写的 PDF 文件《大型语言模型综合概述》。然后，我们将利用 LangChain 库处理文档并对其进行分块，以获得相关嵌入。

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import CharacterTextSplitter
loader = PyPDFLoader("https://arxiv.org/pdf/2307.06435")
documents = loader.load()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
docs = text_splitter.split_documents(documents)

• 初始化向量存储：

from langchain_community.vectorstores.azure_cosmos_db import (
    AzureCosmosDBVectorSearch,
    CosmosDBSimilarityType,
    CosmosDBVectorSearchType,
)
collection = client[DB_NAME][COLLECTION_NAME]

vectorstore = AzureCosmosDBVectorSearch.from_documents(
    docs,
    embeddings,
    collection=collection,
    index_name=INDEX_NAME,
)

• 填充向量存储：

num_lists = 100100
dimensions = 1536
similarity_algorithm = CosmosDBSimilarityType.COS
kind = CosmosDBVectorSearchType.VECTOR_IVF
m = 16
ef_construction = 64

vectorstore.create_index(
    num_lists, dimensions, similarity_algorithm, kind, m, ef_construction
)

请注意，在填充向量存储空间之前，我们初始化了以下变量：

• num_lists： 用于索引的列表或分区数量（影响性能和准确性）。
• dimensions：维度： 向量的维度（本例中为 1536）。
• similarity_algorithm： 用于向量搜索的相似度量（COS 表示余弦相似度）。
• kind： 类型： 向量搜索算法的类型（VECTOR_IVF 表示反转文件索引）。
• ef_construction：它决定了在索引构建阶段要考虑的邻居数量，并影响向量搜索的性能和准确性。数值越大，索引质量越好，但代价是索引构建时间越长。

索引建立完成后，你会看到类似下面的信息：

{'raw': {'defaultShard': {'numIndexesBefore': 1,
   'numIndexesAfter': 2,
   'createdCollectionAutomatically': False,
   'ok': 1}},
 'ok': 1}

现在，让我们初始化一个 LangChain 问答链，创建一个智能代理，用自然语言回答我们的询问。

# Retrieve and generate using the relevant snippets of the blog.
from langchain import hub
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
retriever = vectorstore.as_retriever()
message = """
Answer this question using the provided context only.
{question}
Context:
{context}
"""
prompt = ChatPromptTemplate.from_messages([("human", message)])
rag_chain = {"context": retriever, "question": RunnablePassthrough()} | prompt | llm
rag_chain.invoke("What is BLOOM?")

输出：

AIMessage(content='BLOOM is a causal decoder model trained on the ROOTS corpus to open-source a large language model (LLM). The architecture of BLOOM includes differences such as ALiBi positional embedding and an additional normalization layer after the embedding layer, as suggested by the bitsandbytes library. These changes are intended to stabilize training and improve downstream performance.', response_metadata={'token_usage': {'completion_tokens': 71, 'prompt_tokens': 17123, 'total_tokens': 17194}, 'model_name': 'gpt-4', 'system_fingerprint': 'fp_811936bd4f', 'prompt_filter_results': [{'prompt_index': 0, 'content_filter_results': {'hate': {'filtered': False, 'severity': 'safe'}, 'self_harm': {'filtered': False, 'severity': 'safe'}, 'sexual': {'filtered': False, 'severity': 'safe'}, 'violence': {'filtered': False, 'severity': 'safe'}}}], 'finish_reason': 'stop', 'logprobs': None, 'content_filter_results': {'hate': {'filtered': False, 'severity': 'safe'}, 'self_harm': {'filtered': False, 'severity': 'safe'}, 'sexual': {'filtered': False, 'severity': 'safe'}, 'violence': {'filtered': False, 'severity': 'safe'}}}, id='run-8ccfc790-1988-4fb4-a582-0774886a479d-0', usage_metadata={'input_tokens': 17123, 'output_tokens': 71, 'total_tokens': 17194})'BLOOM is a causal decoder model trained on the ROOTS corpus to open-source a large language model (LLM). The architecture of BLOOM includes differences such as ALiBi positional embedding and an additional normalization layer after the embedding layer, as suggested by the bitsandbytes library. These changes are intended to stabilize training and improve downstream performance.', response_metadata={'token_usage': {'completion_tokens': 71, 'prompt_tokens': 17123, 'total_tokens': 17194}, 'model_name': 'gpt-4', 'system_fingerprint': 'fp_811936bd4f', 'prompt_filter_results': [{'prompt_index': 0, 'content_filter_results': {'hate': {'filtered': False, 'severity': 'safe'}, 'self_harm': {'filtered': False, 'severity': 'safe'}, 'sexual': {'filtered': False, 'severity': 'safe'}, 'violence': {'filtered': False, 'severity': 'safe'}}}], 'finish_reason': 'stop', 'logprobs': None, 'content_filter_results': {'hate': {'filtered': False, 'severity': 'safe'}, 'self_harm': {'filtered': False, 'severity': 'safe'}, 'sexual': {'filtered': False, 'severity': 'safe'}, 'violence': {'filtered': False, 'severity': 'safe'}}}, id='run-8ccfc790-1988-4fb4-a582-0774886a479d-0', usage_metadata={'input_tokens': 17123, 'output_tokens': 71, 'total_tokens': 17194})

如你所见，该模型能够返回正确答案。

现在，让我们设想一下，这是一个面向客户的人工智能代理，每秒会收到数千个问题。显然，我们必须面对延迟问题，因为我们希望保证用户在使用我们的代理时能获得出色的性能。然而，即使使用最具可扩展性的基础架构（在数据库和 LLM 基础架构方面--如 Azure OpenAI PTU），检索和生成步骤的运行也需要一定的生理时间，尤其是在涉及到庞大而复杂的知识库时。

那么我们该如何解决这个问题呢？缓存！我们的想法是在内存数据库中存储最常见的问题和相应的回复，这样如果有新客户提出同样的问题，代理就能更快地检索到这些问题。

更具体地说，既然我们谈论的是 LLM，那就使用语义缓存（Semantic Caching）。

创建语义缓存

语义缓存的革命性在于属性语义。事实上，在GenAI之前的场景中，我们可以看到缓存的工作原理如下：

在这种情况下，用户的查询和保存的问题之间需要有关键词匹配才能进入缓存。

另一方面，在语义缓存中，将有一个嵌入模型来理解用户查询的语义，并将其与内存中的问答对进行比较。这意味着用户的查询不一定要与内存中的查询完全匹配，只需具有相同的语义即可。

让我们看看如何做到这一点：

from langchain.globals import set_llm_cache
import time
# Default value for these params
num_lists = 3
dimensions = 1536
similarity_algorithm = CosmosDBSimilarityType.COS
kind = CosmosDBVectorSearchType.VECTOR_IVF
m = 16
ef_construction = 64
ef_search = 40
score_threshold = 0.9
application_name = "LANGCHAIN_CACHING_PYTHON"

set_llm_cache(
    AzureCosmosDBSemanticCache(
        cosmosdb_connection_string=CONNECTION_STRING,
        cosmosdb_client=None,
        embedding=embeddings,
        database_name=DB_NAME,
        collection_name=COLLECTION_NAME,
        num_lists=num_lists,
        similarity=similarity_algorithm,
        kind=kind,
        dimensions=dimensions,
        m=m,
        ef_construction=ef_construction,
        ef_search=ef_search,
        score_threshold=score_threshold,
        application_name=application_name,
    )
)

与向量数据库中的语义搜索类似，我们也需要为缓存内存定义向量搜索参数。这里的一个重要参数是 score_treshold：事实上，阈值越低，命中缓存的可能性就越低，因此必须根据我们的应用策略（性能与响应的准确性）来设置它。

现在，让我们试着向数据库提问并监控响应时间：

%%time
rag_chain.invoke("What is BLOOM?")"What is BLOOM?")

CPU times: total: 125 ms
Wall time: 27.5 s
AIMessage(content='BLOOM i125s a causal decoder model trained on the ROOTS corpus to open-source a large language model (LLM). The architecture of BLOOM includes differences such as ALiBi positional embedding and an additional normalization layer after the embedding layer, as suggested by the bitsandbytes library. These changes are intended to stabilize training and improve downstream performance.', response_metadata={'token_usage': {'completion_tokens': 71, 'prompt_tokens': 17123, 'total_tokens': 17194}, 'model_name': 'gpt-4', 'system_fingerprint': 'fp_811936bd4f', 'prompt_filter_results': [{'prompt_index': 0, 'content_filter_results': {'hate': {'filtered': False, 'severity': 'safe'}, 'self_harm': {'filtered': False, 'severity': 'safe'}, 'sexual': {'filtered': False, 'severity': 'safe'}, 'violence': {'filtered': False, 'severity': 'safe'}}}], 'finish_reason': 'stop', 'logprobs': None, 'content_filter_results': {'hate': {'filtered': False, 'severity': 'safe'}, 'self_harm': {'filtered': False, 'severity': 'safe'}, 'sexual': {'filtered': False, 'severity': 'safe'}, 'violence': {'filtered': False, 'severity': 'safe'}}}, id='run-e00630f1-ac2e-46e7-9e75-6c585ec58c50-0', usage_metadata={'input_tokens': 17123, 'output_tokens': 71, 'total_tokens': 17194})

现在，问题和回复都已保存到语义缓存中，因此理论上，如果我们问同样的问题（或类似的问题），应该会得到更快的回复。让我们试试看：

%%time
rag_chain.invoke("What is BLOOM?")
CPU times: total: 15.6 ms
Wall time: 4.78 s
AIMessage(content='BLOOM is a causal decoder model trained on the ROOTS corpus to open-source a large language model (LLM). The architecture of BLOOM includes differences such as ALiBi positional embedding and an additional normalization layer after the embedding layer, as suggested by the bitsandbytes library. These changes are intended to stabilize training and improve downstream performance.', response_metadata={'token_usage': {'completion_tokens': 71, 'prompt_tokens': 17123, 'total_tokens': 17194}, 'model_name': 'gpt-4', 'system_fingerprint': 'fp_811936bd4f', 'prompt_filter_results': [{'prompt_index': 0, 'content_filter_results': {'hate': {'filtered': False, 'severity': 'safe'}, 'self_harm': {'filtered': False, 'severity': 'safe'}, 'sexual': {'filtered': False, 'severity': 'safe'}, 'violence': {'filtered': False, 'severity': 'safe'}}}], 'finish_reason': 'stop', 'logprobs': None, 'content_filter_results': {'hate': {'filtered': False, 'severity': 'safe'}, 'self_harm': {'filtered': False, 'severity': 'safe'}, 'sexual': {'filtered': False, 'severity': 'safe'}, 'violence': {'filtered': False, 'severity': 'safe'}}}, id='run-e00630f1-ac2e-46e7-9e75-6c585ec58c50-0', usage_metadata={'input_tokens': 17123, 'output_tokens': 71, 'total_tokens': 17194})

如你所见，我们的响应时间缩短了 87.5%。

结论

语义缓存是人工智能代理领域的一个游戏规则，尤其是在高吞吐量应用领域。有了内存向量库，我们就能更轻松地实现预期性能，同时在用户查询和保存的问答对之间保持高精度（语义）匹配。显然，在生产部署中，我们需要定义一个 “频繁问答 ”模式来决定保存哪些问答对，这可以通过适当的监控来实现。

总之，由于 CosmosDB 的灵活性，我们可以利用它改进我们架构的多个方面，同时保持单一的数据库服务作为后端。