这次我们来看一个偏工程落地的方向:用 Claude Managed Agents 的思路,把一个 AI Agent 从“能跑的 demo”做成“能上生产的应用”。主题词有两个,一个是 Managed,另一个是生产级。Managed 解决的是托管与运行方式的问题,生产级解决的是稳定性、一致性、可观测性、可重试性这一类工程问题。很多人在本地把 Agent 跑通了,一放到业务环境里就发现各种问题:没有日志、没有超时、工具调用经常失败、批量任务一多就全挂。这篇文章就是来补齐这些工程短板的。
先给结论:如果你是后端开发、AI 应用研发,或者正在给公司做智能体落地,这篇文章值得直接收藏。我们会从环境准备开始,先搭建一个最小可运行的 Agent,再逐步加入工具调用、会话持久化、人工审批、批量并发和成本监控。整个过程不依赖本地 GPU 集群,Agent 的推理与工具编排托管在 Claude 服务端,开发侧只需要通过 API/SDK 控制。文章不写空泛概念,全部按可执行的步骤展开,你跟着走就能得到一套生产可用的智能体骨架。
下面是核心信息速览:Claude Managed Agents 本质上是一种托管式 Agent 运行方式,模型推理和 Agent 循环在云侧执行,本地不需要维护推理服务;开发侧通过 Anthropic SDK 或 HTTP API 发起请求,传递提示词、工具定义和上下文;服务端返回模型回复、工具调用指令和结构化结果。你可以把它理解成“把 Agent 的主循环托管出去,自己只负责业务逻辑和工具实现”。这种模式最大的好处是部署简单、不需要运维 GPU 节点、天然支持多任务并发和结果回传,尤其适合内容生成、工单处理、数据分析等生产场景。
1. Claude Managed Agents 核心能力速览
| 能力项 | 说明 |
|---|---|
| 项目类型 | AI Agent 编排与托管执行服务 |
| 运行环境 | Claude 云侧托管,本地无需 GPU 推理集群 |
| 硬件门槛 | 低,只需要能正常访问官方 API 的服务器或开发机 |
| 开发方式 | Anthropic SDK / REST API |
| 核心能力 | 多工具调用、任务编排、上下文管理、结构化输出 |
| 批量任务 | 支持并发提交、队列化处理、异步回传 |
| 扩展方式 | 自定义工具函数、外部 HTTP 接口、MCP 服务 |
| 观测能力 | 日志、token 消耗、延迟、错误率、结果比对 |
| 适合场景 | 内容生成、信息整理、流程自动化、客服辅助、运维助手 |
| 主要限制 | 依赖官方账号配额、需要外网连通、按 token 计费 |
补充一句:具体可用模型、配额上限、费率和接口路径要以你账号后台实际开通情况为准。不同账号的模型列表和速率限制不完全一样,生产环境建议先读一遍官方 API 文档,再确定选哪个模型。
2. 适用场景与使用边界
2.1 适合做什么
从实际落地看,Claude Managed Agents 更适合下面几类任务:
第一,内容生产的批量化。比如给几千篇文章生成摘要、提取关键词、生成 SEO 标题,这种任务对延迟不敏感,但对输出格式稳定性要求高,非常适合用托管 Agent 批量处理。
第二,消息和工单的自动分类。把用户反馈、客服工单、系统告警文本交给 Agent,让它按预定义分类体系打标、定级、提取责任方,再对接工单系统自动流转。
第三,知识库问答和信息整合。把企业内部文档做切片和检索,把检索结果交给 Agent 生成回答。这种场景模型的上下文能力、工具调用能力都能发挥出来。
第四,流程型任务编排。比如从邮件里提取订单信息、调用库存接口查余量、生成对账摘要、提交审批单。Agent 按规则调用多个工具,最后输出结构化结果。
2.2 不适合做什么
托管运行不意味没有边界。
第一,不适合超低延迟场景。如果业务要求首字返回在几百毫秒以内,且并发量极高,托管 Agent 的推理延迟和限流策略可能不满足要求。
第二,不适合完全离线环境。Agent 推理在 Claude 云侧执行,内网隔离且完全不允许外发的场景不能直接用,需要先做网络合规评估。
第三,不建议直接执行高危操作。比如自动转账、删除生产库、修改权限这类动作,不要让 Agent 不加限制地执行,必须加入人工审批和操作审计。
2.3 合规与安全边界
使用托管 Agent 时,进入请求的文本、工具返回数据都可能经过第三方服务端处理。因此,客户个人信息、未脱敏的财务数据、内部源码片段,都要先评估授权范围;涉及人脸、声音、医疗、金融等敏感数据,更要严格确认合规边界。生产落地前,建议和法务、安全团队一起列一个数据分类清单,明确哪些内容允许交给 Agent 处理,哪些必须本地化或脱敏后才能使用。
3. 环境准备与前置条件
这次实操不需要高配显卡,也不需要安装本地大模型,核心前置条件只有三个:一个可用的 Claude API 账号、一台能访问官方 API 的开发机、一个 Python 3.9 以上的环境。
3.1 获取 API Key
登录 Claude 官方账号后台,在 API Keys 页面创建一个密钥。创建后马上保存,因为密钥只显示一次,后面无法再完整查看。把密钥放到环境变量里,不要硬编码进代码仓库。
Windows 临时设置:
$env:ANTHROPIC_API_KEY="sk-ant-你的密钥"Linux / macOS 临时设置:
export ANTHROPIC_API_KEY="sk-ant-你的密钥"3.2 安装依赖
创建一个干净目录,初始化 Python 虚拟环境并安装依赖。
mkdir agent-project && cd agent-project python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate pip install anthropic python-dotenvanthropic是官方 SDK,python-dotenv用于读取.env配置文件。建议在项目根目录创建一个.env文件:
ANTHROPIC_API_KEY=sk-ant-你的密钥同时把.env加入.gitignore,防止密钥误提交。
3.3 网络要求
你的服务器或开发机必须能正常访问 Claude 官方 API 域名。如果你在公司内网,需要提前确认安全策略是否放行外网请求。如果请求一直超时,先检查网络连通性,再检查 API Key 是否有效,不要一上来就怀疑代码。
3.4 目录结构建议
生产项目不要把所有代码塞进一个文件。这里给一个推荐结构:
agent-project/ ├── .env ├── requirements.txt ├── src/ │ ├── __init__.py │ ├── main.py │ ├── agent.py │ └── tools/ │ ├── __init__.py │ └── weather.py ├── logs/ │ └── app.log ├── inputs/ └── outputs/日志、输入、输出分目录管理,后面做批量任务和问题排查会方便很多。
4. 搭建最小可运行 Agent
4.1 Agent 的核心循环
先把 Agent 的原理压缩成一句话:模型根据用户输入和工具定义决定是否调用工具;工具执行后把结果回传给模型;模型再根据结果生成最终回答或继续调用下一个工具。这个循环就是 Agent 主循环。
在 Claude Managed Agents 模式下,这个循环在托管侧完成,开发侧负责定义消息列表、工具列表和处理返回结果。下面写一个最小实现。
4.2 初始化客户端
import os from anthropic import Anthropic client = Anthropic( api_key=os.getenv("ANTHROPIC_API_KEY"), )4.3 定义工具
定义一个查询天气的工具,注意description要写清楚边界和参数含义,这直接影响模型判断是否调用工具。
TOOLS = [ { "name": "get_weather", "description": "查询指定城市的当前天气情况,输入城市名,返回温度和天气描述", "input_schema": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称,如 北京、上海、广州" } }, "required": ["city"] } } ]4.4 编写主循环
import json def run_agent(user_prompt: str): messages = [ {"role": "user", "content": user_prompt} ] while True: response = client.messages.create( model="claude-xxx", # 替换为你账号可用模型 max_tokens=2048, tools=TOOLS, messages=messages, ) content_blocks = response.content messages.append({ "role": "assistant", "content": content_blocks, }) if response.stop_reason == "tool_use": for block in content_blocks: if block.type == "tool_use": result = execute_tool(block.name, block.input) messages.append({ "role": "user", "content": [ { "type": "tool_result", "tool_use_id": block.id, "content": json.dumps(result, ensure_ascii=False) } ] }) else: return content_blocks4.5 工具执行函数
def execute_tool(name: str, input_data: dict): if name == "get_weather": city = input_data.get("city") # 这里替换为真实天气 API 调用,或先返回模拟数据 return {"city": city, "temperature": 26, "weather": "多云"} raise ValueError(f"未知工具: {name}")4.6 启动验证
if __name__ == "__main__": result = run_agent("北京现在天气怎么样?适合穿什么衣服?") for block in result: if block.type == "text": print(block.text)跑起来后你会在终端看到 Agent 先调用工具的完整过程,再输出最终回答。判断运行成功不只是看有没有文字输出,还要确认两点:一是日志里出现了tool_use调用,二是最终回答基于工具返回结果生成,而不是模型自己编的天气。
常见失败原因集中在四个地方:API Key 没有正确设置;模型名不可用;工具 schema 不合法;网络请求超时。排除这四类后,最小 Agent 基本都能跑通。
5. 工具调用:让 Agent 具备操作能力
5.1 工具描述的重要性
Agent 是否调用工具,很大程度取决于工具定义写得准不准确。工具名称要动词开头,描述要写清“什么场景调用”“需要哪些输入”“返回什么结构”。比如create_ticket(order_id, reason)比func1(param1, param2)更容易被模型正确调用。
5.2 接入真实业务接口
实际项目中,工具函数内部通常会请求外部系统,比如查询 CRM、调用物流接口、查询数据库。这里给一个通用模板:
import requests def call_business_api(url: str, params: dict) -> dict: response = requests.get(url, params=params, timeout=10) response.raise_for_status() return response.json() def execute_tool(name: str, input_data: dict): if name == "get_weather": data = call_business_api("https://api.example.com/weather", input_data) return data if name == "query_order": data = call_business_api("https://api.example.com/order", input_data) return data raise ValueError(f"未知工具: {name}")注意timeout一定要设置。生产环境任何外部接口都可能慢,不设超时会导致 Agent 循环挂死。
5.3 工具结果要精简
工具返回结果会重新进入模型上下文,越长消耗的 token 越多,也越容易干扰模型判断。因此在工具内部就做好字段裁剪,只返回模型真正需要的信息。
5.4 通过 MCP 扩展外部能力
如果团队内部已经封装了多个 MCP 服务,可以让 Agent 通过 MCP 协议接入这些服务。MCP 的思路是把外部工具标准化成统一协议,Agent 不需要关心工具背后的实现语言和部署方式,只需要按标准协议调用。生产落地时,你可以把公司内部已经存在的 API、数据库查询、审批系统逐步封装成 MCP 工具,降低接入成本。
5.5 工具调用的失败处理
工具调用可能失败,不能把异常直接抛给 Agent 循环。更稳的做法是在工具内部捕获异常,并返回一个包含错误信息的结构化结果,让模型根据错误信息决定重试还是换方案。
def safe_execute_tool(name: str, input_data: dict) -> dict: try: return {"success": True, "data": execute_tool(name, input_data)} except Exception as e: return {"success": False, "error": str(e)}6. 生产级改造:会话、状态与人工审批
6.1 会话状态管理
最小 Agent 是无状态的,每次对话只关注当前请求。但生产环境往往需要多轮对话。你需要把messages按会话维度持久化,比如存在 Redis 或 PostgreSQL 中。
推荐结构:
def load_session(session_id: str) -> list: # 从 Redis 或数据库读取该会话的历史 messages pass def save_session(session_id: str, messages: list) -> None: # 将最新的 messages 写回存储 pass每次请求先加载历史消息,追加新用户输入,调用模型后保存最新消息。这里要重点控制上下文长度,超过阈值时做历史裁剪或摘要压缩,避免无限制膨胀。
6.2 人工审批织入
生产级 Agent 不能所有操作都自动执行。对于敏感工具,比如对外发消息、转账、删除数据,可以在工具定义中加入审批标识。
APPROVAL_TOOLS = {"send_email", "delete_record", "transfer_amount"} if block.type == "tool_use": tool_name = block.name if tool_name in APPROVAL_TOOLS: return { "status": "pending_approval", "tool_use_id": block.id, "message": f"工具 {tool_name} 需要人工审批" } # 否则正常执行这种设计把 Agent 从“自动决策”变成“辅助决策”,既保留效率,又把风险控制在大脑手里。
6.3 幂等与重试
工具执行可能因为网络原因失败,也可能“执行成功但结果没回来”。为避免重复执行业务操作,建议给每个待执行工具生成一个request_id,在业务接口里做幂等校验。同样的request_id重复进来,直接返回上一次结果,不重复写库、不重复发消息。
6.4 安全与提示注入
生产环境里,用户输入可能被恶意构造。不要让外部用户的文本直接拼接进需要执行的关键参数。工具调用前做参数校验和枚举校验,比如部门 ID 必须在内网枚举列表内,金额不能超过阈值。对于公共场景,还要防止用户诱导 Agent 输出系统提示词或越权工具调用,工具白名单是底线。
6.5 日志与可观测
每条请求至少记录以下字段:
| 字段 | 示例 |
|---|---|
| request_id | req_20250101_001 |
| session_id | session_0001 |
| model | claude-xxx |
| prompt_tokens | 1200 |
| completion_tokens | 400 |
| latency_ms | 2300 |
| stop_reason | end_turn / tool_use |
| tool_used | get_weather |
| status | success / failure |
日志不建议只写到控制台,要落到文件或日志系统,方便后续排查问题和做成本统计。
7. 批量任务与并发调度
生产环境经常面对一个需求:几千条数据要统一处理。逐条同步调用会非常慢,也不利于控制成本。批量任务的核心是并发控制、失败重试、结果落盘三件事。
7.1 使用 AsyncAnthropic 做并发
官方 SDK 提供了异步客户端,配合信号量可以控制并发数量。
import asyncio from anthropic import AsyncAnthropic client = AsyncAnthropic() semaphore = asyncio.Semaphore(10) async def process_one(text: str) -> dict: async with semaphore: try: resp = await client.messages.create( model="claude-xxx", max_tokens=1024, messages=[{"role": "user", "content": text}], ) return {"input": text, "output": resp.content[0].text, "status": "ok"} except Exception as e: return {"input": text, "output": str(e), "status": "failed"} async def run_batch(items: list[str]) -> list[dict]: tasks = [process_one(item) for item in items] results = await asyncio.gather(*tasks, return_exceptions=False) return results if __name__ == "__main__": items = ["文本1", "文本2", "文本3"] results = asyncio.run(run_batch(items)) print(results)这里的Semaphore(10)表示最多同时 10 个请求。具体并发数要结合你账号的速率限制来定,不能无脑调大。
7.2 失败重试与退避
遇到限流或超时,直接抛错不可取。更稳的做法是加入指数退避重试。
import time def retry_call(func, max_retries=3): for attempt in range(max_retries): try: return func() except Exception: if attempt == max_retries - 1: raise time.sleep(min(2 ** attempt, 30))7.3 结果落盘
批量任务结果建议统一写成 JSON 或 CSV,并记录入参、输出、耗时和状态。否则一旦中途中断,很难定位哪些数据已经处理过。
import json import datetime def save_results(results: list[dict], path: str): data = { "created_at": datetime.datetime.now().isoformat(), "results": results } with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2)7.4 队列化设计
如果你的批量任务来自消息队列,建议按“消费者模式”设计:从队列拉取一条任务,处理一条,回调标记状态。这样即使进程崩溃,消息也不会丢,重启后可以继续消费。简单场景用 Redis 队列即可,不用一开始就上重型任务编排系统。
8. 成本、性能与稳定性观察
8.1 关注哪些指标
生产环境不能只看功能跑通,还要盯住三个指标。
第一个是延迟。一次 Agent 调用往往要经历多轮模型往返,延迟不是单次 API 的耗时,而是整个循环的总耗时。如果某类任务总是超过预期,要看是模型推理慢、工具接口慢,还是上下文太长导致计算变慢。
第二个是 token 消耗。输入 token、输出 token、工具参数和工具结果都计入成本。建议每次请求都记录 token 用量,按业务维度汇总,找出哪些环节消耗最高。
第三个是失败率。失败率高于预期时,优先看工具接口超时、限流、上下文超限这几类问题,不要盲目提高并发。
8.2 成本优化思路
成本不是一句空话,可以从四个方向压。
一是精简系统提示词和工具描述,减少每次请求的固定输入 token。二是限制max_tokens,防止模型一次性生成过长内容。三是做结果缓存,相同或相似的请求直接回缓存,不重复调用模型。四是模型路由,简单任务用成本更低的模型,复杂任务用更强模型,不要一个模型打天下。
8.3 性能观察方法
在开发环境可以手动打印每次请求的response.usage字段,观察输入和输出 token 分布。生产环境建议把这些指标接入监控系统,按小时或天聚合。稳定性是逐步调出来的,没有监控就没有优化依据。
9. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 返回 401 认证失败 | API Key 错误或未设置 | 检查环境变量是否读取成功 | 重新配置正确的 ANTHROPIC_API_KEY |
| 一直请求超时 | 网络不通或接口域名不可达 | 用 curl 测试 API 连通性 | 修复网络策略或换运行环境 |
| 提示模型名不可用 | model 参数不是该账号可用模型 | 去后台确认模型列表 | 换成实际可用模型 |
| 上下文长度超限 | 多轮 messages 过长 | 打印 messages 总 token 数 | 做历史裁剪或摘要压缩 |
| 返回 429 限流 | 并发太高或超过配额 | 查看响应头中的限流信息 | 降低并发、加退避重试 |
| 工具返回格式错误 | 工具函数异常透传 | 在工具内捕获并结构化返回 | 使用 safe_execute_tool |
| 结果不稳定 | 温度过高或提示词模糊 | 对比多次输出 | 降低 temperature、固化输出模板 |
| 批量任务卡住 | 同步请求阻塞或没设 timeout | 检查日志和进程状态 | 改异步、加超时、加失败重试 |
10. 最佳实践与合规建议
10.1 工程层面的最佳实践
第一,先小样本验证。正式跑全量批量任务前,先拿 10 条数据试跑,确认输出格式、字段完整度和成本符合预期,再放开全量。
第二,保留一套最小可运行配置。把第 4 节的最小 Agent 单独保存为一个模板,后续新项目从这里复制,而不是每次从零开始。
第三,提示词和工具定义纳入版本管理。提示词、工具描述和代码一起提交,哪个版本改了什么,线上出问题才能快速回滚。
第四,批量任务必须加日志和失败重试。不要裸跑 for 循环,否则中断一次就得全量重来。
第五,接口服务要限制访问范围。如果 Agent 以 HTTP API 形式开放给内部系统,务必加认证鉴权,不要裸挂在公网。
10.2 合规与授权方面的建议
如果你的 Agent 涉及个人信息、企业敏感数据、版权素材、人脸或声音信息,必须确认数据来源合法、处理方式获得授权,并且按照安全策略做脱敏和最小化。涉及对外发送消息、修改系统配置、删除数据等高风险操作,必须加入人工审批和事后审计。起步阶段控制权限范围,碰到合同、金融、医疗等内容,需要结合具体法律法规判断,必要时咨询专业人员。
10.3 从 demo 到生产的最小路径
如果你想快速验证这个方向,建议按下面的顺序走:跑通第 4 节最小 Agent,确认 API 连通性;加一个真实工具,验证工具调用闭环;加日志和错误处理,让程序可观测;加并发和重试,处理一批真实数据;最后根据业务需要决定是否加入会话持久化、人工审批和消息队列。每一步都有明确验证点,下一步的失败面就不会太大。
这次的内容已经把 Agent 从 demo 到生产的主干路径拆开了:托管执行、工具调用、会话管理、人工审批、批量并发、成本和稳定性观察。生产级智能体的难点从来不在“调用一次模型”,而在围绕模型建立一套可观测、可控制、可恢复的工程体系。建议先跑通最小 Agent,再逐步补工具和审批,把控制权和审计抓在自己手里。