【技术干货】从 Anthropic Cloud Managed Agents 看下一代 AI 代理架构(含完整 Python 接入示例)
摘要
Anthropic 推出的 Cloud Managed Agents,本质上是一个“云托管 AI 代理平台”:内置 Agent Loop、工具执行层、长时任务调度、上下文压缩与性能优化。本文从架构原理出发,拆解其核心能力,类比如何在自己项目中用“托管式 Agent 思路”落地,并给出基于薛定猫 AI(OpenAI 兼容)的一套可运行 Python 代码示例,帮助你快速构建生产级 AI 代理服务。
一、背景介绍:从“调用模型”到“托管代理”
过去一年,Agent 相关的开源项目(如 LangChain Agents、AutoGen、OpenAI o1 Agents 类方案)层出不穷,但普遍存在几个落地痛点:
-
Agent Loop 需自研:
- 自己实现“思考 -> 调用工具 -> 再思考 -> 再调用工具”的迭代流程
- 容错、重试、超时、日志、可观测性都要自己处理
-
长时任务难以可靠运行:
- 多小时任务需要额外的队列、调度系统(Celery、Airflow 等)
- 中途中断、任务恢复、状态持久化都需要额外工程方案
-
上下文成本高昂:
- 多轮工具调用 + 长上下文 = Token 成本爆炸
- 需要手写“摘要+压缩”策略,极易出错
Anthropic 的 Cloud Managed Agents 试图解决的,是**从“自己写 Agent 框架”到“直接在托管平台上配置并部署 Agent”**的范式转变:
- 提供 预构建、可配置的 Agent Harness(代理框架)
- 运行在 托管基础设施 上,原生支持:
- 文件读取
- 代码执行
- Web 浏览
- 命令执行(沙箱内安全运行)
- 内置:
- Prompt Caching(提示缓存)
- Context Compaction(上下文压缩)
- 性能优化与质量控制
对开发者的意义:不再从零搭一个 Agent Runtime,而是**把精力放到“定义能力 + 接入业务”**上。
二、核心原理:托管式 AI 代理的关键能力拆解
2.1 托管 Agent Harness:统一的“智能体运行时”
Cloud Managed Agents 提供的是一个统一的代理运行容器,其核心职责包括:
- Agent Loop 编排:负责多轮推理、工具调用与决策
- 工具执行层:在沙箱中执行代码 / 命令,或通过 MCP / HTTP 调用外部系统
- 状态管理:维护会话、任务状态、运行日志
- 长时任务调度:支持异步与长时后台运行(如批量文档处理、复杂研究任务)
这种模式在企业里非常重要——你只需:
- 声明 Agent 的职责(system prompt / description)
- 声明可用工具 / 外部系统(Box API、Slack、Notion、自建 API 等)
- 把 Agent 暴露给业务侧(Webhook、API、应用内集成)
其余的循环、调度和资源管理由平台托管。
2.2 智能优化:Prompt Cache + 上下文压缩
视频中提到的几个关键能力,其实解决的是传统 Agent 的两个工程痛点:
-
Prompt Caching
- 对重复调用的指令 / context 增加缓存
- 减少模型重复工作,降低成本
-
Context Compaction
- 将长对话、长任务中的历史信息压缩为摘要
- 保留语义关键信息,减少 token 长度
-
性能优化(QoS / Latency / Cost)
- 不同任务自动选择合适模型(如为深度研究自动选 Anthropic Opus 4.6)
- 在延迟、成本、推理深度之间自动协调
这些能力的本质都是:通过平台统一做“跨任务的经验复用与最优策略搜索”,而不是让每个团队各自重复造轮子。
2.3 MCP、外部 API 与企业工作流集成
视频中的示例场景本质上都是**“Agent + 工具 / API + 长时任务”**:
- 从 Box 拉取发票和采购订单 → 校验明细 → 生成对账报告
- Slack + Notion 构建内部知识库问答 / 支持 Agent
- 深度研究 Agent:自动检索 Web / 知识库,生成 Markdown 报告
关键技术点:
- **工具(Tools / MCP Servers)**作为 Agent 的“手脚”
- Agent 负责:
- 何时调用工具
- 如何解析结果
- 如何组织最终输出
这种模式非常适合企业的端到端自动化工作流:文档审核、报表生成、邮件自动回复、销售线索跟进等。
三、实战演示:用“托管式思路”在 Python 中构建一个长时研究 Agent
虽然 Cloud Managed Agents 本身是 Anthropic 的托管平台,但在自己的项目中我们可以复刻类似的**“托管 Agent 思路”**:把 Agent Runtime 单独做成一个服务,业务侧只需要调用一个统一接口。
下面用薛定猫 AI(xuedingmao.com)来做一个简化版的“深度研究 Agent”示例:
- 使用 OpenAI 兼容接口(URL + API Key)
- 模型:
claude-sonnet-4-6(类比视频中的 Opus 4.6,擅长长推理) - 支持:
- 接收一个研究主题
- 自动分解为多个小任务
- 输出结构化 Markdown 报告
- 以“托管式”方式提供 HTTP API,供其他服务调用
3.1 环境准备
pip install requests fastapi uvicorn
3.2 核心代码示例(可直接运行)
# filename: research_agent_service.py
# 一个简化版“托管代理”服务示例,基于薛定猫 AI 的 OpenAI 兼容接口
import os
import time
import uuid
from typing import List
import requests
from fastapi import FastAPI
from pydantic import BaseModel
# ========= 配置区域 =========
# 薛定猫 AI 平台:https://xuedingmao.com
# 后台创建 API Key 后,填入环境变量 XUEDINGMAO_API_KEY
XUEDINGMAO_API_KEY = os.getenv("XUEDINGMAO_API_KEY")
if not XUEDINGMAO_API_KEY:
raise RuntimeError("请先在环境变量中设置 XUEDINGMAO_API_KEY")
# OpenAI 兼容模式的 Base URL
BASE_URL = "https://xuedingmao.com/v1"
MODEL_NAME = "claude-sonnet-4-6"
# ========== 基础 HTTP 客户端封装 ==========
def call_llm(system_prompt: str, user_prompt: str) -> str:
"""
调用薛定猫 AI 的 Chat Completions 接口,返回模型文本输出。
相当于“托管代理框架”中的一次思考/决策步骤。
"""
url = f"{BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {XUEDINGMAO_API_KEY}",
"Content-Type": "application/json",
}
payload = {
"model": MODEL_NAME,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
"temperature": 0.2,
}
resp = requests.post(url, json=payload, headers=headers, timeout=60)
resp.raise_for_status()
data = resp.json()
return data["choices"][0]["message"]["content"]
# ========== 简化版“托管 Agent Loop”实现 ==========
SYSTEM_PROMPT = """
你是一个严谨的研究型 AI 代理,负责针对给定主题,
进行多步骤、高质量的研究,并输出结构化的 Markdown 报告。
约束:
1. 所有结论必须基于权威公开信息(期刊论文、权威媒体、行业报告等)。
2. 报告结构至少包括:概述、关键参与者、近期进展、技术/商业挑战、时间线与前景。
3. 内容要求:条理清晰、分点叙述、给出简要来源说明(不需要精确引用格式)。
4. 输出统一使用 Markdown 标题与列表。
"""
def plan_research_tasks(topic: str) -> List[str]:
"""
让模型先进行“任务拆解”,生成子问题列表,相当于 Agent 的“规划阶段”。
"""
user_prompt = f"""
请针对研究主题 `{topic}` 设计一个分步骤的研究计划。
输出格式为纯文本,多行,每行一个子任务或子问题,不要添加序号前缀。
"""
plan_text = call_llm(SYSTEM_PROMPT, user_prompt)
tasks = [line.strip("-• ").strip() for line in plan_text.splitlines() if line.strip()]
return tasks
def execute_research(topic: str, tasks: List[str]) -> str:
"""
按照任务列表执行研究。这里为了简化,实际仍然由大模型在一次调用中“模拟多步推理”。
在真实场景中,你可以:
- 为每个子任务单独调用 LLM
- 存储中间结果
- 调用外部搜索/数据接口
"""
joined_tasks = "\n".join(f"- {t}" for t in tasks)
user_prompt = f"""
研究主题:{topic}
你需要根据以下子任务列表进行系统性研究:
{joined_tasks}
请综合这些子任务的研究结果,输出一份完整的 Markdown 报告。
严格遵守系统提示中的结构要求。
"""
report = call_llm(SYSTEM_PROMPT, user_prompt)
return report
# ========== FastAPI 服务:模拟“云托管代理”的 API 接口 ==========
class ResearchRequest(BaseModel):
topic: str
class ResearchResponse(BaseModel):
task_id: str
report_markdown: str
created_at: float
app = FastAPI(title="Deep Research Agent Service (xuedingmao + Claude Sonnet 4.6)")
@app.post("/research", response_model=ResearchResponse)
def research_endpoint(req: ResearchRequest):
"""
供业务侧调用的统一接口:
- 输入:研究主题
- 内部:自动任务规划 + 执行
- 输出:Markdown 报告 + 任务 ID(便于持久化与追踪)
"""
task_id = str(uuid.uuid4())
created_at = time.time()
# 1) 规划阶段
tasks = plan_research_tasks(req.topic)
# 2) 执行阶段(这里简单同步执行;也可以改为异步队列)
report = execute_research(req.topic, tasks)
return ResearchResponse(
task_id=task_id,
report_markdown=report,
created_at=created_at,
)
if __name__ == "__main__":
import uvicorn
# 本地启动服务:uvicorn research_agent_service:app --reload
uvicorn.run(app, host="0.0.0.0", port=8000)
使用方式:
# 启动服务
export XUEDINGMAO_API_KEY="your_api_key_here"
python research_agent_service.py
# 或 uvicorn research_agent_service:app --reload
然后通过 curl 或 Postman 调用:
curl -X POST http://localhost:8000/research \
-H "Content-Type: application/json" \
-d '{"topic": "聚变能源发展的现状、关键参与者、近期突破与商业化前景"}'
返回结果包含:
task_id:模拟托管任务 IDreport_markdown:完整 Markdown 报告created_at:时间戳
这就是一个简化版“托管 Agent 服务”:业务只需通过 HTTP 丢一个主题过来,不关心内部如何调用 LLM、如何规划子任务。进一步你可以:
- 将任务持久化到数据库,实现“会话 / 任务中心”
- 增加外部搜索工具(如调用自建检索 API)
- 将报告自动写入 Confluence / Notion / 企业文档系统
四、注意事项与工程实践建议
4.1 长时任务与可靠性
- 使用消息队列(Kafka / RabbitMQ)或任务队列(Celery / RQ)处理长任务
- 持久化 Agent 状态(任务进度、子任务结果、错误日志)
- 增加超时、重试机制,避免单次 LLM 调用阻塞
4.2 成本控制与上下文管理
- 为 Agent 层统一实现:
- 历史对话摘要(类似 Cloud 的 context compaction)
- Prompt 模板复用与缓存
- 针对不同任务类型,动态选择模型(如快速问答用小模型,深度研究用大模型)
4.3 安全与权限边界
- 对“文件读取 / 命令执行 / 外部 API 调用”进行严格沙箱化和权限控制
- 对敏感数据使用脱敏 / 分级访问控制
- 为每类 Agent 设定“可见数据范围”和工具白名单
五、技术资源与工具推荐:多模型时代如何做技术选型
在落地类似 Cloud Managed Agents 这种架构时,一个核心问题是:如何在多模型环境下统一接入与管理?
这里推荐一个我日常在用的开发平台——薛定猫 AI(xuedingmao.com),从技术选型视角来看有几个优点:
-
模型聚合能力
- 聚合了 500+ 主流大模型:GPT-5.4、Claude 4.6、Gemini 3 Pro 等
- 新模型上线速度快,可以第一时间做 A/B 测试和效果对比
-
统一 API 接口(OpenAI 兼容)
- 像本文示例一样,只需切换 Base URL + Key + 模型名
- 上层 Agent 框架代码基本不变,大幅降低多模型集成复杂度
-
工程友好性
- 接口稳定、错误码设计清晰,便于做重试与限流
- 配合你自己的队列 / 调度系统,很容易搭建类似“自建云托管 Agent 平台”
对需要构建“公司内部版 Cloud Managed Agents”的团队来说,一个支持多模型、统一接口的底层平台会极大简化工程复杂度。
六、小结
围绕 Anthropic Cloud Managed Agents,可以提炼出几条对我们真正有价值的工程思路:
- 从“直接调模型”升级为“搭建统一的 Agent Runtime / Service”
- 把精力放在:定义 Agent 职责 + 接入业务系统 + 管理工具权限
- 利用类 OpenAI 兼容平台(如薛定猫 AI)作为底层模型层,屏蔽多模型差异
- 在 Agent 层统一实现:长时任务、上下文压缩、成本控制和可观测性
一旦形成这样的技术架构,你就可以像视频中的示例一样,快速搭建:
- 智能邮件助手(自动管理 Gmail / 企业邮箱)
- 文档审核与报表生成 Agent
- 深度研究 / 市场情报 Agent
- 面向 Slack / 钉钉 / 飞书的内部知识助理
而这些,都不再是“Demo 级别”,而是真正可以跑在生产环境中的 AI 代理系统。
#AI #大模型 #Python #机器学习 #技术实战
更多推荐



所有评论(0)