Build a streaming agent with Python¶
使用 Python 构建流式传输智能体¶
With this quickstart, you'll learn to create a simple agent and use ADK Streaming to enable voice and video communication with it that is low-latency and bidirectional. We will install ADK, set up a basic "Google Search" agent, try running the agent with Streaming with adk web tool, and then explain how to build a simple asynchronous web app by yourself using ADK Streaming and FastAPI.
通过此快速入门,您将学习创建一个简单的智能体,并使用 ADK 流式传输实现低延迟、双向的语音和视频通信。我们将安装 ADK,设置一个基本的"Google Search"智能体,尝试使用 adk web 工具通过流式传输运行智能体,然后解释如何使用 ADK 流式传输和 FastAPI 自己构建一个简单的异步 Web 应用程序。
Note: This guide assumes you have experience using a terminal in Windows, Mac, and Linux environments.
注意: 本指南假设您有在 Windows、Mac 和 Linux 环境中使用终端的经验。
Supported models for voice/video streaming¶
支持语音/视频流式传输的模型¶
In order to use voice/video streaming in ADK, you will need to use Gemini models that support of the Live API. You can find the model ID(s) that supports the Gemini Live API in the documentation:
要在 ADK 中使用语音/视频流式传输,您需要使用支持 Live API 的 Gemini 模型。您可以在文档中找到支持 Gemini Live API 的模型 ID:
1. Setup Environment & Install ADK¶
1. 设置环境 & 安装 ADK¶
Create & Activate Virtual Environment (Recommended):
创建并激活虚拟环境（推荐）:
# Create
python -m venv .venv
# Activate (each new terminal)
# macOS/Linux: source .venv/bin/activate
# Windows CMD: .venv\Scripts\activate.bat
# Windows PowerShell: .venv\Scripts\Activate.ps1
Install ADK:
安装 ADK:
2. Project Structure¶
2. 项目结构¶
Create the following folder structure with empty files:
创建以下文件夹结构和空文件:
adk-streaming/ # Project folder
└── app/ # the web app folder
├── .env # Gemini API key
└── google_search_agent/ # Agent folder
├── __init__.py # Python package
└── agent.py # Agent definition
agent.py¶
agent.py¶
Copy-paste the following code block into the agent.py file.
将以下代码块复制粘贴到 agent.py 文件中。
For model, please double check the model ID as described earlier in the Models section.
对于 model,请仔细检查前面 模型部分中描述的模型 ID。
from google.adk.agents import LlmAgent
from google.adk.tools import google_search
from google.adk.core.lms import google as google_lm
# Define the model
# 定义模型
model = google_lm.LmFactory.create(
"gemini-2.0-flash-exp",
api_key="<YOUR_GEMINI_API_KEY>"
)
# Create the agent
# 创建智能体
agent = LlmAgent(
name="google_search_agent",
model=model,
instruction=(
"You are a helpful assistant that can perform Google searches to "
"answer questions. Use the search tool when needed to provide "
"accurate and up-to-date information."
),
tools=[google_search.GoogleSearchTool()],
)
if __name__ == "__main__":
print("Agent created successfully!")
.env¶
.env¶
Create the .env file and add your API key:
创建 .env 文件并添加您的 API 密钥:
Important: Replace your_actual_api_key_here with your actual Gemini API key. You can get it from Google AI Studio.
重要: 将 your_actual_api_key_here 替换为您的实际 Gemini API 密钥。您可以从 Google AI Studio 获取。
3. Test Agent with adk web¶
3. 使用 adk web 测试智能体¶
You can quickly test your agent using the ADK web tool:
您可以使用 ADK web 工具快速测试您的智能体:
This will start a local web interface where you can interact with your agent.
这将启动一个本地 Web 界面,您可以在其中与智能体交互。
4. Build a Simple Web App with FastAPI¶
4. 使用 FastAPI 构建简单的 Web 应用程序¶
Now, let's create a simple web application that allows streaming interaction with the agent.
现在,让我们创建一个简单的 Web 应用程序,允许与智能体进行流式传输交互。
Update .env¶
更新 .env¶
Make sure your .env file looks like this:
确保您的 .env 文件如下所示:
Create main.py¶
创建 main.py¶
Create a main.py file in the app directory:
在 app 目录中创建 main.py 文件:
import os
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from google.adk.core.runner import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import uvicorn
from dotenv import load_dotenv
# Load environment variables
# 加载环境变量
load_dotenv()
app = FastAPI(title="ADK Streaming Agent")
# Import the agent
# 导入智能体
from google_search_agent.agent import agent
# Create session service
# 创建会话服务
session_service = InMemorySessionService()
@app.get("/")
async def root():
return {"message": "ADK Streaming Agent is running!"}
@app.post("/chat")
async def chat(request: dict):
"""
Streaming chat endpoint
流式传输聊天端点
"""
user_message = request.get("message", "")
# Create a new session or get existing one
# 创建新会话或获取现有会话
session = session_service.create_session(
user_id="default_user",
session_id="default_session"
)
# Create a runner
# 创建运行器
runner = Runner(
agent=agent,
session_service=session_service
)
async def generate():
"""Generator function for streaming responses"""
"""用于流式传输响应的生成器函数"""
# Run the agent asynchronously
# 异步运行智能体
async for event in runner.run_async(
session=session,
user_message=types.Content(
parts=[types.Part(text=user_message)]
)
):
if hasattr(event, 'content') and event.content:
# Yield the content
# 生成内容
for part in event.content.parts:
if part.text:
yield part.text
return StreamingResponse(generate())
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
Install Additional Dependencies¶
安装额外的依赖项¶
Run the Web App¶
运行 Web 应用程序¶
Your web app will be available at http://localhost:8000.
您的 Web 应用程序将在 http://localhost:8000 上可用。
Test the Streaming Endpoint¶
测试流式传输端点¶
You can test the streaming endpoint using curl:
您可以使用 curl 测试流式传输端点:
curl -X POST "http://localhost:8000/chat" \
-H "Content-Type: application/json" \
-d '{"message": "What is the latest news about AI?"}'
Summary¶
总结¶
In this quickstart, you:
在此快速入门中,您:
- Set up the ADK environment and installed dependencies 设置了 ADK 环境并安装了依赖项
- Created a Google Search agent with streaming capabilities 创建了具有流式传输功能的 Google Search 智能体
- Tested the agent using
adk web使用adk web测试了智能体 - Built a simple FastAPI web application with streaming support 构建了一个支持流式传输的简单 FastAPI Web 应用程序
Next Steps¶
下一步¶
- Explore the Streaming documentation for more advanced features 探索流式传输文档以了解更高级的功能
- Learn about Session Management 了解会话管理
- Check out the API Reference 查看API 参考