您可以将MongoDB与 Google ADK 集成，以配置代理搜索工具和持久对话历史记录，二者均由MongoDB支持。本教程使用 MongoDbToolset 类启用Google ADK代理能够搜索包含企鹅数据的集合。然后，本教程使用 MongoDbSessionService 类，允许代理访问权限对话中的较早回合。
在本教程中，您将执行以下任务：
设置您的环境。
创建集合和向量搜索索引。
构建一个代理来搜索您的集合。
添加持久会话，以便代理记住之前的轮次。
运行文件并查看输出。
背景
Google ADK 是一个用于构建AI助手的开源Python框架。本教程使用以下 Google ADK 组件来配置代理并启用MongoDB访问权限：
MongoDbToolset：在MongoDB 数据库上提供MongoDB Vector Search 和混合搜索工具MongoDbSessionService：在MongoDB中保留代理会话、事件和状态
本教程将 MongoDbToolset 类配置为使用Atlas自动嵌入，该功能可嵌入存储的文档和搜索查询，而无需单独的嵌入模型或 Google凭证。
先决条件
要完成本教程，您必须具备以下资源：
以下MongoDB 集群类型之一：
使用 Python 和 Docker 创建的本地 Atlas 部署。安装
atlas-local-lib-py(pip install atlas-local-lib-py)，以编程方式创建和管理本地部署。要了解更多信息，请参阅 atlas-local-lib-py 存储库。安装了Search 和向量搜索的 MongoDB Community 集群。
Python v3.11 或更高版本。
设置环境
在本部分中，您将通过安装所需的依赖项、设置环境变量和创建项目文件来准备环境。
设置环境变量。
在您的 google-adk-mongodb-project 目录中，创建一个 .env 文件并添加以下代码：
GOOGLE_API_KEY="<gemini-api-key>"MONGODB_URI="<connection-string>"将 <connection-string> 替换为您的 Atlas 集群或本地部署的连接字符串。
连接字符串应使用以下格式：
mongodb+srv://<db_username>:<db_password>@<clusterName>.<hostname>.mongodb.net重要: 确保连接字符串包含数据库用户的凭证。要学习有关查找连接字符串的更多信息，请参阅通过客户端库连接到集群。
创建项目文件。
在 google-adk-mongodb-project目录中，创建名为 main.py 的文件。将以下代码添加到此文件以加载环境变量并导入所需的模块：
import osimport timeimport asynciofrom dotenv import load_dotenvfrom pymongo import MongoClientfrom pymongo.operations import SearchIndexModelfrom google.adk.agents import Agentfrom google.adk.integrations.mongodb import ( MongoDbToolset, MongoDbToolSettings, MongoDbSessionService,)from google.adk.runners import Runnerfrom google.genai import types
load_dotenv()在后续步骤中，您将向此文件添加代码，以创建您的集合、构建并运行您的代理。
创建集合和索引
在本部分中，您将在MongoDB集合中插入有关企鹅的示例文档，并创建Atlas自动嵌入所需的MongoDB Vector Search索引。
插入示例文档。
将以下代码添加到您的 main.py文件中，以将五个描述企鹅的文档插入到您的集合中。 embeddings字段存储Atlas自动嵌入的文本：
collection.insert_many( [ { "title": "Emperor Penguins", "embeddings": "Emperor penguins are the tallest and heaviest of all penguin species, standing up to 4 feet.", }, { "title": "Penguin Habitats", "embeddings": "Penguins are flightless seabirds that live almost exclusively below the equator. Some island-dwellers can be found in warmer climates.", }, { "title": "Penguins of Patagonia", "embeddings": "Patagonia is home to five penguin species: Magellanic, Humboldt, Gentoo, Southern Rockhopper, and King.", }, { "title": "Penguin Diving Depths", "embeddings": "Emperor penguins can dive deeper than any other bird, reaching depths of over 500 meters.", }, { "title": "Penguin Nesting Habits", "embeddings": "Some penguin species mate for life and return to the same nesting site every year.", }, ])创建Atlas自动嵌入索引。
将以下代码添加到 main.py文件，以在 embeddings字段上创建MongoDB Vector Search索引。 autoEmbed字段类型指示Atlas自动生成并存储该字段的嵌入：
search_index_model = SearchIndexModel( definition={ "fields": [ { "type": "autoEmbed", "path": "embeddings", "model": "voyage-4", "modality": "text", } ] }, name="autoembed_index", type="vectorSearch",)
collection.create_search_index(model=search_index_model)
# Index build time varies with collection size and cluster typeprint("Waiting for the MongoDB Vector Search index to finish building...")time.sleep(60)
client.close()构建索引大约需要一分钟时间。在构建时，索引处于初始同步状态。构建完成后，您可以开始查询集合中的数据。
配置代理工具和会话
在本部分中，您将构建一个使用 MongoDbToolset 类搜索集合的代理。然后，配置 MongoDbSessionService 类以在MongoDB中保留座席的对话历史记录。
配置工具集。
将以下代码添加到 main.py文件，将 MongoDbToolset 绑定到数据库，并将其配置为使用Atlas自动嵌入：
toolset = MongoDbToolset( connection_string=os.environ["MONGODB_URI"], database_name="google_adk_tutorial", settings=MongoDbToolSettings( use_mongodb_auto_embedding=True, default_embedding_field="embeddings", default_vector_index_name="autoembed_index", ),)由于 use_mongodb_auto_embedding 选项设立为 True，因此该工具集会将查询文本以纯文本形式发送到Atlas，而不是调用 Google 嵌入模型。
定义代理。
将以下代码添加到 main.py文件中，以创建使用 Gemini 模型和工具集的代理：
root_agent = Agent( model="gemini-2.5-flash", name="penguin_facts_agent", description="Answers questions about penguins.", instruction=( "Search the penguins collection with your MongoDB" " tools to answer questions about penguins. Remember" " what the user has told you earlier in the" " conversation." ), tools=[toolset],)配置会话服务和运行器。
将以下代码添加到 main.py文件以配置 MongoDbSessionService 类，并将其与代理一起传递给 Runner对象：
session_service = MongoDbSessionService( connection_string=os.environ["MONGODB_URI"], database_name="google_adk_tutorial",)
runner = Runner( agent=root_agent, app_name="google_adk_tutorial", session_service=session_service,)默认下，MongoDbSessionService 会创建建议使用的索引，除非您将 ensure_indexes 参数设立为 False。要学习；了解更多信息，请参阅 Google ADK 参考文档中的索引指南。
在两个对话轮次中运行代理。
将以下代码添加到 main.py文件以创建会话，然后通过该会话向代理发送两条消息：
async def ask(runner, session_id, text): content = types.Content(role="user", parts=[types.Part(text=text)]) async for event in runner.run_async( user_id="tutorial_user", session_id=session_id, new_message=content, ): if event.is_final_response() and event.content: print(f"\n[user]: {text}") print(f"[agent]: {event.content.parts[0].text}")
async def main(): session = await session_service.create_session( app_name="google_adk_tutorial", user_id="tutorial_user", )
await ask( runner, session.id, "How deep can emperor penguins dive?" ) await ask( runner, session.id, "What was the first thing I asked you?" )
asyncio.run(main())由于这两个调用使用相同的 session_id 值，因此 MongoDbSessionService 会在代理响应第二条消息之前加载第一个回合的事件。如果为第二次呼叫创建新会话，代理没有第一次呼叫的记录。
运行文件
要运行 main.py 文件，请从您的 google-adk-mongodb-project 目录运行以下命令：
python main.py如果命令成功运行，输出如下所示。
Waiting for the MongoDB Vector Search index to finish building...
[user]: How deep can emperor penguins dive?[agent]: Emperor penguins can dive deeper than any other bird,reaching depths of over 500 meters.
[user]: What was the first thing I asked you?[agent]: The first thing you asked me was about how deep emperorpenguins can dive.后续步骤
要学习；了解有关 MongoDbToolset、MongoDbSessionService 和 MongoDbMemoryService 的更多信息，包括其完整参数和方法参考，请参阅将MongoDB与 Google ADK 集成。
要学习；了解有关 Google ADK 的更多信息，请参阅 adk-Python GitHub存储库。