AgentKit Runtime:从本地到云端的 Agent 部署利器
AgentKit Runtime 是一个统一的、托管式的 Agent 运行环境,旨在为您提供一个安全、高效、可扩展的 Serverless 平台,让您能专注于 Agent 的业务逻辑,而非繁琐的底层运维。
AgentKit Runtime 为您解决什么?
- 运维负担重:您是否曾为环境搭建、网络配置、依赖安装、手动扩缩容和故障排查而耗费大量精力?
- 缺乏统一管理平台:为不同的 Agent 或会话创建和管理独立的运行环境,是否让您感到调度混乱、难以维护?
- 安全与隔离挑战:您是否担心多用户或多 Agent 场景下的权限冲突、数据泄露或恶意访问风险?
- 监控与日志缺失:您是否需要花费额外成本来配置日志系统、指标收集和告警规则,以便实时监控 Agent 的健康状况?
AgentKit Runtime 通过提供一个 Serverless 的托管环境,系统性地解决了以上所有问题。
核心优势
1. Serverless 架构:极致简化运维,优化成本
- 零基础设施管理:无需关心底层服务器和复杂的环境配置,实现真正的“代码即服务”。
- 统一运行时平台:在同一环境中高效运行和调度多个 Agent 实例(Session),并提供统一的接入点。
- 自动化生命周期管理:根据实时流量从零自动扩展到大规模并发,实现无人值守的弹性伸缩。
- 按量付费:仅在 Agent 运行时产生费用,无流量时资源自动缩减至零,最大限度节省云成本。
- 内建监控与日志:提供开箱即用的实时监控和日志系统,轻松追踪 Agent 运行状态,快速定位问题。
2. 企业级安全隔离
- 多重隔离保障:基于云厂商成熟的虚拟化和容器技术,确保每个 Agent Runtime 都在独立、受保护的环境中运行。
- 精细化权限控制:与 Identity 和网关鉴权深度集成,严格隔离会话上下文与访问权限,有效防止数据串扰和越权风险。
3. 框架无关的快速集成
- 广泛的框架支持:深度融合 veADK,同时兼容主流的 Python Agent 框架,仅需少量适配即可快速接入。
- 多协议支持:原生支持 HTTP、MCP、A2A 等多种通信协议,满足不同应用场景的需求。
- 一键部署:通过 AgentKit SDK,您可以轻松实现从本地 IDE 到云端 Runtime 的一键部署。
三步上手:从本地开发到云端部署
本指南将引导您完成从环境配置、本地开发调试,到最终将 Agent 部署到生产环境的全过程。
前提条件
在开始之前,请确保您已准备好:
- 一个已开通 AgentKit 服务的火山引擎账号,并获取其 AccessKey & SecretKey。
提示:请确保您的 AccessKey 拥有
AgentKitFullAccess权限。 - 本地已安装 Python 3.10+。
第 1 步:环境配置
我们推荐使用
uv或python3 -m venv来管理您的 Python 虚拟环境。以下示例将使用uv。
uv安装指南请参考:uv installation
1. 创建并激活虚拟环境
mkdir agentkit-runtime-quickstart
cd agentkit-runtime-quickstart
uv venv --python 3.12.0
source .venv/bin/activate2. 安装核心依赖
您可以通过
--index参数指定镜像源以加快安装速度。
uv add veadk-python
uv add agentkit-sdk-python第 2 步:创建并本地调试 Agent
1. 初始化 Agent 项目
我们强烈推荐您使用 agentkit-cli 来快速启动项目。init 命令可以从模板创建一个结构完整的、支持流式输出的 Agent 项目。
# 使用 agentkit-cli 从模板创建项目
agentkit init这会自动生成一个包含 simple_agent.py 和 requirements.txt 的项目结构,让您能立即开始开发。
2. 智能体代码概览
使用 basic_stream 模板生成的 simple_agent.py 核心代码如下。它展示了如何基于 AgentkitSimpleApp 包装一个 veADK 的 Agent,并通过 Runner 执行,最终以 SSE(Server-Sent Events)事件流返回结果。
# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import logging
from google.adk.agents import RunConfig
from google.adk.agents.run_config import StreamingMode
from google.genai.types import Content, Part
from veadk import Agent, Runner
from agentkit.apps import AgentkitSimpleApp
from veadk.prompts.agent_default_prompt import DEFAULT_DESCRIPTION, DEFAULT_INSTRUCTION
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
app = AgentkitSimpleApp()
app_name = "simple_streamable_app"
agent_name = "Agent"
description = DEFAULT_DESCRIPTION
system_prompt = DEFAULT_INSTRUCTION
tools = []
# from veadk.tools.builtin_tools.web_search import web_search
# tools.append(web_search)
agent = Agent(
name=agent_name,
description=description,
instruction=system_prompt,
tools=tools,
)
agent.model._additional_args["stream_options"] = {"include_usage": True}
runner = Runner(agent=agent, app_name=app_name)
@app.entrypoint
async def run(payload: dict, headers: dict):
prompt = payload["prompt"]
user_id = headers["user_id"]
session_id = headers["session_id"]
logger.info(
f"Running agent with prompt: {prompt}, user_id: {user_id}, session_id: {session_id}"
)
session_service = runner.short_term_memory.session_service # type: ignore
# prevent session recreation
session = await session_service.get_session(
app_name=app_name, user_id=user_id, session_id=session_id
)
if not session:
await session_service.create_session(
app_name=app_name, user_id=user_id, session_id=session_id
)
new_message = Content(role="user", parts=[Part(text=prompt)])
try:
async for event in runner.run_async(
user_id=user_id,
session_id=session_id,
new_message=new_message,
run_config=RunConfig(streaming_mode=StreamingMode.SSE),
):
# Format as SSE data
sse_event = event.model_dump_json(exclude_none=True, by_alias=True)
logger.debug("Generated event in agent run streaming: %s", sse_event)
yield sse_event
except Exception as e:
logger.exception("Error in event_generator: %s", e)
# You might want to yield an error event here
error_data = json.dumps({"error": str(e)})
yield error_data
@app.ping
def ping() -> str:
return "pong!"
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)代码亮点:
- 默认使用火山方舟的豆包模型,并自动处理模型 API Key 的获取。
- 内置了会话管理逻辑,确保多轮对话的连续性。
- 通过
AgentkitSimpleApp实现了与 AgentKit Runtime 的无缝集成。
3. 配置环境变量
在启动 Agent 服务之前,您需要配置火山引擎的访问凭证:
# 必需:配置您的火山引擎访问凭证
export VOLCENGINE_ACCESS_KEY=your_ak
export VOLCENGINE_SECRET_KEY=your_sk重要提示:请务必将
your_ak和your_sk替换为您的真实凭证。
4. 本地启动并调用服务
一切就绪后,运行以下命令启动 Agent 服务:
python simple_agent.py服务将监听 http://0.0.0.0:8000。现在,打开另一个终端,使用 curl 来测试您的 Agent:
curl --location 'http://localhost:8000/invoke' \
--header 'Content-Type: application/json' \
--header 'user_id: test' \
--header 'session_id: local_session' \
--data '{"prompt": "讲一个笑话!"}'参数说明:
user_id:用户标识符session_id:会话标识符prompt:用户的自然语言请求
**执行日志示例: **
调用成功后,控制台会输出详细的执行日志:
INFO: Started server process [38120]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO: 127.0.0.1:60807 - "POST /invoke HTTP/1.1" 200 OK
INFO:__main__:Running agent with prompt: 讲一个笑话!, user_id: test, session_id: local_session
2025-12-01 20:12:50 | DEBUG | runner.py:151 - Event output: Why
2025-12-01 20:12:50 | DEBUG | runner.py:151 - Event output: did
2025-12-01 20:12:50 | DEBUG | runner.py:151 - Event output: the
2025-12-01 20:12:50 | DEBUG | runner.py:151 - Event output: AI
2025-12-01 20:12:50 | DEBUG | runner.py:151 - Event output: apply
2025-12-01 20:12:50 | DEBUG | runner.py:151 - Event output: for
2025-12-01 20:12:50 | DEBUG | runner.py:151 - Event output: a
2025-12-01 20:12:50 | DEBUG | runner.py:151 - Event output: job
2025-12-01 20:12:50 | DEBUG | runner.py:151 - Event output: at
2025-12-01 20:12:50 | DEBUG | runner.py:151 - Event output: the
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: bakery
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: ?
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: Because
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: it
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: wanted
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: to
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: work
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: on
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: its
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: **
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: d
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: ough
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: **
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: processing
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: skills
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: !
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: 😄
2025-12-01 20:12:51 | DEBUG | runner.py:151 - Event output: Why did the AI apply for a job at the bakery?
Because it wanted to work on its **dough** processing skills! 😄第 3 步:部署到生产环境
完成本地开发和测试后,您可以将智能体部署到 AgentKit 平台:
- 打包代码:确保所有依赖都已正确配置在
requirements.txt中,使用模版生成的代码默认已经包含了所有必要的依赖。 - 平台部署:根据[快速开始]指南,通过
agentkit config和agentkit launch命令,配置和部署应用。 - 调用应用:部署完成后,您可以通过平台提供的 API 或 SDK 调用智能体。调用示例:bash
agentkit invoke "讲一个笑话!"
至此,您已成功完成了从本地开发到云端部署的全过程!
