1. 项目概述与核心价值

最近在折腾AI应用开发，特别是围绕Claude API构建一些自动化工具时，发现一个挺普遍的问题：每次新建一个项目，都得从零开始搭建环境、配置API密钥、设计对话模板、处理错误重试……这些重复性工作不仅耗时，还容易出错。直到我在GitHub上发现了 walidboulanouar/ay-claude-templates 这个仓库，才感觉找到了一个能极大提升开发效率的“脚手架”。

简单来说， ay-claude-templates 是一个专门为Anthropic Claude API设计的、开箱即用的项目模板集合。它不是一个完整的应用，而是一个高度模块化的起点。你可以把它理解为一个“乐高积木箱”，里面已经为你准备好了搭建各种Claude对话应用所需的基础零件——标准化的消息处理模块、预配置的提示词模板、健壮的错误处理机制，以及清晰的目录结构。无论你是想快速构建一个命令行聊天工具、一个Web服务后端，还是一个复杂的多轮对话代理，这个模板都能让你跳过繁琐的初始化步骤，直接聚焦在业务逻辑的创新上。

这个项目的核心价值在于“标准化”和“可复用性”。作者 walidboulanouar 显然是从大量的实战中提炼出了最佳实践，并将其固化到了代码结构中。对于初学者，它是一份绝佳的学习资料，展示了如何专业地组织一个AI项目；对于有经验的开发者，它是一个高效的加速器，能节省至少半天到一天的搭建时间。接下来，我就结合自己的使用体验，深入拆解这个模板的设计思路、核心模块，并分享如何基于它快速启动你自己的Claude应用。

2. 项目结构与设计哲学解析

2.1 目录架构：清晰分离关注点

克隆项目后，第一眼看到的就是其清晰、标准的目录结构。这不仅仅是代码的摆放位置，更是其设计哲学的直观体现。

ay-claude-templates/
├── .env.example
├── .gitignore
├── README.md
├── requirements.txt
├── src/
│   ├── __init__.py
│   ├── claude_client.py
│   ├── config/
│   │   ├── __init__.py
│   │   └── settings.py
│   ├── prompts/
│   │   ├── __init__.py
│   │   └── templates.py
│   ├── utils/
│   │   ├── __init__.py
│   │   └── helpers.py
│   └── main.py
└── tests/
    ├── __init__.py
    └── test_claude_client.py

设计解读与实操意义：

1. src/ 作为核心源码区 ：这是所有业务逻辑的所在地。将代码集中于此，与配置文件、文档、测试隔离，符合现代Python项目的常见约定（如 src-layout ），有利于打包和依赖管理。
2. config/ 模块化配置 ：独立的配置目录意味着所有环境变量、API密钥、模型参数等“可变因素”都被集中管理。 settings.py 通常会使用 pydantic 之类的库进行强类型验证和加载，确保配置的可靠性和安全性。这是从“脚本”走向“工程”的关键一步。
3. prompts/ 提示词工程专区 ：这是本项目的一大亮点。将提示词模板从业务代码中剥离出来，单独放在 prompts/templates.py 中。这样做的好处是：
   • 可维护性 ：修改提示词无需翻找业务逻辑代码。
   • 可复用性 ：同一套提示词模板可以在不同场景下调用。
   • 版本控制 ：可以像管理代码一样，对提示词的迭代进行版本管理和对比。
4. utils/ 工具函数集 ：存放诸如日志记录、异常处理、数据清洗等辅助函数。保持 claude_client.py 等核心文件的简洁。
5. tests/ 测试目录 ：包含了针对核心客户端的基础测试，为项目质量提供了基础保障，也示范了如何为AI应用编写测试（例如，模拟API响应）。

注意 ：在实际使用中，我建议你立即将 .env.example 复制为 .env ，并填入你的 ANTHROPIC_API_KEY 。这是安全实践，避免将密钥硬编码在代码中或意外提交到版本库。

2.2 核心设计哲学：客户端与业务逻辑解耦

模板的核心是 src/claude_client.py 。它通常实现了一个 ClaudeClient 类，这个类的职责非常单一：封装与Anthropic API的直接交互。包括构建符合API格式的请求体、发送HTTP请求、处理响应和基础错误（如网络超时、认证失败）。

为什么这种解耦至关重要？

想象一下，如果你的业务逻辑代码（比如 main.py ）里到处散落着 requests.post(url, headers=..., json=...) 的调用，一旦API版本更新、请求格式变化，或者你需要增加重试逻辑、更换HTTP库，修改起来将是一场灾难。而通过一个专用的 Client 类，所有这些变化都被隔离在了一个文件中。你的业务代码只需要调用类似 client.chat(messages=..., model=...) 这样高级、易懂的方法。

一个典型的 ClaudeClient 类可能包含的方法：

• __init__(self, api_key, base_url, timeout) : 初始化客户端。
• chat(self, messages, model, max_tokens, temperature, **kwargs) : 发起对话补全请求。
• stream_chat(self, ...) : 处理流式响应（如果模板支持）。
• _handle_response(self, response) : 内部方法，统一处理API响应，解析JSON，转换错误码。

这种设计让你在构建复杂应用时，底层通信的稳定性得到了保障，你可以更专注于上层对话流程的设计和提示词的打磨。

3. 核心模块深度拆解与实战配置

3.1 配置管理 ( config/settings.py )：安全与灵活性的基石

一个健壮的AI应用，配置管理是第一步，也是最容易出错的一步。 ay-claude-templates 的配置模块提供了很好的范本。

# src/config/settings.py 示例扩展
from pydantic import Field
from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # 从 .env 文件加载
    ANTHROPIC_API_KEY: str = Field(..., min_length=1, description="Anthropic API密钥")
    ANTHROPIC_API_BASE_URL: str = "https://api.anthropic.com"
    ANTHROPIC_API_VERSION: str = "2023-06-01"
    
    # 模型默认配置
    DEFAULT_MODEL: str = "claude-3-haiku-20240307"
    DEFAULT_MAX_TOKENS: int = 1024
    DEFAULT_TEMPERATURE: float = 0.7
    
    # 应用行为配置
    REQUEST_TIMEOUT: int = 30
    ENABLE_LOGGING: bool = True
    LOG_LEVEL: str = "INFO"
    
    # 重试配置
    MAX_RETRIES: int = 3
    RETRY_BACKOFF_FACTOR: float = 0.5
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"
        case_sensitive = False

settings = Settings()

关键点解析与实操建议：

1. 使用Pydantic ： BaseSettings 不仅能从环境变量加载，还能提供强大的数据验证和类型提示。 Field(..., min_length=1) 确保API密钥不为空，这在应用启动时就能发现问题，而不是在运行时崩溃。
2. 分层配置 ：配置项被清晰地分为API相关、模型默认值、应用行为、重试策略等。这比把所有配置堆在一个字典里要清晰得多。
3. 默认值的智慧 ： DEFAULT_MODEL 选择了 claude-3-haiku 。这是一个成本、速度、能力平衡的模型，非常适合作为开发默认值。 DEFAULT_TEMPERATURE=0.7 提供了适度的创造性，又不会过于天马行空。
4. 超时与重试 ： REQUEST_TIMEOUT 和 MAX_RETRIES 是生产级应用必备的。网络请求总可能失败，有了重试机制（尤其是指数退避 RETRY_BACKOFF_FACTOR ），应用的鲁棒性会大大提升。

实操心得 ：我强烈建议你根据项目需求扩展这个配置类。例如，增加 PROXY 设置（如果需要）、 CACHE_ENABLED （用于缓存昂贵的API调用结果）、不同环境（开发、测试、生产）的配置区分等。使用 pydantic 的 validator 还可以实现更复杂的校验逻辑。

3.2 提示词模板 ( prompts/templates.py )：将“魔法”工程化

提示词是AI应用的灵魂，但也是最容易变得混乱的部分。这个模板将提示词管理提升到了工程层面。

# src/prompts/templates.py 示例扩展
from string import Template
from typing import Dict, Any

class PromptTemplates:
    """系统级和任务级提示词模板"""
    
    # 系统提示词（定义AI的角色和行为准则）
    SYSTEM_GENERIC_ASSISTANT = "You are a helpful, accurate, and harmless AI assistant."
    SYSTEM_CODE_EXPERT = "You are an expert software engineer. Provide clean, efficient, and well-documented code. Explain your reasoning concisely."
    SYSTEM_CREATIVE_WRITER = "You are a creative writer with a distinctive and engaging style. Respond in a narrative and evocative manner."
    
    # 任务提示词模板（使用string.Template进行变量替换）
    SUMMARIZE_TEXT = Template(
        "Please summarize the following text concisely, capturing the key points:\n\n"
        "Text: $text\n\n"
        "Summary:"
    )
    
    ANALYZE_SENTIMENT = Template(
        "Analyze the sentiment expressed in the following passage. "
        "Classify it as 'positive', 'negative', or 'neutral', and provide a brief reason.\n\n"
        "Passage: $passage\n\n"
        "Analysis:"
    )
    
    GENERATE_CODE_FROM_DESC = Template(
        "Write a $language function that $description. "
        "Include clear comments and error handling if appropriate.\n\n"
        "Function:"
    )
    
    CHAT_WITH_CONTEXT = Template(
        "Previous conversation context:\n$context\n\n"
        "Human: $current_query\n\n"
        "Assistant:"
    )
    
    @classmethod
    def get_system_prompt(cls, role: str = "generic") -> str:
        """获取系统提示词"""
        prompt_map = {
            "generic": cls.SYSTEM_GENERIC_ASSISTANT,
            "code_expert": cls.SYSTEM_CODE_EXPERT,
            "creative_writer": cls.SYSTEM_CREATIVE_WRITER,
        }
        return prompt_map.get(role, cls.SYSTEM_GENERIC_ASSISTANT)
    
    @classmethod
    def render(cls, template_name: str, **kwargs) -> str:
        """渲染任务提示词模板"""
        template_obj = getattr(cls, template_name.upper(), None)
        if not isinstance(template_obj, Template):
            raise ValueError(f"Template '{template_name}' not found or is not a Template.")
        return template_obj.safe_substitute(**kwargs)

设计精妙之处与扩展思路：

1. 系统提示词与任务提示词分离 ：这是Claude API（Messages API）的标准用法。系统提示词设定AI的“人设”，任务提示词提出具体请求。这种分离使得你可以轻松切换AI角色（如从“代码专家”切换到“创意写手”）而无需重写整个提示。
2. 使用 string.Template ：相比于f-string或 % 格式化， Template 更安全、意图更清晰。 safe_substitute 方法能优雅地处理缺失变量，避免运行时错误。
3. 模板类与方法化 ：通过 get_system_prompt 和 render 类方法，提供了统一、干净的调用接口。业务代码中只需 PromptTemplates.render('summarize_text', text=long_article) 即可。
4. 可扩展性极强 ：你可以很容易地添加新的模板。例如，添加一个 EXTRACT_ENTITIES 模板用于信息抽取，或者一个 TRANSLATE_TO_LANG 模板用于翻译。

实战进阶：动态模板与少量示例（Few-Shot） 对于更复杂的场景，模板可以不仅仅是字符串替换。你可以设计返回字典或列表的模板，用于构建复杂的消息历史。

# 进阶示例：包含少量示例（Few-Shot）的模板
FEW_SHOT_CLASSIFICATION = {
    "system": "You are a precise classification assistant.",
    "examples": [ # 少量示例对
        {"input": "I love this product! It's amazing.", "output": "positive"},
        {"input": "This is the worst experience ever.", "output": "negative"},
        {"input": "The package arrived on Tuesday.", "output": "neutral"},
    ],
    "template": Template("Classify the sentiment: $input")
}

# 在客户端中，你需要编写逻辑将此结构转换为API所需的messages格式。

3.3 核心客户端 ( claude_client.py )：稳健通信的保障

这是与Anthropic API交互的桥梁。一个优秀的客户端需要处理认证、请求构造、错误处理、重试、日志等方方面面。

# src/claude_client.py 核心部分详解
import logging
import time
from typing import Dict, List, Any, Optional, Iterator
import requests
from requests.exceptions import RequestException, Timeout

from .config.settings import settings

logger = logging.getLogger(__name__)

class ClaudeClient:
    """Anthropic Claude API 客户端"""
    
    def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None):
        self.api_key = api_key or settings.ANTHROPIC_API_KEY
        self.base_url = base_url or settings.ANTHROPIC_API_BASE_URL
        self.timeout = settings.REQUEST_TIMEOUT
        self.max_retries = settings.MAX_RETRIES
        self.backoff_factor = settings.RETRY_BACKOFF_FACTOR
        
        # 构建会话和固定请求头
        self.session = requests.Session()
        self.session.headers.update({
            "x-api-key": self.api_key,
            "anthropic-version": settings.ANTHROPIC_API_VERSION,
            "content-type": "application/json"
        })
        logger.info(f"ClaudeClient initialized with base URL: {self.base_url}")
    
    def _make_request(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
        """内部方法：执行HTTP请求，包含重试逻辑"""
        url = f"{self.base_url}{endpoint}"
        
        for attempt in range(self.max_retries + 1): # +1 包含首次尝试
            try:
                logger.debug(f"Attempt {attempt+1}/{self.max_retries+1}: POST {url}")
                response = self.session.post(url, json=payload, timeout=self.timeout)
                response.raise_for_status() # 检查HTTP错误（4xx, 5xx）
                return response.json()
                
            except Timeout:
                logger.warning(f"Request timeout on attempt {attempt+1}")
                if attempt == self.max_retries:
                    raise Exception(f"API request timed out after {self.max_retries+1} attempts.")
            except RequestException as e:
                logger.error(f"Request failed on attempt {attempt+1}: {e}")
                if attempt == self.max_retries:
                    raise Exception(f"API request failed after {self.max_retries+1} attempts: {e}")
            
            # 指数退避等待
            if attempt < self.max_retries:
                wait_time = self.backoff_factor * (2 ** attempt)
                logger.info(f"Retrying in {wait_time:.2f} seconds...")
                time.sleep(wait_time)
    
    def chat(self, 
             messages: List[Dict[str, str]], 
             model: Optional[str] = None,
             max_tokens: Optional[int] = None,
             temperature: Optional[float] = None,
             **kwargs) -> Dict[str, Any]:
        """
        发起非流式对话请求
        
        Args:
            messages: 消息列表，格式如 [{"role": "user", "content": "Hello"}]
            model: 模型名称，默认使用配置
            max_tokens: 生成的最大token数
            temperature: 温度参数，控制随机性
            **kwargs: 其他API参数（如top_p, stop_sequences等）
        
        Returns:
            API返回的完整JSON响应字典
        """
        model = model or settings.DEFAULT_MODEL
        max_tokens = max_tokens or settings.DEFAULT_MAX_TOKENS
        temperature = temperature or settings.DEFAULT_TEMPERATURE
        
        payload = {
            "model": model,
            "messages": messages,
            "max_tokens": max_tokens,
            "temperature": temperature,
            **kwargs # 合并其他可选参数
        }
        
        # 移除值为None的参数，保持请求体清洁
        payload = {k: v for k, v in payload.items() if v is not None}
        
        logger.info(f"Sending chat request to model '{model}', {len(messages)} messages.")
        response_data = self._make_request("/v1/messages", payload)
        
        # 提取主要回复内容，方便调用者使用
        if "content" in response_data and len(response_data["content"]) > 0:
            response_data["extracted_text"] = response_data["content"][0]["text"]
        
        logger.debug(f"Received response with ID: {response_data.get('id')}")
        return response_data
    
    def stream_chat(self, messages: List[Dict[str, str]], **kwargs) -> Iterator[str]:
        """
        发起流式对话请求（生成器）
        
        Yields:
            每个流式块的文本内容
        """
        # 注意：流式请求需要设置 stream=True，且处理方式不同
        # 此处为简化示例，实际实现需处理Server-Sent Events (SSE)
        payload = {
            "model": kwargs.pop("model", settings.DEFAULT_MODEL),
            "messages": messages,
            "max_tokens": kwargs.pop("max_tokens", settings.DEFAULT_MAX_TOKENS),
            "temperature": kwargs.pop("temperature", settings.DEFAULT_TEMPERATURE),
            "stream": True,
            **kwargs
        }
        # 流式处理逻辑（略，通常使用requests的iter_lines或专用SSE客户端）
        # yield from _handle_stream(response)
        pass

关键实现细节与避坑指南：

1. 使用 requests.Session ：复用TCP连接，提升多次请求的性能。
2. 集中设置请求头 ：在 __init__ 中设置好 x-api-key 和 anthropic-version 等固定头，避免每次请求重复设置。
3. 健壮的重试机制 ( _make_request ) ：这是生产级代码的标志。它处理了网络超时和请求异常，并实现了指数退避。注意，重试只适用于网络类错误（ Timeout , RequestException ）。对于API返回的业务逻辑错误（如 400 Bad Request ），不应重试，因为重试相同的错误请求没有意义。 response.raise_for_status() 会抛出HTTPError，被 RequestException 捕获。
4. 参数清理与默认值 ： chat 方法中， payload = {k: v for k, v in payload.items() if v is not None} 这行代码移除了所有 None 值，防止向API发送无效字段。同时，它巧妙地使用 or 运算符和配置中的默认值，提供了灵活的调用方式。
5. 日志记录 ：在不同级别（ info , debug , warning , error ）记录关键操作和错误，对于调试和监控至关重要。
6. 流式处理占位 ： stream_chat 方法展示了流式接口的框架。实际实现需要使用 requests 的 stream=True 模式并解析SSE格式，或者使用 anthropic 官方SDK中更成熟的流式处理逻辑。

重要提醒 ：对于真正的生产环境，你还需要考虑增加请求速率限制（Rate Limiting）处理、更精细的错误分类（如区分额度不足、模型过载等）、以及可能的响应缓存机制。

4. 基于模板的快速启动与项目实战

4.1 五分钟快速启动指南

假设你已经克隆了 ay-claude-templates 仓库，下面是如何在五分钟内将其改造成一个可运行的命令行聊天工具。

步骤一：环境准备与配置

# 1. 进入项目目录
cd ay-claude-templates

# 2. 创建虚拟环境（推荐）
python -m venv venv
# Windows: venv\Scripts\activate
# Mac/Linux: source venv/bin/activate

# 3. 安装依赖
pip install -r requirements.txt  # 通常包含 requests, pydantic, python-dotenv等

# 4. 复制环境变量文件并配置
cp .env.example .env
# 编辑 .env 文件，填入你的 ANTHROPIC_API_KEY

步骤二：创建你的第一个应用脚本 在项目根目录或 src 同级创建一个新文件，例如 my_chatbot.py 。

# my_chatbot.py
import sys
from src.claude_client import ClaudeClient
from src.prompts.templates import PromptTemplates

def simple_chat():
    """一个简单的交互式命令行聊天"""
    client = ClaudeClient()
    
    print("Claude 命令行聊天助手 (输入 'quit' 退出)")
    print("-" * 40)
    
    # 初始化对话历史
    messages = [
        {"role": "system", "content": PromptTemplates.get_system_prompt("generic")}
    ]
    
    while True:
        try:
            user_input = input("\nYou: ").strip()
            if user_input.lower() in ['quit', 'exit', 'q']:
                print("再见！")
                break
            if not user_input:
                continue
                
            # 将用户输入加入历史
            messages.append({"role": "user", "content": user_input})
            
            print("Claude: ", end="", flush=True)
            
            # 调用API
            response = client.chat(
                messages=messages,
                max_tokens=500, # 可以根据需要调整
            )
            
            # 获取回复文本
            assistant_reply = response.get("extracted_text", "（未收到有效回复）")
            print(assistant_reply)
            
            # 将助手回复加入历史，维持上下文
            messages.append({"role": "assistant", "content": assistant_reply})
            
        except KeyboardInterrupt:
            print("\n\n程序被中断。")
            break
        except Exception as e:
            print(f"\n发生错误: {e}")
            # 可以选择移除最后一次出错的用户输入，避免历史混乱
            if messages and messages[-1]["role"] == "user":
                messages.pop()
            # 或者简单提示后继续
            continue

if __name__ == "__main__":
    simple_chat()

步骤三：运行

python my_chatbot.py

现在，你就拥有了一个具备上下文记忆的、基于Claude API的命令行聊天工具。它使用了模板中提供的标准化客户端和提示词管理。

4.2 实战案例一：构建一个文本摘要服务

让我们利用模板，快速构建一个更实用的服务：一个文本摘要API端点。

步骤一：设计摘要提示词 在 src/prompts/templates.py 中添加更专业的摘要模板：

# 在PromptTemplates类中添加
    SUMMARIZE_LONG_DOCUMENT = Template(
        "You are a concise summarization expert. Please provide a summary of the following text.\n"
        "**Requirements:**\n"
        "1. Capture the main thesis and key supporting points.\n"
        "2. Limit the summary to $max_sentences sentences.\n"
        "3. Use clear and objective language.\n"
        "4. Do not include your own opinions or external information.\n\n"
        "Text to summarize:\n$text\n\n"
        "Summary:"
    )

步骤二：创建摘要服务模块 创建 src/summarizer.py :

# src/summarizer.py
import logging
from typing import Optional
from .claude_client import ClaudeClient
from .prompts.templates import PromptTemplates

logger = logging.getLogger(__name__)

class TextSummarizer:
    def __init__(self, client: Optional[ClaudeClient] = None):
        self.client = client or ClaudeClient()
    
    def summarize(self, 
                  text: str, 
                  max_sentences: int = 3,
                  model: str = "claude-3-haiku-20240307") -> str:
        """
        对长文本进行摘要
        
        Args:
            text: 需要摘要的文本
            max_sentences: 摘要最大句数
            model: 使用的Claude模型
        
        Returns:
            摘要文本
        """
        if not text or len(text.strip()) < 50:
            logger.warning("Input text is too short for summarization.")
            return text[:200] + "..." if len(text) > 200 else text
        
        # 构建提示词
        prompt = PromptTemplates.render(
            'summarize_long_document',
            text=text,
            max_sentences=max_sentences
        )
        
        messages = [
            {"role": "user", "content": prompt}
        ]
        
        try:
            response = self.client.chat(
                messages=messages,
                model=model,
                max_tokens=500, # 为摘要分配足够的token
                temperature=0.3 # 较低的温度，确保摘要更忠实、稳定
            )
            
            summary = response.get("extracted_text", "").strip()
            logger.info(f"Successfully summarized text (length: {len(text)} -> {len(summary)})")
            return summary
            
        except Exception as e:
            logger.error(f"Summarization failed: {e}")
            # 优雅降级：返回文本开头作为简单摘要
            return f"{text[:150]}... [摘要生成失败]"
    
    def summarize_batch(self, texts: list, **kwargs) -> list:
        """批量摘要（简单串行实现，生产环境可考虑并发）"""
        return [self.summarize(text, **kwargs) for text in texts]

步骤三：集成到Web服务（使用FastAPI示例） 安装FastAPI: pip install fastapi uvicorn

创建 app.py :

# app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from src.summarizer import TextSummarizer
import logging

logging.basicConfig(level=logging.INFO)
app = FastAPI(title="Claude Text Summarization API")
summarizer = TextSummarizer()

class SummarizeRequest(BaseModel):
    text: str = Field(..., min_length=10, description="需要摘要的文本")
    max_sentences: int = Field(3, ge=1, le=10, description="摘要最大句数")
    model: str = Field("claude-3-haiku-20240307", description="Claude模型")

class SummarizeResponse(BaseModel):
    summary: str
    original_length: int
    summary_length: int
    model_used: str

@app.post("/summarize", response_model=SummarizeResponse)
async def summarize_text(request: SummarizeRequest):
    """文本摘要端点"""
    try:
        summary = summarizer.summarize(
            text=request.text,
            max_sentences=request.max_sentences,
            model=request.model
        )
        
        return SummarizeResponse(
            summary=summary,
            original_length=len(request.text),
            summary_length=len(summary),
            model_used=request.model
        )
        
    except Exception as e:
        logging.error(f"API error: {e}")
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {"status": "healthy", "service": "claude-summarizer"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

步骤四：运行并测试

# 启动服务
python app.py

# 另开终端，测试API
curl -X POST "http://localhost:8000/summarize" \
  -H "Content-Type: application/json" \
  -d '{
    "text": "这里是你的长篇文章内容...可以是一篇新闻、报告或论文摘要。人工智能是当今科技发展的重要方向，它正在改变我们的生活和工作方式。机器学习、深度学习等技术在图像识别、自然语言处理等领域取得了突破性进展。未来，AI将在医疗、教育、交通等更多领域发挥重要作用。",
    "max_sentences": 2,
    "model": "claude-3-haiku-20240307"
  }'

通过这个案例，你看到了如何从模板出发，快速构建一个具备专业提示词、健壮错误处理、可配置参数，并且能对外提供API服务的完整应用。模板提供的结构让这一切变得有条不紊。

4.3 实战案例二：构建一个带记忆的对话代理

更复杂的场景是构建一个能记住多轮对话上下文，并且有特定领域知识的代理。我们利用模板来构建一个“技术文档助手”。

步骤一：设计系统提示词和上下文管理 在 prompts/templates.py 中增加：

    SYSTEM_TECH_DOC_ASSISTANT = (
        "You are an expert technical documentation assistant. Your role is to help users understand, "
        "navigate, and work with technical documentation and code. You are precise, detail-oriented, "
        "and always reference specific details from the conversation history when possible. "
        "If you're unsure about something, acknowledge it rather than guessing."
    )

创建 src/agents/tech_doc_agent.py :

# src/agents/tech_doc_agent.py
import logging
from typing import List, Dict, Optional
from datetime import datetime
from ..claude_client import ClaudeClient
from ..prompts.templates import PromptTemplates

logger = logging.getLogger(__name__)

class TechDocAgent:
    """技术文档对话代理，具备上下文记忆"""
    
    def __init__(self, client: Optional[ClaudeClient] = None, max_history: int = 20):
        self.client = client or ClaudeClient()
        self.max_history = max_history  # 最大对话轮次记忆
        self.conversation_history: List[Dict[str, str]] = []
        self.system_prompt = PromptTemplates.get_system_prompt("tech_doc_assistant")
        
        # 初始化系统消息
        self._initialize_conversation()
    
    def _initialize_conversation(self):
        """初始化对话，加入系统提示"""
        self.conversation_history = [
            {"role": "system", "content": self.system_prompt},
            {"role": "assistant", "content": "您好！我是技术文档助手，可以帮您解答关于文档、API、代码等方面的问题。请告诉我您需要什么帮助？"}
        ]
    
    def _trim_history(self):
        """修剪对话历史，防止超出token限制"""
        # 简单的策略：保留系统消息和最近N轮对话
        if len(self.conversation_history) > self.max_history + 1:  # +1 是系统消息
            # 保留系统消息和最近的用户/助手对话
            self.conversation_history = [self.conversation_history[0]] + self.conversation_history[-(self.max_history):]
            logger.debug(f"Trimmed conversation history to {len(self.conversation_history)} messages")
    
    def add_user_message(self, message: str):
        """添加用户消息到历史"""
        if not message.strip():
            return
        self.conversation_history.append({"role": "user", "content": message.strip()})
    
    def generate_response(self, user_message: Optional[str] = None) -> str:
        """
        生成助手回复
        
        Args:
            user_message: 可选，如果提供则先添加到历史
        
        Returns:
            助手的回复文本
        """
        if user_message:
            self.add_user_message(user_message)
        
        if len(self.conversation_history) <= 1:  # 只有系统消息
            return "请先输入您的问题。"
        
        try:
            logger.info(f"Generating response with {len(self.conversation_history)} messages in history")
            
            response = self.client.chat(
                messages=self.conversation_history,
                model="claude-3-sonnet-20240229",  # 使用能力更强的模型处理复杂对话
                max_tokens=800,
                temperature=0.5  # 适度的创造性
            )
            
            assistant_reply = response.get("extracted_text", "").strip()
            
            if assistant_reply:
                self.conversation_history.append({"role": "assistant", "content": assistant_reply})
                self._trim_history()  # 添加回复后修剪历史
                
            return assistant_reply
            
        except Exception as e:
            logger.error(f"Failed to generate response: {e}")
            error_msg = "抱歉，处理您的请求时出现了问题。请稍后再试或简化您的问题。"
            self.conversation_history.append({"role": "assistant", "content": error_msg})
            return error_msg
    
    def reset_conversation(self):
        """重置对话历史"""
        self._initialize_conversation()
        logger.info("Conversation reset")
    
    def get_conversation_summary(self) -> Dict:
        """获取当前对话的摘要信息"""
        user_msgs = [msg for msg in self.conversation_history if msg["role"] == "user"]
        assistant_msgs = [msg for msg in self.conversation_history if msg["role"] == "assistant"]
        
        return {
            "total_turns": len(user_msgs),
            "last_user_message": user_msgs[-1]["content"][:100] + "..." if user_msgs else None,
            "start_time": datetime.now().isoformat(),
            "history_length": len(self.conversation_history)
        }

步骤二：创建对话管理工具 创建 src/utils/conversation_manager.py 来提供更高级的上下文管理功能：

# src/utils/conversation_manager.py
import json
import hashlib
from typing import Dict, List, Any, Optional
from pathlib import Path

class ConversationManager:
    """管理对话会话的保存、加载和检索"""
    
    def __init__(self, storage_dir: str = "./conversation_sessions"):
        self.storage_dir = Path(storage_dir)
        self.storage_dir.mkdir(exist_ok=True)
    
    def _generate_session_id(self, initial_prompt: str) -> str:
        """根据初始提示生成会话ID"""
        hash_input = f"{initial_prompt}_{datetime.now().timestamp()}"
        return hashlib.md5(hash_input.encode()).hexdigest()[:8]
    
    def save_session(self, 
                    session_id: str, 
                    history: List[Dict[str, str]], 
                    metadata: Optional[Dict] = None):
        """保存对话会话到文件"""
        session_file = self.storage_dir / f"{session_id}.json"
        
        session_data = {
            "session_id": session_id,
            "history": history,
            "metadata": metadata or {},
            "saved_at": datetime.now().isoformat()
        }
        
        with open(session_file, 'w', encoding='utf-8') as f:
            json.dump(session_data, f, ensure_ascii=False, indent=2)
    
    def load_session(self, session_id: str) -> Optional[Dict[str, Any]]:
        """从文件加载对话会话"""
        session_file = self.storage_dir / f"{session_id}.json"
        
        if not session_file.exists():
            return None
        
        with open(session_file, 'r', encoding='utf-8') as f:
            return json.load(f)
    
    def list_sessions(self) -> List[Dict[str, Any]]:
        """列出所有保存的会话"""
        sessions = []
        for session_file in self.storage_dir.glob("*.json"):
            try:
                with open(session_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                    sessions.append({
                        "session_id": data.get("session_id", session_file.stem),
                        "saved_at": data.get("saved_at"),
                        "message_count": len(data.get("history", [])),
                        "metadata": data.get("metadata", {})
                    })
            except (json.JSONDecodeError, IOError) as e:
                print(f"Error reading session file {session_file}: {e}")
        
        return sorted(sessions, key=lambda x: x.get("saved_at", ""), reverse=True)

步骤三：创建完整的对话应用 创建 tech_doc_chat.py :

# tech_doc_chat.py
import sys
import logging
from src.agents.tech_doc_agent import TechDocAgent
from src.utils.conversation_manager import ConversationManager

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

def main():
    print("=" * 60)
    print("技术文档助手 v1.0")
    print("命令: /save, /load, /reset, /list, /exit")
    print("=" * 60)
    
    agent = TechDocAgent()
    manager = ConversationManager()
    current_session_id = None
    
    while True:
        try:
            user_input = input("\n您: ").strip()
            
            # 处理命令
            if user_input.startswith('/'):
                cmd = user_input[1:].lower().split()[0] if user_input[1:] else ""
                
                if cmd == "exit":
                    print("再见！")
                    break
                    
                elif cmd == "reset":
                    agent.reset_conversation()
                    current_session_id = None
                    print("对话已重置。")
                    continue
                    
                elif cmd == "save":
                    if current_session_id:
                        # 保存当前会话
                        manager.save_session(
                            session_id=current_session_id,
                            history=agent.conversation_history,
                            metadata={"description": "技术文档对话"}
                        )
                        print(f"会话已保存，ID: {current_session_id}")
                    else:
                        # 创建新会话
                        from datetime import datetime
                        new_id = manager._generate_session_id(agent.conversation_history[0]["content"])
                        manager.save_session(
                            session_id=new_id,
                            history=agent.conversation_history,
                            metadata={"description": "技术文档对话", "created": datetime.now().isoformat()}
                        )
                        current_session_id = new_id
                        print(f"新会话已创建并保存，ID: {new_id}")
                    continue
                    
                elif cmd == "list":
                    sessions = manager.list_sessions()
                    if not sessions:
                        print("没有保存的会话。")
                    else:
                        print("\n保存的会话:")
                        for i, sess in enumerate(sessions[:5], 1):  # 显示最近5个
                            print(f"{i}. ID: {sess['session_id']} | 消息数: {sess['message_count']} | 保存时间: {sess['saved_at'][:19]}")
                    continue
                    
                elif cmd == "load" and len(user_input.split()) > 1:
                    session_id = user_input.split()[1]
                    session_data = manager.load_session(session_id)
                    if session_data:
                        agent.conversation_history = session_data["history"]
                        current_session_id = session_id
                        print(f"已加载会话 {session_id}，包含 {len(agent.conversation_history)-1} 轮对话。")
                        # 显示最后一条消息
                        last_msg = agent.conversation_history[-1]["content"][:100]
                        print(f"最后消息: {last_msg}...")
                    else:
                        print(f"未找到会话 {session_id}")
                    continue
                    
                else:
                    print(f"未知命令: {cmd}")
                    print("可用命令: /save, /load <id>, /reset, /list, /exit")
                    continue
            
            # 普通对话
            if not user_input:
                continue
                
            print("\n助手: ", end="", flush=True)
            response = agent.generate_response(user_input)
            print(response)
            
            # 显示对话统计
            summary = agent.get_conversation_summary()
            print(f"\n[对话轮次: {summary['total_turns']} | 历史长度: {summary['history_length']}]")
            
        except KeyboardInterrupt:
            print("\n\n对话中断。")
            save_prompt = input("是否保存当前会话？ (y/n): ").lower()
            if save_prompt == 'y':
                if not current_session_id:
                    from datetime import datetime
                    current_session_id = manager._generate_session_id(agent.conversation_history[0]["content"])
                manager.save_session(current_session_id, agent.conversation_history)
                print(f"会话已保存为: {current_session_id}")
            break
            
        except Exception as e:
            logging.error(f"Unexpected error: {e}")
            print("发生意外错误，请重试。")

if __name__ == "__main__":
    main()

这个案例展示了如何基于模板构建一个复杂的、具备状态管理的对话代理。它包含了：

1. 上下文记忆 ：通过维护 conversation_history 列表实现。
2. 历史修剪 ：防止上下文过长导致API调用token超限或成本过高。
3. 会话持久化 ：能够保存和加载对话状态。
4. 交互式命令 ：提供用户友好的控制命令。

5. 高级技巧、优化与生产级考量

5.1 性能优化：异步处理与并发请求

当需要处理大量文本或并发用户请求时，同步的API调用会成为瓶颈。我们可以利用异步编程来提升性能。

创建异步客户端 ( src/async_claude_client.py ):

# src/async_claude_client.py
import aiohttp
import asyncio
import logging
from typing import Dict, List, Any, Optional
from .config.settings import settings

logger = logging.getLogger(__name__)

class AsyncClaudeClient:
    """异步Claude API客户端"""
    
    def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None):
        self.api_key = api_key or settings.ANTHROPIC_API_KEY
        self.base_url = base_url or settings.ANTHROPIC_API_BASE_URL
        self.timeout = aiohttp.ClientTimeout(total=settings.REQUEST_TIMEOUT)
        self.max_retries = settings.MAX_RETRIES
        self.session: Optional[aiohttp.ClientSession] = None
    
    async def __aenter__(self):
        """异步上下文管理器入口"""
        self.session = aiohttp.ClientSession(
            headers={
                "x-api-key": self.api_key,
                "anthropic-version": settings.ANTHROPIC_API_VERSION,
                "content-type": "application/json"
            },
            timeout=self.timeout
        )
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """异步上下文管理器出口"""
        if self.session:
            await self.session.close()
    
    async def _make_request(self, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
        """内部异步请求方法，带重试"""
        url = f"{self.base_url}{endpoint}"
        
        for attempt in range(self.max_retries + 1):
            try:
                logger.debug(f"Async attempt {attempt+1}/{self.max_retries+1}: POST {url}")
                
                async with self.session.post(url, json=payload) as response:
                    response.raise_for_status()
                    return await response.json()
                    
            except asyncio.TimeoutError:
                logger.warning(f"Async request timeout on attempt {attempt+1}")
                if attempt == self.max_retries:
                    raise Exception(f"API request timed out after {self.max_retries+1} attempts.")
            except aiohttp.ClientError as e:
                logger.error(f"Async request failed on attempt {attempt+1}: {e}")
                if attempt == self.max_retries:
                    raise Exception(f"API request failed after {self.max_retries+1} attempts: {e}")
            
            # 指数退避
            if attempt < self.max_retries:
                wait_time = settings.RETRY_BACKOFF_FACTOR * (2 ** attempt)
                logger.info(f"Retrying in {wait_time:.2f} seconds...")
                await asyncio.sleep(wait_time)
    
    async def chat(self, 
                   messages: List[Dict[str, str]], 
                   model: Optional[str] = None,
                   max_tokens: Optional[int] = None,
                   temperature: Optional[float] = None,
                   **kwargs) -> Dict[str, Any]:
        """异步对话请求"""
        model = model or settings.DEFAULT_MODEL
        max_tokens = max_tokens or settings.DEFAULT_MAX_TOKENS
        temperature = temperature or settings.DEFAULT_TEMPERATURE
        
        payload = {
            "model": model,
            "messages": messages,
            "max_tokens": max_tokens,
            "temperature": temperature,
            **{k: v for k, v in kwargs.items() if v is not None}
        }
        
        logger.info(f"Async sending chat request to model '{model}'")
        response_data = await self._make_request("/v1/messages", payload)
        
        if "content" in response_data and len(response_data["content"]) > 0:
            response_data["extracted_text"] = response_data["content"][0]["text"]
        
        return response_data
    
    async def chat_batch(self, 
                        messages_list: List[List[Dict[str, str]]], 
                        **kwargs) -> List[Dict[str, Any]]:
        """批量异步请求（注意API速率限制）"""
        tasks = []
        for messages in messages_list:
            task = self.chat(messages, **kwargs)
            tasks.append(task)
        
        # 限制并发数，避免触发API速率限制
        semaphore = asyncio.Semaphore(5)  # 最大5个并发
        
        async def limited_task(task):
            async with semaphore:
                return await task
        
        limited_tasks = [limited_task(task) for task in tasks]
        results = await asyncio.gather(*limited_tasks, return_exceptions=True)
        
        # 处理结果，分离成功和失败
        successful = []
        failed = []
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                logger.error(f"Request {i} failed: {result}")
                failed.append({"index": i, "error": str(result)})
            else:
                successful.append(result)
        
        logger.info(f"Batch completed: {len(successful)} successful, {len(failed)} failed")
        return successful, failed

使用异步客户端进行批量处理：

# batch_process.py
import asyncio
import logging
from src.async_claude_client import AsyncClaudeClient
from src.prompts.templates import PromptTemplates

logging.basicConfig(level=logging.INFO)

async def batch_summarize_articles(articles: List[str]):
    """批量摘要多篇文章"""
    
    # 准备消息列表
    messages_list = []
    for article in articles:
        prompt = PromptTemplates.render('summarize_long_document', 
                                       text=article, 
                                       max_sentences=2)
        messages_list.append([{"role": "user", "content": prompt}])
    
    # 使用异步客户端
    async with AsyncClaudeClient() as client:
        successful, failed = await client.chat_batch(
            messages_list,
            model="claude-3-haiku-20240307",
            max_tokens=300,
            temperature=0.3
        )
    
    # 处理结果
    summaries = []
    for result in successful:
        summary = result.get("extracted_text", "")
        summaries.append(summary)
    
    return summaries, failed

# 使用示例
if __name__ == "__main__":
    articles = [
        "第一篇文章内容...",
        "第二篇文章内容...",
        # ... 更多文章
    ]
    
    summaries, errors = asyncio.run(batch_summarize_articles(articles[:10]))  # 先测试10篇
    
    for i, summary in enumerate(summaries):
        print(f"文章{i+1}摘要: {summary[:100]}...")
    
    if errors:
        print(f"\n处理失败的文章索引: {[e['index'] for e in errors]}")

5.2 成本控制与监控

使用Claude API会产生费用，对于生产应用，成本控制至关重要。

实现一个带成本估算的装饰器：

# src/utils/cost_tracker.py
import functools
import time
from typing import Callable, Any
import logging
from dataclasses import dataclass, field
from datetime import datetime

logger = logging.getLogger(__name__)

# Claude API定价（示例，请以官方最新价格为准）
CLAUDE_PRICING = {
    "claude-3-haiku-20240307": {"input": 0.25, "output": 1.25},  # $ per 1M tokens
    "claude-3-sonnet-20240229": {"input": 3.0, "output": 15.0},
    "claude-3-opus-20240229": {"input": 15.0, "output": 75.0},
}

@dataclass
class CostRecord:
    """单次调用成本记录"""
    timestamp: datetime
    model: str
    input_tokens: int
    output_tokens: int
    input_cost: float = field(init=False)
    output_cost: float = field(init=False)
    total_cost: float = field(init=False)
    
    def __post_init__(self):
        pricing = CLAUDE_PRICING.get(self.model, CLAUDE_PRICING["claude-3-haiku-20240307"])
        self.input_cost = (self.input_tokens / 1_000_000) * pricing["input"]
        self.output_cost = (self.output_tokens / 1_000_000) * pricing["output"]
        self.total_cost = self.input_cost + self.output_cost

class CostTracker:
    """API调用成本跟踪器"""
    
    def __init__(self):
        self.records: List[CostRecord] = []
        self.total_cost = 0.0
    
    def add_record(self, record: CostRecord):
        self.records.append(record)
        self.total_cost += record.total_cost
        logger.info(f"API调用成本: ${record.total_cost:.6f} (输入: {record.input_tokens}t, 输出: {record.output_tokens}t)")
    
    def get_daily_summary(self, date: datetime = None) -> Dict:
        """获取指定日期的成本摘要"""
        if date is None:
            date = datetime.now().date()
        
        daily_records = [r for r in self.records if r.timestamp.date() == date]
        total_cost = sum(r.total_cost for r in daily_records)
        
        return {
            "date": date.isoformat(),
            "call_count": len(daily_records),
            "total_input_tokens": sum(r.input_tokens for r in daily_records),
            "total_output_tokens": sum(r.output_tokens for r in daily_records),
            "total_cost": total_cost,
            "by_model": self._group_by_model(daily_records)
        }
    
    def _group_by_model(self, records: List[CostRecord]) -> Dict:
        """按模型分组统计"""
        grouped = {}
        for record in records:
            if record.model not in grouped:
                grouped[record.model] = {
                    "call_count": 0,
                    "input_tokens": 0,
                    "output_tokens": 0,
                    "cost": 0.0
                }
            grouped[record.model]["call_count"] += 1
            grouped[record.model]["input_tokens"] += record.input_tokens
            grouped[record.model]["output_tokens"] += record.output_tokens
            grouped[record.model]["cost"] += record.total_cost
        
        return grouped

# 全局成本跟踪器实例
cost_tracker = CostTracker()

def track_cost(func: Callable) -> Callable:
    """成本跟踪装饰器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # 调用原始函数
        start_time = time.time()
        result = func(*args, **kwargs)
        elapsed = time.time() - start_time
        
        # 提取成本信息（假设返回的response中包含token使用量）
        if isinstance(result, dict) and "usage" in result:
            usage = result["usage"]
            model = kwargs.get("model") or getattr(args[0] if args else None, "model", "unknown")
            
            record = CostRecord(
                timestamp=datetime.now(),
                model=model,
                input_tokens=usage.get("input_tokens", 0),
                output_tokens=usage.get("output_tokens", 0)
            )
            
            cost_tracker.add_record(record)
            
            # 添加成本信息到结果中
            result["cost_info"] = {
                "input_tokens": record.input_tokens,
                "output_tokens": record.output_tokens,
                "estimated_cost": record.total_cost,
                "processing_time": elapsed
            }
        
        return result
    return wrapper

# 在客户端中使用装饰器
class TrackedClaudeClient(ClaudeClient):
    """带成本跟踪的客户端"""
    
    @track_cost
    def chat(self, *args, **kwargs):
        return super().chat(*args, **kwargs)

使用成本跟踪：

# 在配置中启用成本跟踪
from src.utils.cost_tracker import TrackedClaudeClient, cost_tracker

client = TrackedClaudeClient()

# 正常使用API
response = client.chat(messages=[{"role": "user", "content": "Hello"}])

# 查看成本信息
if "cost_info" in response:
    print(f"本次调用消耗: ${response['cost_info']['estimated_cost']:.6f}")
    print(f"处理时间: {response['cost_info']['processing_time']:.2f}秒")

# 查看当日摘要
daily_summary = cost_tracker.get_daily_summary()
print(f"今日API调用统计: {daily_summary['call_count']}次, 总成本: ${daily_summary['total_cost']:.4f}")

5.3 错误处理与重试策略优化

基础的网络重试之外，还需要处理API特定的错误。

增强的错误处理器：

# src/utils/error_handler.py
import time
from typing import Optional, Callable, Any
import logging
from requests.exceptions import RequestException, Timeout

logger = logging.getLogger(__name__)

class ClaudeAPIError(Exception):
    """Claude API特定错误"""
    def __init__(self, message: str, status_code: Optional[int] = None, error_type: Optional[str] = None):
        self.message = message
        self.status_code = status_code
        self.error_type = error_type
        super().__init__(self.message)

def retry_with_backoff(
    max_retries: int = 3,
    initial_delay: float = 1.0,
    backoff_factor: float = 2.0,
    retryable_errors: Optional[list] = None
):
    """
    带退避的重试装饰器，支持特定错误重试
    
    Args:
        max_retries: 最大重试次数
        initial_delay: 初始延迟（秒）
        backoff_factor: 退避因子
        retryable_errors: 可重试的错误类型列表
    """
    if retryable_errors is None:
        retryable_errors = [Timeout, ConnectionError]
    
    def decorator(func: Callable) -> Callable:
        def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                    
                except Exception as e:
                    last_exception = e
                    
                    # 检查是否可重试的错误
                    retryable = any(isinstance(e, error_type) for error_type in retryable_errors)
                    
                    # 检查API错误是否可重试
                    if isinstance(e, ClaudeAPIError):
                        # 这些错误通常不应该重试
                        if e.status_code in [400, 401, 403, 404, 429]:
                            if e.status_code == 429:  # 速率限制，可以重试
                                retryable = True
                                logger.warning(f"Rate limited (429), will retry")
                            else:
                                retryable = False
                                logger.error(f"API error {e.status_code}: {e.message}")
                    
                    if not retryable or attempt == max_retries:
                        raise last_exception
                    
                    # 计算等待时间
                    wait_time = initial_delay * (backoff_factor ** attempt)
                    logger.warning(
                        f"Attempt {attempt+1}/{max_retries+1} failed: {type(e).__name__}: {str(e)[:100]}"
                    )
                    logger.info(f"Waiting {wait_time:.2f} seconds before retry...")
                    time.sleep(wait_time)
            
            raise last_exception
        
        return wrapper
    return decorator

def handle_api_response(response):
    """处理API响应，转换错误"""
    if response.status_code == 200:
        return response.json()
    
    # 处理错误响应
    error_info = {}
    try:
        error_info = response.json()
    except:
        error_info = {"error": {"message": response.text}}
    
    error_msg = error_info.get("error", {}).get("message", "Unknown API error")
    error_type = error_info.get("error", {}).get("type", "unknown")
    
    # 根据状态码和错误类型提供更友好的消息
    if response.status_code == 400:
        raise ClaudeAPIError(f"Bad request: {error_msg}", 400, error_type)
    elif response.status_code == 401:
        raise ClaudeAPIError("Invalid API key or authentication failed", 401, error_type)
    elif response.status_code == 403:
        raise ClaudeAPIError("Insufficient permissions or access denied", 403, error_type)
    elif response.status_code == 404:
        raise ClaudeAPIError("Resource not found", 404, error_type)
    elif response.status_code == 429:
        raise ClaudeAPIError(f"Rate limit exceeded: {error_msg}", 429, error_type)
    elif 500 <= response.status_code < 600:
        raise ClaudeAPIError(f"Server error ({response.status_code}): {error_msg}", 
                           response.status_code, error_type)
    else:
        raise ClaudeAPIError(f"HTTP {response.status_code}: {error_msg}", 
                           response.status_code, error_type)

# 在客户端中使用增强的错误处理
class RobustClaudeClient(ClaudeClient):
    """增强错误处理的客户端"""
    
    @retry_with_backoff(max_retries=3, initial_delay=1.0, backoff_factor=2.0)
    def _make_request(self, endpoint: str, payload: dict) -> dict:
        """重写请求方法，使用增强错误处理"""
        url = f"{self.base_url}{endpoint}"
        
        try:
            response = self.session.post(url, json=payload, timeout=self.timeout)
            
            # 使用新的错误处理器
            return handle_api_response(response)
            
        except Timeout:
            logger.error(f"Request to {url} timed out after {self.timeout}s")
            raise
        except RequestException as e:
            logger.error(f"Request to {url} failed: {e}")
            raise

5.4 提示词工程进阶技巧

基于模板的提示词管理可以进一步优化，支持更复杂的场景。

动态提示词模板与变量验证：

# src/prompts/advanced_templates.py
from string import Template
from typing import Dict, Any, List, Optional
from pydantic import BaseModel, ValidationError, validator
import json

class TemplateVariable(BaseModel):
    """提示词变量定义"""
    name: str
    description: str
    required: bool = True
    default: Optional[Any] = None
    validation: Optional[str] = None  # 如 'email', 'url', 'int', 'float', 'json'

class AdvancedPromptTemplate:
    """高级提示词模板，支持变量验证和类型转换"""
    
    def __init__(self, 
                 name: str,
                 template: str,
                 variables: List[TemplateVariable],
                 system_prompt: Optional[str] = None,
                 description: str = ""):
        self.name = name
        self.template = template
        self.variables = {v.name: v for v in variables}
        self.system_prompt = system_prompt
        self.description = description
    
    def validate_and_render(self, **kwargs) -> Dict[str, Any]:
        """验证变量并渲染模板"""
        # 检查必填变量
        for var_name, var_def in self.variables.items():
            if var_def.required and var_name not in kwargs:
                raise ValueError(f"Missing required variable: {var_name}")
        
        # 应用默认值
        rendered_vars = {}
        for var_name, var_def in self.variables.items():
            value = kwargs.get(var_name, var_def.default)
            
            # 验证变量
            if value is not None and var_def.validation:
                value = self._validate_variable(var_name, value, var_def.validation)
            
            rendered_vars[var_name] = value
        
        # 渲染模板
        template_obj = Template(self.template)
        rendered_prompt = template_obj.safe_substitute(rendered_vars)
        
        # 构建消息
        messages = []
        if self.system_prompt:
            messages.append({"role": "system", "content": self.system_prompt})
        
        messages.append({"role": "user", "content": rendered_prompt})
        
        return {
            "messages": messages,
            "variables": rendered_vars,
            "template_name": self.name
        }
    
    def _validate_variable(self, var_name: str, value: Any, validation: str) -> Any:
        """验证变量值"""
        try:
            if validation == 'json':
                if isinstance(value, str):
                    return json.loads(value)
                return value
            elif validation == 'int':
                return int(value)
            elif validation == 'float':
                return float(value)
            elif validation == 'email':
                if '@' not in str(value):
                    raise ValueError(f"{var_name} must be a valid email")
                return value
            elif validation == 'url':
                if not str(value).startswith(('http://', 'https://')):
                    raise ValueError(f"{var_name} must be a valid URL")
                return value
            else:
                return value
        except (ValueError, TypeError, json.JSONDecodeError) as e:
            raise ValueError(f"Invalid value for {var_name}: {e}")

class AdvancedPromptManager:
    """高级提示词管理器"""
    
    def __init__(self):
        self.templates: Dict[str, AdvancedPromptTemplate] = {}
        self._register_default_templates()
    
    def _register_default_templates(self):
        """注册默认模板"""
        # 代码审查模板
        code_review_vars = [
            TemplateVariable(name="code", description="要审查的代码", required=True),
            TemplateVariable(name="language", description="编程语言", required=True, default="python"),
            TemplateVariable(name="focus_areas", description="审查重点", required=False, 
                           default="readability, performance, security, best practices"),
        ]
        
        self.register_template(
            name="code_review",
            template="""请审查以下${language}代码，重点关注：${focus_areas}

代码：
\`\`\`${language}
${code}
\`\`\`

请提供：
1. 潜在问题列表
2. 改进建议
3. 安全注意事项
4. 总体评分（1-10分）""",
            variables=code_review_vars,
            system_prompt="你是一个经验丰富的软件工程师，擅长代码审查和优化。",
            description="代码审查助手"
        )
        
        # 数据提取模板
        data_extraction_vars = [
            TemplateVariable(name="text", description="源文本", required=True),
            TemplateVariable(name="schema", description="JSON Schema", required=True, validation="json"),
            TemplateVariable(name="examples", description="示例", required=False, validation="json"),
        ]
        
        self.register_template(
            name="data_extraction",
            template="""从以下文本中提取结构化信息，严格遵循提供的JSON Schema格式。

文本：
${text}

JSON Schema：
${schema}

${examples if examples else ''}

请输出一个有效的JSON对象，仅包含提取的数据，不要添加解释。""",
            variables=data_extraction_vars,
            system_prompt="你是一个精准的数据提取专家，能够从文本中提取结构化信息。",
            description="结构化数据提取"
        )
    
    def register_template(self, 
                         name: str,
                         template: str,
                         variables: List[TemplateVariable],
                         system_prompt: Optional[str] = None,
                         description: str = ""):
        """注册新模板"""
        self.templates[name] = AdvancedPromptTemplate(
            name=name,
            template=template,
            variables=variables,
            system_prompt=system_prompt,
            description=description
        )
    
    def render(self, template_name: str, **kwargs) -> Dict[str, Any]:
        """渲染模板"""
        if template_name not in self.templates:
            raise ValueError(f"Template '{template_name}' not found")
        
        return self.templates[template_name].validate_and_render(**kwargs)
    
    def list_templates(self) -> List[Dict[str, Any]]:
        """列出所有可用模板"""
        return [
            {
                "name": name,
                "description": template.description,
                "variables": [vars.dict() for vars in template.variables.values()],
                "has_system_prompt": bool(template.system_prompt)
            }
            for name, t