Claude工具命令实战：如何通过CCR调用自部署的OpenAI模型

1次阅读

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

在实际业务场景中，我们经常会遇到需要同时使用多个 AI 模型服务的需求。比如，一个客服系统可能既需要 OpenAI 的文本生成能力，又需要 Claude 的工具命令功能来处理特定业务逻辑。直接调用多个 API 会导致代码复杂度增加，维护困难。而通过 CCR(Claude Command Relay)集成自部署的 OpenAI 模型，可以带来以下优势：

统一 API 入口，简化客户端代码
可以利用 Claude 强大的工具命令功能来增强 OpenAI 模型的能力
便于实现请求路由、限流、缓存等公共功能
提高系统整体的可观测性和可维护性

优点：
– 简单直接，延迟低
– 完全控制请求和响应流程

缺点：
– 需要自行处理鉴权、限流等问题
– 难以与其他 AI 服务集成
– 缺乏统一的监控和管理

优点：
– 统一的 API 网关，简化客户端
– 内置工具命令支持
– 集中管理鉴权、限流等
– 便于扩展和集成其他服务

缺点：
– 增加了一层网络跳转，略微增加延迟
– 需要额外的配置和维护

首先设置必要的环境变量：

import os
from dotenv import load_dotenv
import jwt
import time
import requests

# 加载环境变量
load_dotenv()

OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
CCR_ENDPOINT = os.getenv('CCR_ENDPOINT', 'https://api.your-ccr-instance.com/v1')
CCR_SECRET = os.getenv('CCR_SECRET')

CCR 使用 JWT 进行鉴权，我们需要生成带签名的 token：

def generate_jwt_token(secret, payload=None, expires_in=3600):
    """
    生成 CCR JWT 鉴权 token
    :param secret: CCR 分配的密钥
    :param payload: 额外 payload
    :param expires_in: token 有效期(秒)
    :return: JWT token 字符串
    """
    if payload is None:
        payload = {}

    # 设置默认 claims
    now = int(time.time())
    payload.update({
        'iat': now,
        'exp': now + expires_in,
        'iss': 'ccr-client'
    })

    return jwt.encode(payload, secret, algorithm='HS256')

下面是构造包含 tool_choice 参数的请求示例：

def call_ccr_with_tools(prompt, tools=None, tool_choice=None):
    """
    通过 CCR 调用 OpenAI 模型，使用 Claude 工具命令
    :param prompt: 用户输入的提示
    :param tools: 工具定义列表
    :param tool_choice: 指定使用的工具
    :return: API 响应
    """
    # 生成 JWT token
    token = generate_jwt_token(CCR_SECRET)

    # 构造请求头
    headers = {'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json'
    }

    # 构造请求体
    payload = {
        'model': 'openai-gpt-4',  # 指定自部署的 OpenAI 模型
        'messages': [{'role': 'user', 'content': prompt}],
        'stream': False  # 非流式响应
    }

    # 添加工具命令相关参数
    if tools:
        payload['tools'] = tools
    if tool_choice:
        payload['tool_choice'] = tool_choice

    try:
        response = requests.post(f'{CCR_ENDPOINT}/chat/completions',
            headers=headers,
            json=payload,
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f'API 请求失败: {e}')
        return None

对于需要处理流式响应的场景：

def call_ccr_stream(prompt, callback):
    """
    处理 CCR 的流式响应
    :param prompt: 用户输入
    :param callback: 处理每个 chunk 的回调函数
    """
    token = generate_jwt_token(CCR_SECRET)
    headers = {'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json',
        'Accept': 'text/event-stream'
    }

    payload = {
        'model': 'openai-gpt-4',
        'messages': [{'role': 'user', 'content': prompt}],
        'stream': True
    }

    try:
        with requests.post(f'{CCR_ENDPOINT}/chat/completions',
            headers=headers,
            json=payload,
            stream=True
        ) as response:
            response.raise_for_status()

            for line in response.iter_lines():
                if line:
                    decoded_line = line.decode('utf-8')
                    if decoded_line.startswith('data:'):
                        chunk = decoded_line[6:]
                        if chunk != '[DONE]':
                            callback(json.loads(chunk))
    except Exception as e:
        print(f'流式请求失败: {e}')

CCR 和 OpenAI 都有请求速率限制，建议实现以下策略：