knowledge/coze_client.py

# coze_client.py
import requests
import json
import logging

class CozeClient:
    """
    封装Coze API的聊天完成操作。
    """
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.api_url = "https://api.coze.com/v1/chat/completions" # Coze API 聊天完成端点

    def send_message(self, bot_id: str, prompt_template: str, input_data: str) -> str | None:
        """
        向Coze机器人发送消息并获取回复。
        Args:
            bot_id (str): Coze 机器人的 Bot ID。
            prompt_template (str): Coze API的提示模板字符串，需要包含 '{input_data}' 占位符。
            input_data (str): 实际要填充到模板中的输入数据。
        Returns:
            str | None: Coze的回复内容，如果失败则返回 None。
        Raises:
            Exception: 如果Coze API调用失败或返回非预期格式。
        """
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }
        
        # 填充提示模板
        user_message_content = prompt_template.format(input_data=input_data)

        payload = {
            "model": bot_id, # 在Coze API中，bot_id 通常作为 model 参数传入
            "messages": [
                {"role": "user", "content": user_message_content}
            ],
            "stream": False # 非流式响应
        }

        try:
            logging.info(f"正在调用 Coze Bot '{bot_id}' (输入内容前50字: '{user_message_content[:50]}...')")
            response = requests.post(self.api_url, headers=headers, json=payload, timeout=60)
            response.raise_for_status() # 检查HTTP响应状态码
            result = response.json()

            # 检查Coze响应结构，提取内容
            if result.get("choices") and result["choices"][0].get("message") and result["choices"][0]["message"].get("content"):
                coze_output_content = result["choices"][0]["message"]["content"]
                logging.info(f"Coze API 调用成功，返回内容长度: {len(coze_output_content)}")
                return coze_output_content
            else:
                logging.warning(f"Coze API 未返回有效内容。原始响应: {json.dumps(result, ensure_ascii=False, indent=2)}")
                raise Exception(f"Coze API returned no valid content: {result}")

        except requests.exceptions.RequestException as e:
            logging.error(f"请求Coze API发生网络错误: {e}")
            raise Exception(f"Coze API Request Failed: {e}")
        except (IndexError, KeyError) as e:
            logging.error(f"Coze API 响应格式错误: {e}. 原始响应: {json.dumps(result, ensure_ascii=False, indent=2) if 'result' in locals() else 'N/A'}")
            raise Exception(f"Coze API Response Format Error: {e}")
        except Exception as e:
            logging.error(f"处理Coze API响应失败: {e}")
            raise Exception(f"Coze API Response Handling Failed: {e}")

# 可以在这里添加一些简单的测试代码，但通常在main.py中进行集成测试
if __name__ == '__main__':
    # 仅作示例，请勿在生产环境直接硬编码敏感信息
    # 配置日志，用于独立测试
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

    print("--- Coze客户端独立测试 (请确保替换YOUR_...) ---")
    TEST_COZE_API_KEY = "YOUR_COZE_API_KEY"
    TEST_COZE_BOT_ID = "YOUR_COZE_BOT_ID"
    TEST_PROMPT_TEMPLATE = "请用一句话总结以下内容: {input_data}"
    TEST_INPUT_TEXT = "Python是一种高级编程语言，由Guido van Rossum创建，并于1991年首次发布。它以其清晰的语法和广泛的库而闻名，适用于Web开发、数据分析、人工智能等多个领域。"

    if "YOUR_" in TEST_COZE_API_KEY:
        logging.warning("请替换 coze_client.py 中的 YOUR_ 占位符为您的实际Coze API信息以运行测试。")
    else:
        try:
            coze_client = CozeClient(TEST_COZE_API_KEY)
            coze_output = coze_client.send_message(
                TEST_COZE_BOT_ID,
                TEST_PROMPT_TEMPLATE,
                TEST_INPUT_TEXT
            )
            if coze_output:
                print(f"\nCoze API 成功响应:\n{coze_output}")
            else:
                print("\nCoze API 未返回有效内容。")
        except Exception as e:
            logging.error(f"Coze客户端独立测试失败: {e}")
    print("--- Coze客户端独立测试结束 ---")