1. 项目概述为什么需要为Claude API Agent构建三层架构最近在做一个基于Claude API的智能体项目从最初的快速原型到最终的生产部署踩了不少坑。我发现很多开发者包括我自己在初期都习惯把所有逻辑塞进一个脚本里——调用API、处理响应、管理对话状态、执行业务逻辑全混在一起。这在Demo阶段没问题但一旦要上线面对真实用户的并发请求、复杂的业务流、以及不可避免的API调用失败或网络抖动这种“一锅炖”的代码就会迅速变成一场灾难。“Three Layers for Production-Grade Claude API Agents in Python”这个标题精准地指向了问题的核心分层。它不是一个具体的工具或库而是一种架构设计思想旨在将Claude API驱动的智能体从脆弱的脚本转变为健壮、可维护、可扩展的生产级服务。这里的“生产级”意味着你的智能体需要具备高可用性、可观测性、错误恢复能力并且能够优雅地处理各种边界情况比如用户输入歧义、API速率限制、上下文长度超限等。简单来说这三层架构是为了解决单一模块的“上帝类”问题通过关注点分离让每一层只做好一件事。想象一下如果你家的水管、电路、网络线全都缠在一起任何一处出问题都得把整面墙砸开。分层架构就是给它们各自铺设独立的管道和线槽检修和维护起来就清晰多了。对于Claude API Agent我们通常面对的是自然语言输入、复杂的推理决策、以及对外部工具或系统的调用这三层分别对应了交互与适配、核心决策与流程、外部集成与执行。2. 核心架构设计三层模型详解与选型逻辑2.1 第一层接口与适配层这一层是智能体与外部世界的“翻译官”和“守门员”。它的核心职责是处理所有输入/输出I/O将外部请求如HTTP API调用、消息队列事件、命令行输入标准化为智能体能理解的内部格式并将智能体的内部响应再转换回外部世界期望的格式。为什么需要这一层直接原因是为了解耦。你的智能体核心逻辑不应该关心请求是来自FastAPI的HTTP POST、来自Celery的异步任务还是来自一个简单的input()函数。接口层负责屏蔽这些差异。更深层的原因是它提供了统一的地方来处理输入验证、身份认证、速率限制、请求日志记录和初步的错误处理如JSON解析错误。例如一个用户可能发送了空消息、超长文本或者包含了恶意脚本代码这些都应该在进入核心逻辑前被拦截和处理。典型组件与实现Web框架集成使用FastAPI或Flask定义清晰的RESTful端点。例如一个/chat端点接收用户消息和会话ID。输入标准化器将不同来源的请求体可能字段名不同转换为一个统一的AgentRequest数据类。这个类通常包含session_id、message、user_id等字段。输出格式化器将智能体返回的复杂内部对象可能包含文本、结构化数据、建议的后续操作序列化为标准的API响应格式如JSON。中间件用于添加跨切面关注点如请求日志、性能监控记录耗时、以及统一的错误响应包装。注意在这一层应避免进行任何与Claude API或业务逻辑直接相关的复杂处理。它的任务是“搬运”和“包装”而不是“思考”。2.2 第二层智能体核心层这是整个系统的“大脑”是真正体现智能体能力的地方。这一层接收来自接口层的标准化请求管理对话状态上下文与Claude API进行交互并根据API的响应驱动业务流程。核心职责与设计考量对话状态管理这是生产级智能体的关键。你不能让每次请求都从头开始对话。需要维护一个“会话”Session概念存储历史消息。通常使用一个ConversationManager类它负责上下文窗口管理Claude API有token限制。管理器需要智能地截断或总结历史对话确保最重要的上下文被保留同时不超出限制。一种常见策略是采用“滑动窗口”或对较早的历史进行摘要。消息格式编排按照Claude API要求的格式如System Prompt, Human Message, Assistant Message组织消息。Claude API客户端封装不要在每个业务逻辑里直接写requests.post。应该封装一个健壮的ClaudeClient类它内部处理重试逻辑网络超时、API限速429错误等情况下的自动重试通常使用指数退避策略。错误处理将API返回的各种错误如无效请求、额度不足、内容过滤转换为内部异常便于上层处理。配置管理集中管理API密钥、基础URL、默认模型参数如max_tokens,temperature。提示词工程与模板管理将System Prompt和常用的对话模板从代码中分离出来可以使用配置文件或模板引擎如Jinja2进行管理。这使得调整智能体的“性格”和指令无需修改代码。业务流程协调器对于复杂的智能体它可能不只是简单问答。它可能需要根据Claude的分析结果调用第三层的某个工具然后根据工具返回的结果再组织新一轮的对话。这个协调逻辑就在这一层。2.3 第三层工具与执行层智能体常常需要“动手”做事情比如查询数据库、调用另一个Web API、执行一个计算、读写文件。这一层就是智能体的“手”和“脚”。Claude API支持“工具使用”Function Calling智能体可以请求调用一个预定义的工具。设计要点工具抽象定义一个统一的Tool基类或协议所有具体工具如SearchWebTool、QueryDatabaseTool、CalculateTool都实现它。接口通常包含name工具名、description给Claude看的描述、parameters参数JSON Schema和execute执行方法。工具注册表一个中心化的ToolRegistry用于注册和管理所有可用工具。智能体核心层在初始化时从注册表获取工具列表并将其描述注入给Claude的System Prompt。安全与沙箱这是生产环境的生命线。工具执行可能涉及敏感操作或资源消耗。权限控制根据用户或会话决定其可以访问哪些工具。输入验证与净化对从Claude解析出的工具调用参数进行严格验证防止注入攻击。资源限制与超时为工具执行设置超时和资源限制如内存、CPU防止恶意或错误调用导致服务瘫痪。副作用管理对于写操作考虑引入确认机制或操作日志便于审计和回滚。执行器负责接收智能体核心层解析出的工具调用请求查找对应工具验证参数安全地执行并将结果格式化返回给核心层。三层之间的数据流可以概括为接口层接收原始请求 - 转换为AgentRequest- 传递给核心层- 核心层管理会话、调用Claude API - 若Claude返回工具调用请求 - 核心层将请求派发给工具层- 工具层执行并返回结果 - 核心层将结果补充到对话上下文再次调用Claude - 获得最终自然语言响应 - 核心层返回AgentResponse给接口层- 接口层格式化为最终输出。3. 分层实现详解与实操要点3.1 接口层实现构建稳健的API网关我们以FastAPI为例构建一个最小但完整的接口层。首先定义清晰的数据模型这是契约驱动的开发基础。from pydantic import BaseModel, Field, validator from typing import Optional, List, Dict, Any from enum import Enum class AgentRequest(BaseModel): 智能体请求标准格式 session_id: str Field(..., description唯一会话标识用于维护对话状态) message: str Field(..., min_length1, max_length5000, description用户输入的消息) user_id: Optional[str] Field(None, description用户标识用于权限和个性化) stream: bool Field(False, description是否启用流式响应) extra_context: Optional[Dict[str, Any]] Field(default_factorydict, description额外的上下文信息如用户资料、环境变量) validator(message) def message_not_empty(cls, v): if not v or v.isspace(): raise ValueError(消息内容不能为空) return v.strip() class AgentResponse(BaseModel): 智能体响应标准格式 success: bool reply: Optional[str] None data: Optional[Dict[str, Any]] None # 用于返回结构化数据如工具调用结果 session_id: str error_code: Optional[str] None error_message: Optional[str] None接下来创建FastAPI应用和路由。这里的关键是依赖注入Dependency Injection我们将核心层的智能体服务实例注入到路由处理函数中实现解耦。from fastapi import FastAPI, Depends, HTTPException, Request from fastapi.responses import StreamingResponse import logging import time app FastAPI(titleClaude智能体服务API) logger logging.getLogger(__name__) # 假设我们已经有一个核心层的AgentService类 from core.agent_service import AgentService def get_agent_service(): 依赖项获取或创建智能体服务实例通常与生命周期管理结合 # 这里可以是简单的单例或从容器中获取 return AgentService.get_instance() app.post(/v1/chat, response_modelAgentResponse) async def chat_endpoint( request: AgentRequest, agent_service: AgentService Depends(get_agent_service), raw_request: Request None # 注入原始请求对象用于获取IP等信息记录日志 ): 处理聊天请求的核心端点 start_time time.time() request_id freq_{int(start_time)}_{hash(raw_request.client.host) if raw_request else 0} logger.info(f[{request_id}] 收到请求session_id: {request.session_id}, 消息长度: {len(request.message)}) try: # 调用核心层服务 response await agent_service.process(request) elapsed time.time() - start_time logger.info(f[{request_id}] 请求处理成功耗时: {elapsed:.3f}s) return response except Exception as e: logger.exception(f[{request_id}] 请求处理失败: {e}) elapsed time.time() - start_time # 根据异常类型返回不同的HTTP状态码和错误信息 # 这里可以定义更精细的业务异常类 raise HTTPException(status_code500, detail{ error_code: INTERNAL_ERROR, error_message: 智能体服务处理异常, request_id: request_id }) app.post(/v1/chat/stream) async def chat_stream_endpoint( request: AgentRequest, agent_service: AgentService Depends(get_agent_service) ): 流式响应端点如果Claude API和你的核心层支持 async def event_generator(): # 假设agent_service.process_stream返回一个异步生成器 async for chunk in agent_service.process_stream(request): # 按照Server-Sent Events (SSE)格式或其他流式协议 yield 数据 yield fdata: {chunk}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)实操心得在接口层一定要做好详尽的请求日志。记录请求ID、会话ID、用户ID脱敏后、处理时间、关键参数。这不仅是调试的利器更是监控服务健康度和分析用户行为的基础。另外对于/v1/chat/stream这样的流式端点要特别注意连接中断的处理确保资源能被正确释放。3.2 核心层实现构建有状态的对话引擎核心层的AgentService是中枢。我们首先构建对话状态管理器。一个简单但有效的实现是使用内存字典适用于单机、小规模生产环境通常会使用Redis或数据库。# core/conversation_manager.py from typing import List, Dict, Any, Optional from dataclasses import dataclass, asdict from datetime import datetime import json import hashlib dataclass class Message: role: str # system, user, assistant, tool content: str name: Optional[str] None # 对于tool角色可以是工具名 timestamp: datetime None def __post_init__(self): if self.timestamp is None: self.timestamp datetime.utcnow() def to_api_format(self) - Dict[str, Any]: 转换为Claude API要求的消息格式 msg {role: self.role, content: self.content} if self.name: msg[name] self.name return msg class ConversationManager: def __init__(self, max_history_tokens: int 8000, system_prompt: str ): self.max_history_tokens max_history_tokens self.system_prompt system_prompt self._sessions: Dict[str, List[Message]] {} # session_id - messages def get_or_create_session(self, session_id: str) - List[Message]: 获取或创建一个会话的消息历史 if session_id not in self._sessions: # 初始化会话加入系统提示 self._sessions[session_id] [] if self.system_prompt: self._sessions[session_id].append(Message(rolesystem, contentself.system_prompt)) return self._sessions[session_id] def add_message(self, session_id: str, message: Message): 向会话中添加一条消息 messages self.get_or_create_session(session_id) messages.append(message) # 触发上下文窗口整理这里简化实际需要计算token数 self._trim_context_if_needed(session_id) def get_messages_for_api(self, session_id: str) - List[Dict[str, Any]]: 获取用于API调用的格式化消息列表 messages self.get_or_create_session(session_id) return [msg.to_api_format() for msg in messages] def _trim_context_if_needed(self, session_id: str): 简单的上下文截断策略保留最新的N条消息但永远保留系统提示 messages self._sessions[session_id] if len(messages) 1: # 只有系统提示 return # 这里是一个简化策略。实际生产环境需要 # 1. 使用tiktoken等库精确计算token数。 # 2. 采用更智能的策略如优先保留最近对话或对早期历史进行摘要。 MAX_MESSAGES 20 # 示例最大消息条数 if len(messages) MAX_MESSAGES: # 保留系统提示和最新的 N-1 条消息 system_msg messages[0] recent_msgs messages[-(MAX_MESSAGES-1):] self._sessions[session_id] [system_msg] recent_msgs def clear_session(self, session_id: str): 清空会话除系统提示外 if session_id in self._sessions and self._sessions[session_id]: system_msg self._sessions[session_id][0] if self._sessions[session_id][0].role system else None self._sessions[session_id] [system_msg] if system_msg else []接下来是封装Claude API客户端。我们使用httpx库支持异步并加入重试逻辑。# core/claude_client.py import httpx import asyncio from typing import Dict, Any, Optional, AsyncIterator import logging from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type logger logging.getLogger(__name__) class ClaudeAPIClient: def __init__(self, api_key: str, base_url: str https://api.anthropic.com, default_model: str claude-3-sonnet-20240229): self.api_key api_key self.base_url base_url.rstrip(/) self.default_model default_model self.client httpx.AsyncClient( headers{ x-api-key: self.api_key, anthropic-version: 2023-06-01, Content-Type: application/json }, timeouthttpx.Timeout(30.0, read60.0) # 设置合理的超时 ) def _is_retryable_error(self, exc: Exception) - bool: 判断异常是否可重试 if isinstance(exc, httpx.HTTPStatusError): # 429 速率限制 5xx 服务器错误通常可重试 return exc.response.status_code in {429, 500, 502, 503, 504} # 网络超时、连接错误等也可重试 return isinstance(exc, (httpx.TimeoutException, httpx.NetworkError)) retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type(lambda e: isinstance(e, Exception) and self._is_retryable_error(e)), reraiseTrue ) async def create_message(self, messages: List[Dict], model: Optional[str] None, max_tokens: int 1024, temperature: float 0.7, tools: Optional[List[Dict]] None, **kwargs) - Dict[str, Any]: 调用Claude Messages API url f{self.base_url}/v1/messages payload { model: model or self.default_model, messages: messages, max_tokens: max_tokens, temperature: temperature, **kwargs } if tools: payload[tools] tools logger.debug(f调用Claude API消息数: {len(messages)}) try: resp await self.client.post(url, jsonpayload) resp.raise_for_status() return resp.json() except httpx.HTTPStatusError as e: error_detail e.response.json().get(error, {}) if e.response.content else {} logger.error(fClaude API HTTP错误: {e.response.status_code}, 详情: {error_detail}) # 可以在这里将API错误转换为自定义的业务异常 raise except Exception as e: logger.error(f调用Claude API发生未知错误: {e}) raise async def close(self): await self.client.aclose()最后我们将它们组合成AgentService。这是协调一切的核心。# core/agent_service.py import asyncio import json from typing import Dict, Any, Optional from .conversation_manager import ConversationManager, Message from .claude_client import ClaudeAPIClient from tools.tool_registry import ToolRegistry, ToolExecutionResult class AgentService: def __init__(self, api_key: str, system_prompt: str, tool_registry: Optional[ToolRegistry] None): self.conv_manager ConversationManager(system_promptsystem_prompt) self.claude_client ClaudeAPIClient(api_keyapi_key) self.tool_registry tool_registry or ToolRegistry() async def process(self, request: AgentRequest) - AgentResponse: 处理单次请求的核心流程 # 1. 添加用户消息到会话历史 self.conv_manager.add_message( request.session_id, Message(roleuser, contentrequest.message) ) # 2. 准备API调用参数 messages self.conv_manager.get_messages_for_api(request.session_id) tools self.tool_registry.get_tools_descriptions() if self.tool_registry else None max_rounds 5 # 防止工具调用循环失控 current_round 0 while current_round max_rounds: current_round 1 # 3. 调用Claude API api_response await self.claude_client.create_message( messagesmessages, toolstools, max_tokens2048 # 为工具调用留足空间 ) # 4. 处理API响应 response_content api_response.get(content, []) stop_reason api_response.get(stop_reason) # 4.1 如果是普通文本回复 if stop_reason end_turn or not any(block.get(type) tool_use for block in response_content): assistant_text .join([block[text] for block in response_content if block[type] text]) # 将助手回复添加到历史 self.conv_manager.add_message( request.session_id, Message(roleassistant, contentassistant_text) ) # 返回最终结果 return AgentResponse( successTrue, replyassistant_text, session_idrequest.session_id ) # 4.2 如果Claude要求调用工具 elif any(block.get(type) tool_use for block in response_content): # 将Claude的回复包含工具调用请求先添加到历史 assistant_content_for_history [] for block in response_content: if block[type] text: assistant_content_for_history.append(block[text]) elif block[type] tool_use: # 工具调用块也以特定格式记录便于后续上下文理解 assistant_content_for_history.append(f[调用工具 {block[name]}输入: {block[input]}]) self.conv_manager.add_message( request.session_id, Message(roleassistant, content.join(assistant_content_for_history)) ) # 执行所有被请求的工具 tool_results_content [] for block in response_content: if block[type] tool_use: tool_name block[name] tool_input block[input] tool_id block[id] # 5. 调用工具层 tool_result: ToolExecutionResult await self.tool_registry.execute_tool( tool_name, tool_input, request.session_id, request.user_id ) # 6. 将工具执行结果格式化为Claude能理解的消息并添加到历史 tool_result_block { type: tool_result, tool_use_id: tool_id, content: tool_result.content } tool_results_content.append(tool_result_block) # 也将结果添加到对话管理器可选取决于你是否想让结果作为历史的一部分 self.conv_manager.add_message( request.session_id, Message(roletool, contentjson.dumps(tool_result.content), nametool_name) ) # 工具执行结果作为新一轮的“用户”消息实际上是工具返回的消息输入 # 注意在Claude Messages API中工具结果需要作为新一轮消息的一部分发送 # 我们需要构造一个新的消息列表包含历史工具结果然后重新循环 # 简化处理我们将工具结果作为一条虚拟的“用户”消息加入然后重新开始while循环 # 更准确的做法是在下一次API调用时messages应包含之前所有消息工具结果块。 # 由于我们的ConversationManager已经记录了tool角色的消息get_messages_for_api会包含它。 # 因此我们只需要更新messages变量继续循环即可。 messages self.conv_manager.get_messages_for_api(request.session_id) # 继续下一轮循环Claude将基于工具结果继续回复 # 如果循环超过最大轮次返回错误或最后一条信息 return AgentResponse( successFalse, reply对话过于复杂或工具调用循环可能已失控。请简化您的问题或重新开始会话。, session_idrequest.session_id, error_codeTOO_MANY_ROUNDS ) async def close(self): await self.claude_client.close()注意事项AgentService.process方法中的while循环是处理多轮工具调用的关键。必须设置最大轮次max_rounds以防止在工具调用逻辑出现死循环时耗尽资源。同时要仔细管理对话历史确保工具调用请求和结果被正确记录否则Claude会丢失上下文。3.3 工具层实现安全可控的“手和脚”工具层的核心是Tool抽象和ToolRegistry。我们先定义基础类。# tools/base_tool.py from abc import ABC, abstractmethod from typing import Dict, Any, Optional from pydantic import BaseModel, ValidationError import inspect class ToolInputSchema(BaseModel): 工具输入参数的Pydantic模型基类。每个具体工具应定义自己的模型。 pass class ToolExecutionResult: def __init__(self, success: bool, content: Any, error_message: Optional[str] None): self.success success self.content content # 可以是字符串或可序列化的字典/列表 self.error_message error_message def to_claude_format(self) - str: 将执行结果转换为Claude API要求的工具结果内容格式通常是字符串 if isinstance(self.content, str): return self.content try: return json.dumps(self.content, ensure_asciiFalse) except: return str(self.content) class Tool(ABC): 所有工具的抽象基类 name: str description: str input_schema: type[ToolInputSchema] def __init__(self): if not hasattr(self, name) or not self.name: raise ValueError(fTool {self.__class__.__name__} must define a name attribute.) if not hasattr(self, description) or not self.description: raise ValueError(fTool {self.__class__.__name__} must define a description attribute.) if not hasattr(self, input_schema): raise ValueError(fTool {self.__class__.__name__} must define an input_schema attribute (a Pydantic model).) def get_json_schema(self) - Dict[str, Any]: 生成供Claude API使用的工具描述JSON Schema # 利用Pydantic模型的schema()方法 schema_dict self.input_schema.schema() return { name: self.name, description: self.description, input_schema: { type: object, properties: schema_dict.get(properties, {}), required: schema_dict.get(required, []) } } abstractmethod async def execute(self, input_data: Dict[str, Any], session_id: str, user_id: Optional[str] None) - ToolExecutionResult: 执行工具的核心方法。必须由子类实现。 pass def _validate_input(self, input_data: Dict[str, Any]) - ToolInputSchema: 使用Pydantic模型验证输入参数 try: return self.input_schema(**input_data) except ValidationError as e: raise ValueError(f工具 {self.name} 输入参数验证失败: {e})然后我们实现一个具体的工具示例一个简单的计算器。# tools/calculator_tool.py from .base_tool import Tool, ToolInputSchema, ToolExecutionResult from pydantic import Field from typing import Literal import math class CalculatorInput(ToolInputSchema): operation: Literal[add, subtract, multiply, divide, power, sqrt] a: float b: Optional[float] Field(None, description第二个操作数对于sqrt操作不需要。) # 可以添加更多验证比如除法时b不能为0 class CalculatorTool(Tool): name calculator description 执行基本的数学运算。支持加(add)、减(subtract)、乘(multiply)、除(divide)、幂运算(power)、平方根(sqrt)。 input_schema CalculatorInput async def execute(self, input_data: Dict[str, Any], session_id: str, user_id: Optional[str] None) - ToolExecutionResult: # 1. 验证输入 validated_input self._validate_input(input_data) # 2. 执行计算 try: op validated_input.operation a validated_input.a b validated_input.b if op add: result a b elif op subtract: result a - b elif op multiply: result a * b elif op divide: if b 0: return ToolExecutionResult(False, None, 除数不能为零) result a / b elif op power: result math.pow(a, b) elif op sqrt: if a 0: return ToolExecutionResult(False, None, 被开方数不能为负) result math.sqrt(a) else: return ToolExecutionResult(False, None, f不支持的运算: {op}) return ToolExecutionResult(True, {operation: op, result: result, input: {a: a, b: b}}) except Exception as e: # 记录日志 # logger.error(fCalculatorTool执行失败: {e}, extra{session_id: session_id}) return ToolExecutionResult(False, None, f计算过程中发生错误: {str(e)})接下来是工具注册表它负责管理和安全地执行工具。# tools/tool_registry.py from typing import Dict, List, Optional import asyncio from .base_tool import Tool, ToolExecutionResult class ToolRegistry: def __init__(self): self._tools: Dict[str, Tool] {} def register_tool(self, tool: Tool): 注册一个工具实例 if tool.name in self._tools: raise ValueError(f工具名 {tool.name} 已被注册。) self._tools[tool.name] tool def register_tools(self, tools: List[Tool]): 批量注册工具 for tool in tools: self.register_tool(tool) def get_tool(self, name: str) - Optional[Tool]: 根据名称获取工具 return self._tools.get(name) def get_tools_descriptions(self) - List[Dict[str, Any]]: 获取所有工具的JSON Schema描述用于传给Claude API return [tool.get_json_schema() for tool in self._tools.values()] async def execute_tool(self, tool_name: str, input_data: Dict, session_id: str, user_id: Optional[str] None, timeout: float 10.0) - ToolExecutionResult: 查找并执行工具包含超时和安全控制 tool self.get_tool(tool_name) if not tool: return ToolExecutionResult(False, None, f未找到工具: {tool_name}) # 这里可以添加权限检查例如根据user_id判断是否有权使用该工具 # if not self._check_permission(user_id, tool_name): # return ToolExecutionResult(False, None, 权限不足无法使用此工具) try: # 为工具执行设置超时防止长时间阻塞 result await asyncio.wait_for( tool.execute(input_data, session_id, user_id), timeouttimeout ) return result except asyncio.TimeoutError: # logger.warning(f工具 {tool_name} 执行超时 (session: {session_id})) return ToolExecutionResult(False, None, f工具 {tool_name} 执行超时) except Exception as e: # logger.exception(f工具 {tool_name} 执行异常: {e}) return ToolExecutionResult(False, None, f工具执行内部错误: {str(e)}) # 可选的权限检查方法 # def _check_permission(self, user_id: Optional[str], tool_name: str) - bool: # # 实现你的权限逻辑 # return True最后在应用启动时初始化并注册工具。# main.py 或应用工厂函数中 from tools.calculator_tool import CalculatorTool from tools.weather_tool import WeatherTool # 假设有另一个工具 from tools.tool_registry import ToolRegistry def create_tool_registry() - ToolRegistry: registry ToolRegistry() registry.register_tools([ CalculatorTool(), WeatherTool(api_keyyour_weather_api_key), # 带配置的工具 # ... 注册其他工具 ]) return registry实操心得工具层的安全是重中之重。除了输入验证还要考虑资源隔离对于可能消耗大量CPU/内存的工具如文件处理、复杂计算考虑在单独的线程池或进程池中运行。沙箱环境对于执行任意代码的工具极度危险需慎用必须使用Docker或安全的沙箱技术进行隔离。审计日志记录所有工具调用的详细信息谁user_id/session_id、何时、调用什么工具、输入参数是什么、结果是什么。这对于调试和合规性至关重要。工具描述给Claude的工具description要清晰准确。模糊的描述会导致Claude错误调用或根本不调用。好的描述应说明工具的目的、输入参数的精确含义和格式。4. 部署、监控与进阶优化4.1 部署架构考量三层架构为部署提供了灵活性。常见的生产部署模式包括单体服务将三层打包在一个Web服务如FastAPI应用中。这是最简单的方式适合中小规模。你需要确保应用是无状态的或将会话状态存储在外部的Redis/DB中以便水平扩展。微服务拆分将各层拆分为独立服务。接口层作为API网关。核心层作为独立的“智能体引擎”服务专注于对话管理和API调用。工具层甚至可以进一步拆分将不同的工具组部署为独立的微服务通过gRPC或内部HTTP API与核心层通信。这提高了系统的解耦度和可扩展性但复杂度也大大增加。异步任务队列对于耗时较长的工具调用如调用一个慢速的外部API核心层不应同步等待。可以将工具调用请求放入一个任务队列如Celery Redis/RabbitMQ然后立即返回一个“正在处理”的响应给用户。再通过WebSocket或轮询告知用户最终结果。这能显著提升接口的响应速度和吞吐量。4.2 可观测性与监控没有监控的生产系统如同盲人骑马。必须为你的智能体添加以下监控维度应用性能监控APM使用如OpenTelemetry、Datadog、New Relic等工具追踪每个请求的完整链路包括在接口层、核心层、工具层的耗时。特别关注Claude API的调用延迟和工具执行时间。业务指标监控请求量/QPS总请求数、成功/失败数。会话指标活跃会话数、平均会话长度轮次。Claude API使用Token消耗输入/输出、模型调用分布、错误类型限速、内容过滤等。工具使用各工具调用频率、成功率、平均耗时。日志聚合使用结构化日志JSON格式并发送到ELK、Loki或云服务商的日志平台。确保每条日志都包含request_id和session_id便于追踪单个请求的全生命周期。告警对关键错误如Claude API持续失败、工具执行超时率高、延迟飙升、Token消耗异常等设置告警。4.3 常见问题排查与优化技巧Claude API返回“rate limit”错误现象日志中出现429状态码。排查检查你的调用频率是否超过套餐限制。Anthropic对不同模型和套餐有不同的RPM每分钟请求数和TPM每分钟Token数限制。解决在ClaudeAPIClient中实现更完善的退避重试机制上文已用tenacity实现。在应用层面实现请求队列或令牌桶算法平滑请求流量避免突发请求触发限流。考虑使用多个API密钥进行负载均衡如果允许且有必要。上下文长度超限Context Window Exhausted现象API返回错误提示输入过长。排查检查ConversationManager中保存的历史消息总token数。使用tiktoken库Anthropic官方推荐或API本身的计数功能进行精确计算。解决实现智能截断不要简单丢弃最老的消息。可以尝试优先保留最近的对话和系统提示对中间的历史进行摘要可以调用Claude的API来生成摘要但这本身消耗token。分片处理对于超长的单次用户输入如上传文档可以尝试将其分片分多次发送给Claude并让Claude进行增量理解这需要更复杂的状态管理。选择更大窗口的模型如果业务需要长上下文优先选用支持更大上下文窗口的模型版本。工具调用陷入死循环现象智能体反复调用同一个或一组工具无法给出最终答案。排查检查AgentService中的max_rounds限制是否生效。分析日志看工具返回的结果是否未能满足Claude的需求导致其反复尝试。解决优化工具描述和结果确保工具的描述清晰返回的结果格式明确、信息充足。模糊的结果会让Claude困惑。在System Prompt中增加约束明确告诉Claude“如果你已经调用过工具X并得到了Y结果请基于此直接回答用户不要再次调用”。实现会话级工具调用记忆记录本次会话中已调用过的工具及其主要输入输出在Claude再次请求调用相同工具时直接返回记忆中的结果或提醒它。响应速度慢现象用户端感知延迟高。排查使用APM工具定位瓶颈。常见瓶颈网络到Claude API的延迟、复杂工具的执行时间、数据库查询慢。解决异步化确保你的Web框架如FastAPI和所有I/O操作数据库、外部API调用都是异步的避免阻塞事件循环。缓存对频繁查询且变化不快的工具结果进行缓存如天气信息、某些计算结果的缓存。流式响应对于文本生成启用Claude API的流式响应并实现服务端的Server-Sent Events (SSE)让用户能边生成边看到结果提升体验。地理优化如果用户和你的服务器、Claude API服务器地理距离远考虑使用CDN或在不同区域部署边缘节点。System Prompt效果不佳现象智能体的行为不符合预期比如角色扮演不到位、频繁调用不该调用的工具。解决迭代优化System Prompt的编写是门艺术。需要不断测试和调整。使用A/B测试框架对比不同Prompt版本的效果。结构化Prompt将System Prompt分成几个清晰的部分角色定义、核心指令、输出格式要求、工具使用规范、禁忌事项。Few-shot示例在System Prompt中直接包含几个高质量的例子用户输入、助手理想输出的例子这是引导Claude行为最有效的方式之一。构建生产级的Claude API智能体是一个系统工程三层架构提供了一个清晰、稳固的起点。它迫使你思考每一部分的职责、边界和交互方式。从这基础的三层出发你可以根据业务复杂度的增长逐步引入更高级的特性如工作流引擎、长期记忆存储、复杂的工具编排等。记住好的架构不是一次性设计出来的而是在不断应对真实世界挑战的过程中迭代演化出来的。
Python三层架构构建生产级Claude API智能体：从设计到部署
1. 项目概述为什么需要为Claude API Agent构建三层架构最近在做一个基于Claude API的智能体项目从最初的快速原型到最终的生产部署踩了不少坑。我发现很多开发者包括我自己在初期都习惯把所有逻辑塞进一个脚本里——调用API、处理响应、管理对话状态、执行业务逻辑全混在一起。这在Demo阶段没问题但一旦要上线面对真实用户的并发请求、复杂的业务流、以及不可避免的API调用失败或网络抖动这种“一锅炖”的代码就会迅速变成一场灾难。“Three Layers for Production-Grade Claude API Agents in Python”这个标题精准地指向了问题的核心分层。它不是一个具体的工具或库而是一种架构设计思想旨在将Claude API驱动的智能体从脆弱的脚本转变为健壮、可维护、可扩展的生产级服务。这里的“生产级”意味着你的智能体需要具备高可用性、可观测性、错误恢复能力并且能够优雅地处理各种边界情况比如用户输入歧义、API速率限制、上下文长度超限等。简单来说这三层架构是为了解决单一模块的“上帝类”问题通过关注点分离让每一层只做好一件事。想象一下如果你家的水管、电路、网络线全都缠在一起任何一处出问题都得把整面墙砸开。分层架构就是给它们各自铺设独立的管道和线槽检修和维护起来就清晰多了。对于Claude API Agent我们通常面对的是自然语言输入、复杂的推理决策、以及对外部工具或系统的调用这三层分别对应了交互与适配、核心决策与流程、外部集成与执行。2. 核心架构设计三层模型详解与选型逻辑2.1 第一层接口与适配层这一层是智能体与外部世界的“翻译官”和“守门员”。它的核心职责是处理所有输入/输出I/O将外部请求如HTTP API调用、消息队列事件、命令行输入标准化为智能体能理解的内部格式并将智能体的内部响应再转换回外部世界期望的格式。为什么需要这一层直接原因是为了解耦。你的智能体核心逻辑不应该关心请求是来自FastAPI的HTTP POST、来自Celery的异步任务还是来自一个简单的input()函数。接口层负责屏蔽这些差异。更深层的原因是它提供了统一的地方来处理输入验证、身份认证、速率限制、请求日志记录和初步的错误处理如JSON解析错误。例如一个用户可能发送了空消息、超长文本或者包含了恶意脚本代码这些都应该在进入核心逻辑前被拦截和处理。典型组件与实现Web框架集成使用FastAPI或Flask定义清晰的RESTful端点。例如一个/chat端点接收用户消息和会话ID。输入标准化器将不同来源的请求体可能字段名不同转换为一个统一的AgentRequest数据类。这个类通常包含session_id、message、user_id等字段。输出格式化器将智能体返回的复杂内部对象可能包含文本、结构化数据、建议的后续操作序列化为标准的API响应格式如JSON。中间件用于添加跨切面关注点如请求日志、性能监控记录耗时、以及统一的错误响应包装。注意在这一层应避免进行任何与Claude API或业务逻辑直接相关的复杂处理。它的任务是“搬运”和“包装”而不是“思考”。2.2 第二层智能体核心层这是整个系统的“大脑”是真正体现智能体能力的地方。这一层接收来自接口层的标准化请求管理对话状态上下文与Claude API进行交互并根据API的响应驱动业务流程。核心职责与设计考量对话状态管理这是生产级智能体的关键。你不能让每次请求都从头开始对话。需要维护一个“会话”Session概念存储历史消息。通常使用一个ConversationManager类它负责上下文窗口管理Claude API有token限制。管理器需要智能地截断或总结历史对话确保最重要的上下文被保留同时不超出限制。一种常见策略是采用“滑动窗口”或对较早的历史进行摘要。消息格式编排按照Claude API要求的格式如System Prompt, Human Message, Assistant Message组织消息。Claude API客户端封装不要在每个业务逻辑里直接写requests.post。应该封装一个健壮的ClaudeClient类它内部处理重试逻辑网络超时、API限速429错误等情况下的自动重试通常使用指数退避策略。错误处理将API返回的各种错误如无效请求、额度不足、内容过滤转换为内部异常便于上层处理。配置管理集中管理API密钥、基础URL、默认模型参数如max_tokens,temperature。提示词工程与模板管理将System Prompt和常用的对话模板从代码中分离出来可以使用配置文件或模板引擎如Jinja2进行管理。这使得调整智能体的“性格”和指令无需修改代码。业务流程协调器对于复杂的智能体它可能不只是简单问答。它可能需要根据Claude的分析结果调用第三层的某个工具然后根据工具返回的结果再组织新一轮的对话。这个协调逻辑就在这一层。2.3 第三层工具与执行层智能体常常需要“动手”做事情比如查询数据库、调用另一个Web API、执行一个计算、读写文件。这一层就是智能体的“手”和“脚”。Claude API支持“工具使用”Function Calling智能体可以请求调用一个预定义的工具。设计要点工具抽象定义一个统一的Tool基类或协议所有具体工具如SearchWebTool、QueryDatabaseTool、CalculateTool都实现它。接口通常包含name工具名、description给Claude看的描述、parameters参数JSON Schema和execute执行方法。工具注册表一个中心化的ToolRegistry用于注册和管理所有可用工具。智能体核心层在初始化时从注册表获取工具列表并将其描述注入给Claude的System Prompt。安全与沙箱这是生产环境的生命线。工具执行可能涉及敏感操作或资源消耗。权限控制根据用户或会话决定其可以访问哪些工具。输入验证与净化对从Claude解析出的工具调用参数进行严格验证防止注入攻击。资源限制与超时为工具执行设置超时和资源限制如内存、CPU防止恶意或错误调用导致服务瘫痪。副作用管理对于写操作考虑引入确认机制或操作日志便于审计和回滚。执行器负责接收智能体核心层解析出的工具调用请求查找对应工具验证参数安全地执行并将结果格式化返回给核心层。三层之间的数据流可以概括为接口层接收原始请求 - 转换为AgentRequest- 传递给核心层- 核心层管理会话、调用Claude API - 若Claude返回工具调用请求 - 核心层将请求派发给工具层- 工具层执行并返回结果 - 核心层将结果补充到对话上下文再次调用Claude - 获得最终自然语言响应 - 核心层返回AgentResponse给接口层- 接口层格式化为最终输出。3. 分层实现详解与实操要点3.1 接口层实现构建稳健的API网关我们以FastAPI为例构建一个最小但完整的接口层。首先定义清晰的数据模型这是契约驱动的开发基础。from pydantic import BaseModel, Field, validator from typing import Optional, List, Dict, Any from enum import Enum class AgentRequest(BaseModel): 智能体请求标准格式 session_id: str Field(..., description唯一会话标识用于维护对话状态) message: str Field(..., min_length1, max_length5000, description用户输入的消息) user_id: Optional[str] Field(None, description用户标识用于权限和个性化) stream: bool Field(False, description是否启用流式响应) extra_context: Optional[Dict[str, Any]] Field(default_factorydict, description额外的上下文信息如用户资料、环境变量) validator(message) def message_not_empty(cls, v): if not v or v.isspace(): raise ValueError(消息内容不能为空) return v.strip() class AgentResponse(BaseModel): 智能体响应标准格式 success: bool reply: Optional[str] None data: Optional[Dict[str, Any]] None # 用于返回结构化数据如工具调用结果 session_id: str error_code: Optional[str] None error_message: Optional[str] None接下来创建FastAPI应用和路由。这里的关键是依赖注入Dependency Injection我们将核心层的智能体服务实例注入到路由处理函数中实现解耦。from fastapi import FastAPI, Depends, HTTPException, Request from fastapi.responses import StreamingResponse import logging import time app FastAPI(titleClaude智能体服务API) logger logging.getLogger(__name__) # 假设我们已经有一个核心层的AgentService类 from core.agent_service import AgentService def get_agent_service(): 依赖项获取或创建智能体服务实例通常与生命周期管理结合 # 这里可以是简单的单例或从容器中获取 return AgentService.get_instance() app.post(/v1/chat, response_modelAgentResponse) async def chat_endpoint( request: AgentRequest, agent_service: AgentService Depends(get_agent_service), raw_request: Request None # 注入原始请求对象用于获取IP等信息记录日志 ): 处理聊天请求的核心端点 start_time time.time() request_id freq_{int(start_time)}_{hash(raw_request.client.host) if raw_request else 0} logger.info(f[{request_id}] 收到请求session_id: {request.session_id}, 消息长度: {len(request.message)}) try: # 调用核心层服务 response await agent_service.process(request) elapsed time.time() - start_time logger.info(f[{request_id}] 请求处理成功耗时: {elapsed:.3f}s) return response except Exception as e: logger.exception(f[{request_id}] 请求处理失败: {e}) elapsed time.time() - start_time # 根据异常类型返回不同的HTTP状态码和错误信息 # 这里可以定义更精细的业务异常类 raise HTTPException(status_code500, detail{ error_code: INTERNAL_ERROR, error_message: 智能体服务处理异常, request_id: request_id }) app.post(/v1/chat/stream) async def chat_stream_endpoint( request: AgentRequest, agent_service: AgentService Depends(get_agent_service) ): 流式响应端点如果Claude API和你的核心层支持 async def event_generator(): # 假设agent_service.process_stream返回一个异步生成器 async for chunk in agent_service.process_stream(request): # 按照Server-Sent Events (SSE)格式或其他流式协议 yield 数据 yield fdata: {chunk}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)实操心得在接口层一定要做好详尽的请求日志。记录请求ID、会话ID、用户ID脱敏后、处理时间、关键参数。这不仅是调试的利器更是监控服务健康度和分析用户行为的基础。另外对于/v1/chat/stream这样的流式端点要特别注意连接中断的处理确保资源能被正确释放。3.2 核心层实现构建有状态的对话引擎核心层的AgentService是中枢。我们首先构建对话状态管理器。一个简单但有效的实现是使用内存字典适用于单机、小规模生产环境通常会使用Redis或数据库。# core/conversation_manager.py from typing import List, Dict, Any, Optional from dataclasses import dataclass, asdict from datetime import datetime import json import hashlib dataclass class Message: role: str # system, user, assistant, tool content: str name: Optional[str] None # 对于tool角色可以是工具名 timestamp: datetime None def __post_init__(self): if self.timestamp is None: self.timestamp datetime.utcnow() def to_api_format(self) - Dict[str, Any]: 转换为Claude API要求的消息格式 msg {role: self.role, content: self.content} if self.name: msg[name] self.name return msg class ConversationManager: def __init__(self, max_history_tokens: int 8000, system_prompt: str ): self.max_history_tokens max_history_tokens self.system_prompt system_prompt self._sessions: Dict[str, List[Message]] {} # session_id - messages def get_or_create_session(self, session_id: str) - List[Message]: 获取或创建一个会话的消息历史 if session_id not in self._sessions: # 初始化会话加入系统提示 self._sessions[session_id] [] if self.system_prompt: self._sessions[session_id].append(Message(rolesystem, contentself.system_prompt)) return self._sessions[session_id] def add_message(self, session_id: str, message: Message): 向会话中添加一条消息 messages self.get_or_create_session(session_id) messages.append(message) # 触发上下文窗口整理这里简化实际需要计算token数 self._trim_context_if_needed(session_id) def get_messages_for_api(self, session_id: str) - List[Dict[str, Any]]: 获取用于API调用的格式化消息列表 messages self.get_or_create_session(session_id) return [msg.to_api_format() for msg in messages] def _trim_context_if_needed(self, session_id: str): 简单的上下文截断策略保留最新的N条消息但永远保留系统提示 messages self._sessions[session_id] if len(messages) 1: # 只有系统提示 return # 这里是一个简化策略。实际生产环境需要 # 1. 使用tiktoken等库精确计算token数。 # 2. 采用更智能的策略如优先保留最近对话或对早期历史进行摘要。 MAX_MESSAGES 20 # 示例最大消息条数 if len(messages) MAX_MESSAGES: # 保留系统提示和最新的 N-1 条消息 system_msg messages[0] recent_msgs messages[-(MAX_MESSAGES-1):] self._sessions[session_id] [system_msg] recent_msgs def clear_session(self, session_id: str): 清空会话除系统提示外 if session_id in self._sessions and self._sessions[session_id]: system_msg self._sessions[session_id][0] if self._sessions[session_id][0].role system else None self._sessions[session_id] [system_msg] if system_msg else []接下来是封装Claude API客户端。我们使用httpx库支持异步并加入重试逻辑。# core/claude_client.py import httpx import asyncio from typing import Dict, Any, Optional, AsyncIterator import logging from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type logger logging.getLogger(__name__) class ClaudeAPIClient: def __init__(self, api_key: str, base_url: str https://api.anthropic.com, default_model: str claude-3-sonnet-20240229): self.api_key api_key self.base_url base_url.rstrip(/) self.default_model default_model self.client httpx.AsyncClient( headers{ x-api-key: self.api_key, anthropic-version: 2023-06-01, Content-Type: application/json }, timeouthttpx.Timeout(30.0, read60.0) # 设置合理的超时 ) def _is_retryable_error(self, exc: Exception) - bool: 判断异常是否可重试 if isinstance(exc, httpx.HTTPStatusError): # 429 速率限制 5xx 服务器错误通常可重试 return exc.response.status_code in {429, 500, 502, 503, 504} # 网络超时、连接错误等也可重试 return isinstance(exc, (httpx.TimeoutException, httpx.NetworkError)) retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type(lambda e: isinstance(e, Exception) and self._is_retryable_error(e)), reraiseTrue ) async def create_message(self, messages: List[Dict], model: Optional[str] None, max_tokens: int 1024, temperature: float 0.7, tools: Optional[List[Dict]] None, **kwargs) - Dict[str, Any]: 调用Claude Messages API url f{self.base_url}/v1/messages payload { model: model or self.default_model, messages: messages, max_tokens: max_tokens, temperature: temperature, **kwargs } if tools: payload[tools] tools logger.debug(f调用Claude API消息数: {len(messages)}) try: resp await self.client.post(url, jsonpayload) resp.raise_for_status() return resp.json() except httpx.HTTPStatusError as e: error_detail e.response.json().get(error, {}) if e.response.content else {} logger.error(fClaude API HTTP错误: {e.response.status_code}, 详情: {error_detail}) # 可以在这里将API错误转换为自定义的业务异常 raise except Exception as e: logger.error(f调用Claude API发生未知错误: {e}) raise async def close(self): await self.client.aclose()最后我们将它们组合成AgentService。这是协调一切的核心。# core/agent_service.py import asyncio import json from typing import Dict, Any, Optional from .conversation_manager import ConversationManager, Message from .claude_client import ClaudeAPIClient from tools.tool_registry import ToolRegistry, ToolExecutionResult class AgentService: def __init__(self, api_key: str, system_prompt: str, tool_registry: Optional[ToolRegistry] None): self.conv_manager ConversationManager(system_promptsystem_prompt) self.claude_client ClaudeAPIClient(api_keyapi_key) self.tool_registry tool_registry or ToolRegistry() async def process(self, request: AgentRequest) - AgentResponse: 处理单次请求的核心流程 # 1. 添加用户消息到会话历史 self.conv_manager.add_message( request.session_id, Message(roleuser, contentrequest.message) ) # 2. 准备API调用参数 messages self.conv_manager.get_messages_for_api(request.session_id) tools self.tool_registry.get_tools_descriptions() if self.tool_registry else None max_rounds 5 # 防止工具调用循环失控 current_round 0 while current_round max_rounds: current_round 1 # 3. 调用Claude API api_response await self.claude_client.create_message( messagesmessages, toolstools, max_tokens2048 # 为工具调用留足空间 ) # 4. 处理API响应 response_content api_response.get(content, []) stop_reason api_response.get(stop_reason) # 4.1 如果是普通文本回复 if stop_reason end_turn or not any(block.get(type) tool_use for block in response_content): assistant_text .join([block[text] for block in response_content if block[type] text]) # 将助手回复添加到历史 self.conv_manager.add_message( request.session_id, Message(roleassistant, contentassistant_text) ) # 返回最终结果 return AgentResponse( successTrue, replyassistant_text, session_idrequest.session_id ) # 4.2 如果Claude要求调用工具 elif any(block.get(type) tool_use for block in response_content): # 将Claude的回复包含工具调用请求先添加到历史 assistant_content_for_history [] for block in response_content: if block[type] text: assistant_content_for_history.append(block[text]) elif block[type] tool_use: # 工具调用块也以特定格式记录便于后续上下文理解 assistant_content_for_history.append(f[调用工具 {block[name]}输入: {block[input]}]) self.conv_manager.add_message( request.session_id, Message(roleassistant, content.join(assistant_content_for_history)) ) # 执行所有被请求的工具 tool_results_content [] for block in response_content: if block[type] tool_use: tool_name block[name] tool_input block[input] tool_id block[id] # 5. 调用工具层 tool_result: ToolExecutionResult await self.tool_registry.execute_tool( tool_name, tool_input, request.session_id, request.user_id ) # 6. 将工具执行结果格式化为Claude能理解的消息并添加到历史 tool_result_block { type: tool_result, tool_use_id: tool_id, content: tool_result.content } tool_results_content.append(tool_result_block) # 也将结果添加到对话管理器可选取决于你是否想让结果作为历史的一部分 self.conv_manager.add_message( request.session_id, Message(roletool, contentjson.dumps(tool_result.content), nametool_name) ) # 工具执行结果作为新一轮的“用户”消息实际上是工具返回的消息输入 # 注意在Claude Messages API中工具结果需要作为新一轮消息的一部分发送 # 我们需要构造一个新的消息列表包含历史工具结果然后重新循环 # 简化处理我们将工具结果作为一条虚拟的“用户”消息加入然后重新开始while循环 # 更准确的做法是在下一次API调用时messages应包含之前所有消息工具结果块。 # 由于我们的ConversationManager已经记录了tool角色的消息get_messages_for_api会包含它。 # 因此我们只需要更新messages变量继续循环即可。 messages self.conv_manager.get_messages_for_api(request.session_id) # 继续下一轮循环Claude将基于工具结果继续回复 # 如果循环超过最大轮次返回错误或最后一条信息 return AgentResponse( successFalse, reply对话过于复杂或工具调用循环可能已失控。请简化您的问题或重新开始会话。, session_idrequest.session_id, error_codeTOO_MANY_ROUNDS ) async def close(self): await self.claude_client.close()注意事项AgentService.process方法中的while循环是处理多轮工具调用的关键。必须设置最大轮次max_rounds以防止在工具调用逻辑出现死循环时耗尽资源。同时要仔细管理对话历史确保工具调用请求和结果被正确记录否则Claude会丢失上下文。3.3 工具层实现安全可控的“手和脚”工具层的核心是Tool抽象和ToolRegistry。我们先定义基础类。# tools/base_tool.py from abc import ABC, abstractmethod from typing import Dict, Any, Optional from pydantic import BaseModel, ValidationError import inspect class ToolInputSchema(BaseModel): 工具输入参数的Pydantic模型基类。每个具体工具应定义自己的模型。 pass class ToolExecutionResult: def __init__(self, success: bool, content: Any, error_message: Optional[str] None): self.success success self.content content # 可以是字符串或可序列化的字典/列表 self.error_message error_message def to_claude_format(self) - str: 将执行结果转换为Claude API要求的工具结果内容格式通常是字符串 if isinstance(self.content, str): return self.content try: return json.dumps(self.content, ensure_asciiFalse) except: return str(self.content) class Tool(ABC): 所有工具的抽象基类 name: str description: str input_schema: type[ToolInputSchema] def __init__(self): if not hasattr(self, name) or not self.name: raise ValueError(fTool {self.__class__.__name__} must define a name attribute.) if not hasattr(self, description) or not self.description: raise ValueError(fTool {self.__class__.__name__} must define a description attribute.) if not hasattr(self, input_schema): raise ValueError(fTool {self.__class__.__name__} must define an input_schema attribute (a Pydantic model).) def get_json_schema(self) - Dict[str, Any]: 生成供Claude API使用的工具描述JSON Schema # 利用Pydantic模型的schema()方法 schema_dict self.input_schema.schema() return { name: self.name, description: self.description, input_schema: { type: object, properties: schema_dict.get(properties, {}), required: schema_dict.get(required, []) } } abstractmethod async def execute(self, input_data: Dict[str, Any], session_id: str, user_id: Optional[str] None) - ToolExecutionResult: 执行工具的核心方法。必须由子类实现。 pass def _validate_input(self, input_data: Dict[str, Any]) - ToolInputSchema: 使用Pydantic模型验证输入参数 try: return self.input_schema(**input_data) except ValidationError as e: raise ValueError(f工具 {self.name} 输入参数验证失败: {e})然后我们实现一个具体的工具示例一个简单的计算器。# tools/calculator_tool.py from .base_tool import Tool, ToolInputSchema, ToolExecutionResult from pydantic import Field from typing import Literal import math class CalculatorInput(ToolInputSchema): operation: Literal[add, subtract, multiply, divide, power, sqrt] a: float b: Optional[float] Field(None, description第二个操作数对于sqrt操作不需要。) # 可以添加更多验证比如除法时b不能为0 class CalculatorTool(Tool): name calculator description 执行基本的数学运算。支持加(add)、减(subtract)、乘(multiply)、除(divide)、幂运算(power)、平方根(sqrt)。 input_schema CalculatorInput async def execute(self, input_data: Dict[str, Any], session_id: str, user_id: Optional[str] None) - ToolExecutionResult: # 1. 验证输入 validated_input self._validate_input(input_data) # 2. 执行计算 try: op validated_input.operation a validated_input.a b validated_input.b if op add: result a b elif op subtract: result a - b elif op multiply: result a * b elif op divide: if b 0: return ToolExecutionResult(False, None, 除数不能为零) result a / b elif op power: result math.pow(a, b) elif op sqrt: if a 0: return ToolExecutionResult(False, None, 被开方数不能为负) result math.sqrt(a) else: return ToolExecutionResult(False, None, f不支持的运算: {op}) return ToolExecutionResult(True, {operation: op, result: result, input: {a: a, b: b}}) except Exception as e: # 记录日志 # logger.error(fCalculatorTool执行失败: {e}, extra{session_id: session_id}) return ToolExecutionResult(False, None, f计算过程中发生错误: {str(e)})接下来是工具注册表它负责管理和安全地执行工具。# tools/tool_registry.py from typing import Dict, List, Optional import asyncio from .base_tool import Tool, ToolExecutionResult class ToolRegistry: def __init__(self): self._tools: Dict[str, Tool] {} def register_tool(self, tool: Tool): 注册一个工具实例 if tool.name in self._tools: raise ValueError(f工具名 {tool.name} 已被注册。) self._tools[tool.name] tool def register_tools(self, tools: List[Tool]): 批量注册工具 for tool in tools: self.register_tool(tool) def get_tool(self, name: str) - Optional[Tool]: 根据名称获取工具 return self._tools.get(name) def get_tools_descriptions(self) - List[Dict[str, Any]]: 获取所有工具的JSON Schema描述用于传给Claude API return [tool.get_json_schema() for tool in self._tools.values()] async def execute_tool(self, tool_name: str, input_data: Dict, session_id: str, user_id: Optional[str] None, timeout: float 10.0) - ToolExecutionResult: 查找并执行工具包含超时和安全控制 tool self.get_tool(tool_name) if not tool: return ToolExecutionResult(False, None, f未找到工具: {tool_name}) # 这里可以添加权限检查例如根据user_id判断是否有权使用该工具 # if not self._check_permission(user_id, tool_name): # return ToolExecutionResult(False, None, 权限不足无法使用此工具) try: # 为工具执行设置超时防止长时间阻塞 result await asyncio.wait_for( tool.execute(input_data, session_id, user_id), timeouttimeout ) return result except asyncio.TimeoutError: # logger.warning(f工具 {tool_name} 执行超时 (session: {session_id})) return ToolExecutionResult(False, None, f工具 {tool_name} 执行超时) except Exception as e: # logger.exception(f工具 {tool_name} 执行异常: {e}) return ToolExecutionResult(False, None, f工具执行内部错误: {str(e)}) # 可选的权限检查方法 # def _check_permission(self, user_id: Optional[str], tool_name: str) - bool: # # 实现你的权限逻辑 # return True最后在应用启动时初始化并注册工具。# main.py 或应用工厂函数中 from tools.calculator_tool import CalculatorTool from tools.weather_tool import WeatherTool # 假设有另一个工具 from tools.tool_registry import ToolRegistry def create_tool_registry() - ToolRegistry: registry ToolRegistry() registry.register_tools([ CalculatorTool(), WeatherTool(api_keyyour_weather_api_key), # 带配置的工具 # ... 注册其他工具 ]) return registry实操心得工具层的安全是重中之重。除了输入验证还要考虑资源隔离对于可能消耗大量CPU/内存的工具如文件处理、复杂计算考虑在单独的线程池或进程池中运行。沙箱环境对于执行任意代码的工具极度危险需慎用必须使用Docker或安全的沙箱技术进行隔离。审计日志记录所有工具调用的详细信息谁user_id/session_id、何时、调用什么工具、输入参数是什么、结果是什么。这对于调试和合规性至关重要。工具描述给Claude的工具description要清晰准确。模糊的描述会导致Claude错误调用或根本不调用。好的描述应说明工具的目的、输入参数的精确含义和格式。4. 部署、监控与进阶优化4.1 部署架构考量三层架构为部署提供了灵活性。常见的生产部署模式包括单体服务将三层打包在一个Web服务如FastAPI应用中。这是最简单的方式适合中小规模。你需要确保应用是无状态的或将会话状态存储在外部的Redis/DB中以便水平扩展。微服务拆分将各层拆分为独立服务。接口层作为API网关。核心层作为独立的“智能体引擎”服务专注于对话管理和API调用。工具层甚至可以进一步拆分将不同的工具组部署为独立的微服务通过gRPC或内部HTTP API与核心层通信。这提高了系统的解耦度和可扩展性但复杂度也大大增加。异步任务队列对于耗时较长的工具调用如调用一个慢速的外部API核心层不应同步等待。可以将工具调用请求放入一个任务队列如Celery Redis/RabbitMQ然后立即返回一个“正在处理”的响应给用户。再通过WebSocket或轮询告知用户最终结果。这能显著提升接口的响应速度和吞吐量。4.2 可观测性与监控没有监控的生产系统如同盲人骑马。必须为你的智能体添加以下监控维度应用性能监控APM使用如OpenTelemetry、Datadog、New Relic等工具追踪每个请求的完整链路包括在接口层、核心层、工具层的耗时。特别关注Claude API的调用延迟和工具执行时间。业务指标监控请求量/QPS总请求数、成功/失败数。会话指标活跃会话数、平均会话长度轮次。Claude API使用Token消耗输入/输出、模型调用分布、错误类型限速、内容过滤等。工具使用各工具调用频率、成功率、平均耗时。日志聚合使用结构化日志JSON格式并发送到ELK、Loki或云服务商的日志平台。确保每条日志都包含request_id和session_id便于追踪单个请求的全生命周期。告警对关键错误如Claude API持续失败、工具执行超时率高、延迟飙升、Token消耗异常等设置告警。4.3 常见问题排查与优化技巧Claude API返回“rate limit”错误现象日志中出现429状态码。排查检查你的调用频率是否超过套餐限制。Anthropic对不同模型和套餐有不同的RPM每分钟请求数和TPM每分钟Token数限制。解决在ClaudeAPIClient中实现更完善的退避重试机制上文已用tenacity实现。在应用层面实现请求队列或令牌桶算法平滑请求流量避免突发请求触发限流。考虑使用多个API密钥进行负载均衡如果允许且有必要。上下文长度超限Context Window Exhausted现象API返回错误提示输入过长。排查检查ConversationManager中保存的历史消息总token数。使用tiktoken库Anthropic官方推荐或API本身的计数功能进行精确计算。解决实现智能截断不要简单丢弃最老的消息。可以尝试优先保留最近的对话和系统提示对中间的历史进行摘要可以调用Claude的API来生成摘要但这本身消耗token。分片处理对于超长的单次用户输入如上传文档可以尝试将其分片分多次发送给Claude并让Claude进行增量理解这需要更复杂的状态管理。选择更大窗口的模型如果业务需要长上下文优先选用支持更大上下文窗口的模型版本。工具调用陷入死循环现象智能体反复调用同一个或一组工具无法给出最终答案。排查检查AgentService中的max_rounds限制是否生效。分析日志看工具返回的结果是否未能满足Claude的需求导致其反复尝试。解决优化工具描述和结果确保工具的描述清晰返回的结果格式明确、信息充足。模糊的结果会让Claude困惑。在System Prompt中增加约束明确告诉Claude“如果你已经调用过工具X并得到了Y结果请基于此直接回答用户不要再次调用”。实现会话级工具调用记忆记录本次会话中已调用过的工具及其主要输入输出在Claude再次请求调用相同工具时直接返回记忆中的结果或提醒它。响应速度慢现象用户端感知延迟高。排查使用APM工具定位瓶颈。常见瓶颈网络到Claude API的延迟、复杂工具的执行时间、数据库查询慢。解决异步化确保你的Web框架如FastAPI和所有I/O操作数据库、外部API调用都是异步的避免阻塞事件循环。缓存对频繁查询且变化不快的工具结果进行缓存如天气信息、某些计算结果的缓存。流式响应对于文本生成启用Claude API的流式响应并实现服务端的Server-Sent Events (SSE)让用户能边生成边看到结果提升体验。地理优化如果用户和你的服务器、Claude API服务器地理距离远考虑使用CDN或在不同区域部署边缘节点。System Prompt效果不佳现象智能体的行为不符合预期比如角色扮演不到位、频繁调用不该调用的工具。解决迭代优化System Prompt的编写是门艺术。需要不断测试和调整。使用A/B测试框架对比不同Prompt版本的效果。结构化Prompt将System Prompt分成几个清晰的部分角色定义、核心指令、输出格式要求、工具使用规范、禁忌事项。Few-shot示例在System Prompt中直接包含几个高质量的例子用户输入、助手理想输出的例子这是引导Claude行为最有效的方式之一。构建生产级的Claude API智能体是一个系统工程三层架构提供了一个清晰、稳固的起点。它迫使你思考每一部分的职责、边界和交互方式。从这基础的三层出发你可以根据业务复杂度的增长逐步引入更高级的特性如工作流引擎、长期记忆存储、复杂的工具编排等。记住好的架构不是一次性设计出来的而是在不断应对真实世界挑战的过程中迭代演化出来的。
