多彩编程 多彩编程MZPH · CODE BLOG
ARTICLE DETAIL

文章详情

深耕前端与后端开发技术的一线实战笔记与踩坑复盘。

Agno × Google Vertex AI:在云端运行 Anthropic Claude Agent 的完整实战指南

Agno × Google Vertex AI:在云端运行 Anthropic Claude Agent 的完整实战指南 人工智能大模型AI AgentAgent 框架多智能体工具调用RAGAgent 工作流【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址https://gitcode.com/GitHub_Trending/ag/agno点击查看免费下载导读本文以 agno 开源仓库的 cookbook/90_models/vertexai 系列示例为主线系统讲解如何在 Google Cloud Vertex AI 平台上通过 agno 构建、运行和管理基于 Anthropic Claude 的 Agent。读者将掌握Vertex AI 的环境准备与认证流程、Claude 模型在 agno 中的接入方式、工具调用/结构化输出/存储/知识库/记忆/多模态输入/扩展思考等核心能力的配置方法以及提示词缓存、重试机制、超时控制、Beta 特性等生产级细节最终能够在自己的 Google Cloud 项目中直接复制运行这些示例。一、目录概览从示例到源码的完整闭环仓库在cookbook/90_models/vertexai下提供了两类内容总览 READMEcookbook/90_models/vertexai/README.md给出运行方式——所有示例均可通过以下命令直接执行.venvs/demo/bin/python cookbook/90_models/vertexai/example.pyClaude 子目录cookbook/90_models/vertexai/claude/README.md包含从环境准备到十余个能力示例的完整步骤说明。配套示例脚本与源码、测试位于角色路径示例脚本cookbook/90_models/vertexai/claude/ 目录下 17 个.py文件模型实现libs/agno/agno/models/vertexai/claude.py基类实现libs/agno/agno/models/anthropic/claude.py集成测试libs/agno/tests/integration/models/vertexai/claude/其中 TEST_LOG.md 说明当前尚未记录目录级 cookbook 测试但源码级集成测试如 test_basic.py、test_thinking.py已覆盖基本对话、流式输出、结构化输出、工具调用、思考、缓存等场景。二、环境准备虚拟环境、环境变量与认证1. 创建并激活虚拟环境python3 -m venv ~/.venvs/aienv source ~/.venvs/aienv/bin/activate2. 导出环境变量Vertex AI 的 Anthropic 网关需要两个关键环境变量export GOOGLE_CLOUD_PROJECTyour-project-id export CLOUD_ML_REGIONyour-region3. 认证 CLIgcloud auth application-default login提示若需要其他认证方式可参考 Google Cloud 官方文档使用服务账号或 Workload Identity 等替代方案。4. 安装依赖uv pip install -U anthropic ddgs duckdb agno其中anthropic是调用 Vertex AI Claude 网关的 SDK。在源码层libs/agno/agno/models/vertexai/claude.py 第 11-14 行会显式导入AnthropicVertex与AsyncAnthropicVertex若未安装anthropic会直接抛出ImportError并提示pip install anthropic。三、模型接入原理agno.models.vertexai.claude.Claude的实现细节1. 类结构libs/agno/agno/models/vertexai/claude.py 中定义了Claude类它继承自agno.models.anthropic.Claude即 libs/agno/agno/models/anthropic/claude.py属于 VertexAI 适配子类dataclass class Claude(AnthropicClaude): id: str claude-sonnet-420250514 name: str Claude provider: str VertexAI默认模型 ID 为claude-sonnet-420250514示例中大量使用provider 标识为VertexAI便于日志与指标区分。2. 客户端参数与凭据解析_get_client_params()方法负责组装 Vertex AI 客户端所需参数优先级为「显式字段 环境变量」region来自字段或CLOUD_ML_REGION环境变量project_id来自字段或ANTHROPIC_VERTEX_PROJECT_ID环境变量base_url来自字段或ANTHROPIC_VERTEX_BASE_URL环境变量timeout若模型设置了超时则一并传入额外参数与默认请求头可分别通过client_params与default_headers注入。get_client()/get_async_client()会缓存客户端实例重复请求复用同一连接若未自定义http_client每个模型实例拥有独立连接可避免多模型主 Agent、MemoryManager 等并发时 HTTP/2 流饱和。3. VertexAI 结构化输出差异__post_init__中有一处关键覆盖self.supports_native_structured_outputs False self.supports_json_schema_outputs False即 VertexAI 通道上的 Claude关闭了原生结构化输出能力结构化输出output_schema将由 Agno 以提示词/JSON 模式等间接方式实现这与直连 Anthropic API 的行为不同是使用前需要知晓的重要差异。四、快速开始Basic Agent同步 / 流式 / 异步1. 完整示例basic.py 演示了最小可用的 Agentimport asyncio from agno.agent import Agent from agno.models.vertexai.claude import Claude agent Agent(modelClaude(idclaude-sonnet-420250514), markdownTrue) if __name__ __main__: # --- 同步 --- agent.print_response(Share a 2 sentence horror story) # --- 同步 流式 --- agent.print_response(Share a 2 sentence horror story, streamTrue) # --- 异步 --- asyncio.run(agent.aprint_response(Share a 2 sentence horror story)) # --- 异步 流式 --- asyncio.run(agent.aprint_response(Share a 2 sentence horror story, streamTrue))2. 四种运行模式对照模式调用方式说明同步agent.print_response(...)阻塞等待完整回复同步 流式agent.print_response(..., streamTrue)边生成边输出异步agent.aprint_response(...)基于asyncio.run的非阻塞调用异步 流式agent.aprint_response(..., streamTrue)异步流式若需把结果存入变量可改用run: RunOutput agent.run(...)再通过run.content访问文本内容。3. 测试佐证集成测试 test_basic.py 验证了单次对话产生[system, user, assistant]三段消息同步/异步、流式/非流式四种路径均返回非空内容每次响应的input_tokens、output_tokens、total_tokens均大于 0且total_tokens input_tokens output_tokens客户端实例在同一模型上被持久复用test_client_persistence断言三次调用共享同一 client 对象。五、Agent with Tools工具调用tool_use.py 演示了如何为 VertexAI Claude Agent 挂载网络搜索工具import asyncio from agno.agent import Agent from agno.models.vertexai.claude import Claude from agno.tools.websearch import WebSearchTools agent Agent( modelClaude(idclaude-sonnet-420250514), tools[WebSearchTools()], markdownTrue, ) if __name__ __main__: agent.print_response(Whats happening in France?) agent.print_response(Whats happening in France?, streamTrue) asyncio.run(agent.aprint_response(Whats happening in France?, streamTrue))运行前需执行uv pip install ddgs安装 DuckDuckGo 搜索依赖。在底层libs/agno/agno/models/anthropic/claude.py 的_parse_provider_response会检测stop_reason tool_use将响应中的tool_use块转换为 Agno 统一的tool_calls结构含id、type: function、name与序列化参数流式场景则通过tool_usecontent block 的 stop 事件捕获。集成测试 test_tool_use.py 用YFinanceTools验证了工具调用与流式工具事件ToolCallStarted/ToolCallCompleted的触发。六、Agent with Structured Output结构化输出structured_output.py 演示了用 Pydantic 模型约束输出格式from typing import List from agno.agent import Agent from agno.models.vertexai.claude import Claude from pydantic import BaseModel, Field class MovieScript(BaseModel): setting: str Field(..., descriptionProvide a nice setting for a blockbuster movie.) ending: str Field(..., descriptionEnding of the movie. If not available, provide a happy ending.) genre: str Field(..., descriptionGenre of the movie. If not available, select action, thriller or romantic comedy.) name: str Field(..., descriptionGive a name to this movie) characters: List[str] Field(..., descriptionName of characters for this movie.) storyline: str Field(..., description3 sentence storyline for the movie. Make it exciting!) movie_agent Agent( modelClaude(idclaude-sonnet-420250514), descriptionYou help people write movie scripts., output_schemaMovieScript, ) run movie_agent.run(New York) pprint(run.content)要点说明通过output_schemaMovieScript声明输出结构PydanticField(..., description...)的描述会被用于引导模型生成由于 VertexAI 通道关闭了原生结构化输出Agno 会走提示词 JSON 解析路径use_json_modeTrue可显式启用 JSON 模式解析结果通过run.content返回为MovieScript实例流式输出同样可用movie_agent.print_response(New York, streamTrue)。测试佐证test_structured_output与test_json_response_mode见 test_basic.py分别验证了默认模式与 JSON 模式下返回内容均为合法的MovieScript实例。七、Agent with Storage持久化会话db.py 演示了使用 SQLite 存储会话历史from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.vertexai.claude import Claude from agno.tools.websearch import WebSearchTools db SqliteDb(db_filetmp/data.db) agent Agent( modelClaude(idclaude-sonnet-420250514), dbdb, tools[WebSearchTools()], add_history_to_contextTrue, ) agent.print_response(How many people live in Canada?) agent.print_response(What is their national anthem called?)dbSqliteDb(db_filetmp/data.db)会话持久化到本地 SQLiteadd_history_to_contextTrue把历史消息作为上下文传入使第二个问题“国歌叫什么”能利用第一轮查询结果依赖uv pip install ddgs sqlalchemy anthropic。测试中对应的存储场景test_with_memory、test_history断言第二轮回答包含第一轮记忆的信息如名字并验证消息按[system, user, assistant, user, assistant]递增累积。八、Agent with Knowledge知识库增强knowledge.py 演示了知识库检索增强from agno.agent import Agent from agno.knowledge.embedder.openai import OpenAIEmbedder from agno.knowledge.knowledge import Knowledge from agno.models.vertexai.claude import Claude from agno.vectordb.pgvector import PgVector db_url postgresqlpsycopg://ai:ailocalhost:5532/ai knowledge Knowledge( vector_dbPgVector( table_namerecipes, db_urldb_url, embedderOpenAIEmbedder(), ), ) knowledge.insert(urlhttps://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf) agent Agent( modelClaude(idclaude-sonnet-420250514), knowledgeknowledge, ) agent.print_response(How to make Thai curry?, markdownTrue)重要前提README 明确提示Claude 在知识库场景下默认使用 OpenAI 嵌入模型因此必须额外导出 OpenAI API Keyexport OPENAI_API_KEY***依赖uv pip install ddgs sqlalchemy pgvector pypdf anthropic openai向量库为本地pgvector如 scripts/run_pgvector.sh 启动的容器知识内容通过 URL 插入 PDF泰式食谱。九、Agent with Memory个性化记忆与会话摘要memory.py 演示了长期记忆与摘要能力步骤为运行./cookbook/scripts/run_pgvector.sh启动带 pgvector 的 PostgreSQL 容器安装依赖uv pip install anthropic sqlalchemy psycopg[binary] pgvector运行python cookbook/90_models/vertexai/claude/memory.py。核心代码from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.models.vertexai.claude import Claude db_url postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url) agent Agent( modelClaude(idclaude-sonnet-420250514), dbdb, # 数据库持久化 update_memory_on_runTrue, # 每次运行更新记忆 enable_session_summariesTrue, # 启用会话摘要 ) agent.print_response(My name is john billings?, streamTrue) agent.print_response(I live in nyc?, streamTrue) agent.print_response(Im going to a concert tomorrow?, streamTrue) agent.print_response(What have we been talking about, do you know my name?, streamTrue)示例中先分享个人信息再追问“我们聊了什么 / 你知道我的名字吗”验证记忆被持久化并可被召回。十、Agent that Analyzes an Image多模态图像输入VertexAI Claude 支持图像输入示例提供 URL 与字节两种方式。1. URL 方式image_input_url.pyfrom agno.agent import Agent from agno.media import Image from agno.models.vertexai.claude import Claude from agno.tools.websearch import WebSearchTools agent Agent( modelClaude(idclaude-sonnet-420250514), tools[WebSearchTools()], markdownTrue, ) agent.print_response( Tell me about this image and search the web for more information., images[Image(urlhttps://fastly.picsum.photos/id/237/200/300.jpg)], streamTrue, )2. 字节方式image_input_bytes.py 先下载图片再以字节传入from pathlib import Path from agno.agent import Agent from agno.media import Image from agno.models.vertexai.claude import Claude from agno.utils.media import download_image image_path Path(__file__).parent.joinpath(sample.jpg) download_image( urlhttps://upload.wikimedia.org/wikipedia/commons/0/0c/GoldenGateBridge-001.jpg, output_pathstr(image_path), ) image_bytes image_path.read_bytes() agent Agent(modelClaude(idclaude-sonnet-420250514), markdownTrue) agent.print_response( Tell me about this image and give me the latest news about it., images[Image(contentimage_bytes)], streamTrue, )对应集成测试 test_multimodal.py 验证了模型能够正确识别金门大桥图片内容。十一、Thinking扩展思考Extended Thinking1. 开启扩展思考thinking.py 展示了通过thinking参数开启深度思考from agno.agent import Agent from agno.models.vertexai.claude import Claude agent Agent( modelClaude( idclaude-sonnet-420250514, max_tokens2048, thinking{type: enabled, budget_tokens: 1024}, ), markdownTrue, ) agent.print_response(Share a very scary 2 sentence horror story) agent.print_response(Share a very scary 2 sentence horror story, streamTrue)参数说明thinking{type: enabled, budget_tokens: 1024}开启思考并设定思考 token 预算max_tokens需为思考 输出预留足够空间示例设为 2048思考内容会体现在响应的reasoning_content中流式场景同样逐段返回。源码佐证基类 libs/agno/agno/models/anthropic/claude.py 定义了不支持思考的模型黑名单如 Claude 3 Haiku 系列并对启用思考的请求调用_validate_thinking_support()做校验流式解析中thinking_delta事件会累积为reasoning_content。2. 思考 工具 存储的组合验证集成测试 test_thinking.py 覆盖了思考的完整形态普通思考、流式思考、异步思考思考 工具调用YFinanceTools查询股价思考内容随会话持久化JsonDb 存储reasoning_contentInterleaved ThinkingClaude 4 专属通过betas[interleaved-thinking-2025-05-14]与default_headers{anthropic-beta: ...}启用思考块与文本块交错输出。十二、Adaptive Thinking自适应思考Claude 4.6README 重点介绍了 Claude 4.6 VertexAI 模型的自适应思考能力通过output_config中的effort参数控制思考深度对应示例 adaptive_thinking.pyfrom agno.agent import Agent from agno.models.vertexai.claude import Claude agent Agent( modelClaude( idclaude-sonnet-4-620250514, max_tokens4096, thinking{type: adaptive}, output_config{effort: high}, ), markdownTrue, ) agent.print_response( Explain the key differences between recursion and iteration, and when you would choose one over the other in software development. ) agent.print_response( What are the trade-offs between microservices and monolithic architectures?, streamTrue, )合法的effort取值README 原表取值说明low最高效显著节省 tokenmedium均衡方案适度节省high默认值适合复杂推理的高能力max绝对最高能力仅 Opus 4.6实现上get_request_params()见 libs/agno/agno/models/vertexai/claude.py会把thinking与output_config原样透传给 Vertex AI 网关因此output_config{effort: ...}可直接生效。十三、Prompt Caching系统提示词缓存prompt_caching.py 演示了如何缓存大型系统提示词以降低延迟与成本from pathlib import Path from agno.agent import Agent from agno.models.vertexai.claude import Claude from agno.utils.media import download_file txt_path Path(__file__).parent.joinpath(system_prompt.txt) download_file(https://agno-public.s3.amazonaws.com/prompts/system_promt.txt, str(txt_path)) system_message txt_path.read_text() agent Agent( modelClaude( idclaude-sonnet-420250514, cache_system_promptTrue, # 开启系统提示词缓存 ), system_messagesystem_message, markdownTrue, ) # 第一次运行创建缓存 response agent.run(Explain the difference between REST and GraphQL APIs with examples) if response and response.metrics: print(fFirst run cache write tokens {response.metrics.cache_write_tokens}) # 第二次运行命中缓存 response agent.run(What are the key principles of clean code and how do I apply them in Python?) if response and response.metrics: print(fSecond run cache read tokens {response.metrics.cache_read_tokens})关键点cache_system_promptTrue触发 Anthropic 风格的前缀缓存适用于静态且较大的系统提示词首次运行写入缓存cache_write_tokens 0第二次运行读取缓存cache_read_tokens 0缓存建立需数秒第二次运行时才能体现收益测试 test_prompt_caching.py 断言了系统消息中cache_control: {type: ephemeral}的注入以及extended_cache_timeTrue时ttl: 1h的扩展缓存test_basic.py 中的test_prompt_caching验证了两次运行间 write→read 的转换。十四、重试与超时生产可用性配置1. 请求重试retry.py 演示了重试参数from agno.agent import Agent from agno.models.vertexai.claude import Claude wrong_model_id vertexai-claude-wrong-id agent Agent( modelClaude( idwrong_model_id, # 故意使用错误 ID 触发重试 retries3, # 请求重试次数 delay_between_retries1, # 重试间隔秒 exponential_backoffTrue, # 启用指数退避间隔逐次翻倍 ), ) agent.print_response(What is the capital of France?)参数速查参数作用retries请求失败后的重试次数delay_between_retries每次重试间的固定延迟秒exponential_backoff开启后延迟按指数翻倍2. 请求超时basic_with_timeout.py 演示超时控制from agno.agent import Agent from agno.models.vertexai.claude import Claude agent Agent(modelClaude(idclaude-sonnet-420250514, timeout5), markdownTrue) agent.print_response(Share a 2 sentence horror story)timeout5限制单次请求 5 秒超时即失败。在_get_client_params()中timeout会被透传给AnthropicVertex客户端。十五、Beta 特性通过betas参数启用实验能力betas.py 演示了如何启用 Anthropic Beta 特性import anthropic from agno.agent import Agent from agno.models.vertexai.claude import Claude betas [context-management-2025-06-27] model Claude(idclaude-sonnet-420250514, betasbetas) agent Agent(modelmodel) if __name__ __main__: all_betas anthropic.types.AnthropicBetaParam print(\n All available Anthropic beta features ) beta_lines \n- .join(str(b) for b in all_betas.__args__[1].__args__) print(f- {beta_lines}) agent.print_response( My name is John Doe and I live in New York City. I like to bike and hike in the Catskill Mountains. )实现上get_request_params()会将betas列表原样放入请求参数见 libs/agno/agno/models/vertexai/claude.py。示例中的context-management-2025-06-27为上下文管理 Beta示例脚本同时打印当前anthropic版本所有可用 Beta 特性清单方便查阅。测试中亦可见interleaved-thinking-2025-05-14这类思考相关的 Beta 头test_thinking.py。十六、PDF 输入文件理解VertexAI Claude 支持 PDF 文件输入示例提供字节与本地路径两种方式均基于 agno.media.Filefrom agno.media import File1. 字节方式pdf_input_bytes.pyfrom pathlib import Path from agno.agent import Agent from agno.media import File from agno.models.vertexai.claude import Claude from agno.utils.media import download_file pdf_path Path(__file__).parent.joinpath(ThaiRecipes.pdf) download_file(https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf, str(pdf_path)) agent Agent(modelClaude(idclaude-sonnet-420250514), markdownTrue) agent.print_response( Summarize the contents of the attached file., files[File(contentpdf_path.read_bytes())], )2. 本地路径方式pdf_input_local.py 与字节方式几乎一致区别仅在传入File(filepathpdf_path)agent.print_response( Summarize the contents of the attached file., files[File(filepathpdf_path)], )两种方式均先通过download_file将示例 PDF 下载到脚本同级目录再交给 Agent 解析与总结。十七、示例运行清单速查以下命令均需在仓库根目录、且已完成第二节环境准备的前提下执行也可用.venvs/demo/bin/python替换python与 总览 README 保持一致场景命令基础对话python cookbook/90_models/vertexai/claude/basic.py工具调用联网搜索python cookbook/90_models/vertexai/claude/tool_use.py结构化输出python cookbook/90_models/vertexai/claude/structured_output.py会话存储SQLitepython cookbook/90_models/vertexai/claude/db.py知识库增强需OPENAI_API_KEYpython cookbook/90_models/vertexai/claude/knowledge.py记忆与会话摘要python cookbook/90_models/vertexai/claude/memory.py图像分析URLpython cookbook/90_models/vertexai/claude/image_input_url.py扩展思考python cookbook/90_models/vertexai/claude/thinking.py自适应思考Claude 4.6python cookbook/90_models/vertexai/claude/adaptive_thinking.py提示词缓存python cookbook/90_models/vertexai/claude/prompt_caching.py重试机制python cookbook/90_models/vertexai/retry.py超时控制python cookbook/90_models/vertexai/claude/basic_with_timeout.pyBeta 特性python cookbook/90_models/vertexai/claude/betas.pyPDF 输入python cookbook/90_models/vertexai/claude/pdf_input_bytes.py/pdf_input_local.py图像输入字节python cookbook/90_models/vertexai/claude/image_input_bytes.py十八、常见问题与注意事项结构化输出差异VertexAI 通道的 Claude 关闭了原生结构化输出supports_native_structured_outputs False结构化输出走 JSON 模式返回类型与直连 Anthropic API 时不同。知识库需额外密钥knowledge.py场景中 Claude 使用 OpenAI 嵌入务必先export OPENAI_API_KEY***否则会因缺少嵌入服务而失败。思考 token 预算开启thinking后max_tokens需同时覆盖思考与输出预算不足会导致截断。缓存生效需要两次请求提示词缓存首次运行创建、第二次运行命中且仅在系统提示词较大时收益明显。环境变量优先级显式传入region/project_id/base_url会覆盖环境变量便于多项目、多区域切换。客户端复用模型实例的 client 会被缓存复用多 Agent 并发时建议让每个模型实例持有独立连接避免 HTTP/2 流饱和。至此从环境搭建、模型接入原理到工具/输出/存储/知识/记忆/多模态/思考/缓存/重试/Beta 等全部能力均可在 cookbook/90_models/vertexai 目录下找到可复制的示例并结合 libs/agno/agno/models/vertexai/claude.py 与集成测试深入理解其实现细节。赞分享人工智能大模型AI AgentAgent 框架多智能体工具调用RAGAgent 工作流【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址https://gitcode.com/GitHub_Trending/ag/agno点击查看免费下载相关推荐在 VoltAgent 中使用 Vertex (Anthropic)通过 google-vertex-anthropic 路由调用 Claude 模型在 VoltAgent 中使用 Vertex Anthropic 通过 google vertex anthropic 路由调用 Claude 模型 本文以人工智能AI AgentAgent 框架后端多智能体RAG工具调用Agent 记忆Agent 工作流AI 评测MCP 服务MCP Clients语音ruflo 云上 Agent 运行时实战指南用 managed-agent 技能驱动 Anthropic Claude Managed Agentsruflo 云上 Agent 运行时实战指南用 managed agent 技能驱动 Anthropic Claude Managed Agents 导读 r人工智能AI Agent多智能体Agent 编排Agent 记忆工具调用代码智能体MCP 服务AI 评测用 VoltAgent 连接 Google Vertex AI在 GCP 项目中运行 Gemini 模型的完整实战指南用 VoltAgent 连接 Google Vertex AI在 GCP 项目中运行 Gemini 模型的完整实战指南 VoltAgent 是一个开源 Typ人工智能AI AgentAgent 框架后端多智能体RAG工具调用Agent 记忆Agent 工作流AI 评测MCP 服务MCP Clients语音上一篇Avahi安全配置保护mDNS网络免受常见攻击的实用策略下一篇Fay开源项目女性开发者访谈探索数字人框架的技术魅力创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表