从标准OpenAI兼容接口改为自定义API接口的完整改造过程。
项目：https://github.com/HumanAIGC-Engineering/OpenAvatarChat
file: src/handlers/llm/openai_compatible/llm_handler_openai_compatible.py

1. 导入模块的改动

原版（标准OpenAI）：

from openai import OpenAI

修改后（自定义API）：

import requests
import json

改动原因：

• 原版：使用OpenAI官方SDK，自动处理API调用、认证、流式响应等
• 修改后：使用requests库直接发送HTTP请求，json库解析响应数据
• 为什么：我的API不是标准的OpenAI格式，需要自定义HTTP请求和响应解析

2. 配置类的改动

原版配置：

class LLMConfig(HandlerBaseConfigModel, BaseModel):
    model_name: str = Field(default="qwen-plus")
    system_prompt: str = Field(default="...")
    api_key: str = Field(default=os.getenv("DASHSCOPE_API_KEY"))
    api_url: str = Field(default=None)
    enable_video_input: bool = Field(default=False)

修改后配置：

class LLMConfig(HandlerBaseConfigModel, BaseModel):
    model_name: str = Field(default="general_practitioner")
    system_prompt: str = Field(default="...")
    api_key: str = Field(default="")// 请替换为真实的api_key
    api_url: str = Field(default="")// 请替换为真实的 api_url
    enable_video_input: bool = Field(default=False)

改动原因：

• model_name：从qwen-plus改为general_practitioner，匹配你的API端点
• api_key：从环境变量改为固定值`真实的api_key
• api_url：从None改为具体的API地址
• 为什么：你的API有特定的模型名称和固定的认证信息

3. 客户端初始化的改动

原版（创建OpenAI客户端）：

def create_context(self, session_context, handler_config=None):
    # ... 其他代码 ...
    context.client = OpenAI(
        api_key=context.api_key,
        base_url=context.api_url,
    )
    return context

修改后（移除客户端）：

def create_context(self, session_context, handler_config=None):
    # ... 其他代码 ...
    # 不再创建OpenAI客户端
    return context

改动原因：

• 原版：创建OpenAI客户端对象，用于后续API调用
• 修改后：不再需要客户端，直接使用requests发送HTTP请求
• 为什么：我的API不是OpenAI格式，不需要OpenAI客户端

4. 核心处理逻辑的完全重写

这是最重要的改动，整个handle方法被完全重写：

原版（使用OpenAI SDK）：

def handle(self, context: HandlerContext, inputs: ChatData, output_definitions: Dict[ChatDataType, HandlerDataInfo]):
    # ... 前置处理 ...
    
    # 使用OpenAI客户端发送请求
    completion = context.client.chat.completions.create(
        model=context.model_name,
        messages=[
            context.system_prompt,
        ] + current_content,
        stream=True,
        stream_options={"include_usage": True}
    )
    
    # 处理流式响应
    for chunk in completion:
        if (chunk and chunk.choices and chunk.choices[0] and chunk.choices[0].delta.content):
            output_text = chunk.choices[0].delta.content
            # ... 输出处理 ...

修改后（自定义HTTP请求）：

def handle(self, context: HandlerContext, inputs: ChatData, output_definitions: Dict[ChatDataType, HandlerDataInfo]):
    # ... 前置处理 ...
    
    try:
        # 1. 构建请求数据
        data = {
            "question": chat_text,  # 你的API需要question字段
        }
        headers = {
            'api-key': context.api_key,  # 你的API使用api-key认证
            'Content-Type': 'application/json',
        }
        
        # 2. 发送HTTP POST请求
        response = requests.post(context.api_url, headers=headers, json=data, timeout=30)
        
        # 3. 解析SSE格式响应
        lines = raw_text.strip().split('\n') if raw_text else []
        for line in lines:
            if line.startswith('data: '):  # 你的API返回SSE格式
                data_content = line[6:]  # 去掉'data: '前缀
                # ... 处理每个数据块 ...

5. 请求格式的改动

原版（OpenAI格式）：

# 请求体
{
    "model": "qwen-plus",
    "messages": [
        {"role": "system", "content": "系统提示"},
        {"role": "user", "content": "用户问题"}
    ],
    "stream": True
}

# 认证方式
Authorization: Bearer sk-xxx

修改后（你的API格式）：

# 请求体
{
    "question": "用户问题"  # 简化的请求格式
}

# 认证方式
api-key: xx# 替换成自己的

改动原因：

• 原版：使用OpenAI的复杂消息格式，包含角色、内容等
• 修改后：使用我的API的简单格式，只需要question字段
• 为什么：我的API设计更简单，不需要复杂的对话历史管理

6. 响应解析的改动

原版（OpenAI流式响应）：

for chunk in completion:
    if chunk.choices[0].delta.content:
        output_text = chunk.choices[0].delta.content
        # 直接输出内容

修改后（SSE格式解析）：

# 解析SSE格式：data: {"result": "内容"}
lines = raw_text.strip().split('\n')
for line in lines:
    if line.startswith('data: '):
        data_content = line[6:]  # 去掉'data: '前缀
        try:
            json_data = json.loads(data_content)
            # 处理JSON数据
        except json.JSONDecodeError:
            # 处理纯文本数据

改动原因：

• 原版：OpenAI SDK自动解析流式响应
• 修改后：手动解析SSE（Server-Sent Events）格式
• 为什么：我的API返回data: 格式的流式数据，需要手动解析

7. 错误处理的改动

原版（简单错误处理）：

# 基本没有错误处理，依赖OpenAI SDK

修改后（详细错误处理）：

# 1. HTTP状态码检查
if response.status_code != 200:
    fallback_text = "抱歉，我这会儿有点忙，稍后再帮您详细解答。"
    # 返回友好错误信息

# 2. API错误负载检查
if (status_val is not None and status_val >= 400) or (json_data.get('message') and not any(k in json_data for k in ['result','answer','content','text'])):
    fallback_text = "抱歉，我这会儿有点忙，稍后再帮您详细解答。"
    # 处理API返回的错误

# 3. 异常捕获
except Exception as e:
    logger.error(f'API调用异常: {str(e)}')
    # 返回兜底信息

改动原因：

• 原版：OpenAI SDK自动处理大部分错误
• 修改后：需要手动处理各种错误情况
• 为什么：我的API可能返回各种错误格式，需要友好地处理

8. 日志记录的改动

原版（简单日志）：

logger.info(f'llm input {context.model_name} {chat_text} ')
logger.info(output_text)

修改后（详细日志）：

logger.info(f'llm input {context.model_name} {chat_text} ')
logger.debug(f'发送请求到 {context.api_url}')
logger.debug(f'请求头: {headers}')
logger.debug(f'请求体: {data}')
logger.debug(f'HTTP {response.status_code}, 原始响应预览: {raw_text[:500]}...')
logger.info(f'parsed reply (chunk): {output_text}')
logger.info(f'final reply: {context.output_texts}')

改动原因：

• 原版：OpenAI SDK内部处理，日志较少
• 修改后：需要详细记录每个步骤，便于调试
• 为什么：自定义API调用需要更多调试信息

总结

这次改动的核心思想是：从使用现成的SDK改为直接控制HTTP请求。

为什么需要这样改动？

1. API格式不同：API不是标准的OpenAI格式
2. 认证方式不同：使用api-key而不是Authorization: Bearer
3. 请求格式不同：只需要question字段，不需要复杂的消息数组
4. 响应格式不同：返回SSE格式的data: 数据流
5. 错误处理不同：需要处理各种自定义错误格式

对初学者的建议：

1. 理解HTTP基础：学会使用requests库发送HTTP请求
2. 理解JSON解析：学会解析和生成JSON数据
3. 理解流式响应：学会处理SSE格式的数据流
4. 理解错误处理：学会优雅地处理各种错误情况
5. 理解日志记录：学会使用日志来调试问题

这种改动虽然复杂，但让我完全控制了API调用的每个细节，可以适应任何自定义的API格式。