OpenRouter与Claude API实战：从零构建智能代码生成工具

3次阅读

共计 1880 个字符，预计需要花费 5 分钟才能阅读完成。

Claude 在代码生成领域有三大核心优势：
1. 深度上下文理解：能记住长达 100K token 的对话历史，适合处理复杂代码逻辑
2. 多语言智能切换：自动识别 Python/Java/Go 等主流语言语法和惯用法
3. 结构化输出：天然支持 Markdown 代码块格式，减少后处理工作量

优点：官方原生支持，功能更新及时
缺点：
需自行处理限流和故障转移
全球访问延迟差异大（尤其非北美地区）
账单分散在不同 AI 服务商

优势亮点：
智能路由：自动选择延迟最低的 API 端点
统一计费：单账单管理多个 AI 模型调用
内置重试：默认实现指数退避机制
注意事项：
存在约 5 -10% 的代理开销
新功能支持有 1 - 2 周延迟

# requirements.txt
openrouter>=1.3.0  # 官方 SDK
httpx[http2]>=0.24.0  # 异步 HTTP 客户端
tiktoken>=0.5.1  # Token 计数器

from openrouter import OpenRouter
from tenacity import retry, stop_after_attempt, wait_exponential

class CodeGenerator:
    def __init__(self, api_key):
        self.client = OpenRouter(api_key)

    @retry(stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    async def generate(self, prompt: str, max_tokens=4000) -> str:
        """时间复杂度：O(1) 网络请求不随输入长度变化"""
        response = await self.client.create_completion(
            model="claude-2",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=min(max_tokens, 4096),  # 硬限制
            stream=True
        )
        return await self._process_stream(response)

async def _process_stream(self, response):
    """处理分块数据流，内存效率 O(1)"""
    buffer = []
    async for chunk in response:
        if chunk.choices and chunk.choices[0].delta.content:
            buffer.append(chunk.choices[0].delta.content)
            if len(buffer) > 100:  # 控制内存占用
                yield "".join(buffer)
                buffer.clear()
    if buffer:
        yield "".join(buffer)

请求方式	QPS	平均延迟	成本 / 千次
单条请求	12	850ms	$1.2
批量(8)	68	720ms	$0.9

def sanitize_code(code: str) -> str:
    """替换 API 密钥等敏感信息"""
    patterns = {r"api_key=\"[^\"]+"": "api_key=***",
        r"password: .+": "password: ***"
    }
    for pat, repl in patterns.items():
        code = re.sub(pat, repl, code)
    return code

Prompt 注入防护：
使用 prompt.replace("$", "") 过滤特殊字符
设置 user_id 参数追踪异常请求

速率限制处理：

async def call_with_backoff(self, prompt):
    try:
        return await self.generate(prompt)
    except RateLimitError:
        await asyncio.sleep(random.uniform(1, 5))
        return await self.generate(prompt)