别再为API整合头疼了！QuantumNous/new-api：一套代码搞定所有大模型调用的终极方案

git clone https://github.com/QuantumNous/new-api.git
cd new-api

python -m venv venv

# Windows系统激活虚拟环境
venv\Scripts\activate

# macOS和Linux系统激活虚拟环境
source venv/bin/activate

pip install -r requirements.txt

pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

cp config.example.yaml config.yaml

# AI服务商配置区域
providers:
  # OpenAI配置
  openai:
    api_key: "sk-your-openai-key-here"
    base_url: "https://api.openai.com/v1"
    model: "gpt-4"

  # Anthropic Claude配置
  anthropic:
    api_key: "sk-ant-your-claude-key-here"
    base_url: "https://api.anthropic.com"
    model: "claude-3-sonnet-20240229"

  # 通义千问配置
  qwen:
    api_key: "your-qwen-api-key"
    base_url: "https://dashscope.aliyuncs.com/compatible-mode/v1"
    model: "qwen-turbo"

  # 智谱AI配置
  zhipu:
    api_key: "your-zhipu-api-key"
    base_url: "https://open.bigmodel.cn/api/paas/v4"
    model: "glm-4"

# 服务端口配置
server:
  host: "0.0.0.0"
  port: 8000

你的应用 
    ↓ 发送统一格式请求
new-api服务器
    ↓ 识别目标模型
    ↓ 转换请求格式
    ↓ 添加认证信息
对应的AI服务商API
    ↓ 返回响应
new-api服务器
    ↓ 统一响应格式
你的应用

{
  "model": "gpt-4",
  "messages": [
    {
      "role": "system",
      "content": "你是一个专业的Python编程助手"
    },
    {
      "role": "user", 
      "content": "请解释什么是装饰器模式"
    }
  ],
  "temperature": 0.7,
  "max_tokens": 1000
}

import requests
import json

url = "http://localhost:8000/v1/chat/completions"

payload = {
    "model": "gpt-4",
    "messages": [{"role": "user", "content": "写一首关于春天的诗"}],
    "stream": True
}

headers = {"Content-Type": "application/json"}

response = requests.post(
    url, 
    headers=headers,
    json=payload,
    stream=True
)

for line in response.iter_lines():
    if line:
        # 解析SSE格式的数据
        decoded = line.decode('utf-8')
        if decoded.startswith('data: '):
            data = decoded[6:]  # 去掉"data: "前缀
            if data != '[DONE]':
                chunk = json.loads(data)
                content = chunk['choices'][0]['delta'].get('content', '')
                print(content, end='', flush=True)

ai-chat/
├── main.py          # 主程序入口
├── config.yaml       # 配置文件
└── requirements.txt # 依赖列表

"""
AI命令行对话工具
支持切换不同的AI模型进行对话
"""

import requests
import json
import sys

# 配置信息
API_BASE = "http://localhost:8000/v1/chat/completions"
DEFAULT_MODEL = "gpt-4"

def chat(model, message, history=None):
    """
    发送对话请求到new-api

    参数:
        model: 要使用的模型名称
        message: 用户输入的消息
        history: 对话历史记录列表

    返回:
        AI的回复内容
    """
    # 构建消息历史
    messages = []

    # 添加系统提示
    messages.append({
        "role": "system",
        "content": "你是一个有帮助的AI助手，请用简洁清晰的语言回答问题。"
    })

    # 添加历史对话
    if history:
        messages.extend(history)

    # 添加当前用户消息
    messages.append({
        "role": "user",
        "content": message
    })

    # 构造请求
    payload = {
        "model": model,
        "messages": messages,
        "temperature": 0.7,
        "max_tokens": 2000
    }

    headers = {
        "Content-Type": "application/json"
    }

    try:
        response = requests.post(API_BASE, headers=headers, json=payload, timeout=60)
        response.raise_for_status()

        result = response.json()
        assistant_message = result['choices'][0]['message']['content']

        return assistant_message, messages + [{"role": "assistant", "content": assistant_message}]

    except requests.exceptions.Timeout:
        return "请求超时，请检查网络连接或API配置。", history
    except requests.exceptions.RequestException as e:
        return f"请求失败: {str(e)}", history
    except json.JSONDecodeError:
        return "解析响应失败，请检查API返回格式。", history

def print_welcome():
    """打印欢迎信息"""
    print("=" * 50)
    print("       AI 命令行对话工具")
    print("=" * 50)
    print("支持的模型: gpt-4, claude-3-sonnet, qwen-plus, glm-4")
    print("输入消息开始对话，输入 'quit' 退出")
    print("输入 'model <名称>' 切换模型")
    print("=" * 50)
    print()

def main():
    """主程序入口"""
    print_welcome()

    current_model = DEFAULT_MODEL
    history = None

    while True:
        try:
            user_input = input(f"[{current_model}] 你: ").strip()
        except KeyboardInterrupt:
            print("\n\n再见!")
            break

        if not user_input:
            continue

        # 处理特殊命令
        if user_input.lower() in ['quit', 'exit', 'q']:
            print("再见!")
            break

        if user_input.lower().startswith('model '):
            new_model = user_input[6:].strip()
            current_model = new_model
            print(f"已切换到模型: {current_model}")
            continue

        if user_input.lower() == 'clear':
            history = None
            print("对话历史已清除")
            continue

        # 发送对话请求
        print(f"[{current_model}] AI: ", end='', flush=True)
        response, history = chat(current_model, user_input, history)
        print(response)
        print()

if __name__ == "__main__":
    main()

# 在项目目录下启动服务
python -m new_api.server

python main.py

==================================================
       AI 命令行对话工具
==================================================
支持的模型: gpt-4, claude-3-sonnet, qwen-plus, glm-4
输入消息开始对话，输入 'quit' 退出
输入 'model <名称>' 切换模型
==================================================

[gpt-4] 你: 你好，请用一句话介绍Python

[gpt-4] AI: Python是一门简洁优雅、极易上手的高级编程语言，特别适合初学者入门和快速开发各种应用。

[gpt-4] 你: model glm-4

已切换到模型: glm-4

[glm-4] 你: 解释一下什么是闭包

[glm-4] AI: 闭包（Closure）是Python中一种特殊的函数对象，它能够记住创建时的环境上下文...

"""
多模态AI聊天机器人
支持文字和图片输入
"""

import base64
import requests
from pathlib import Path
from datetime import datetime

class MultimodalChatBot:
    """
    多模态聊天机器人类
    支持文字和图片的混合输入
    """

    def __init__(self, api_base="http://localhost:8000/v1"):
        self.api_base = api_base
        self.conversation_history = [
            {
                "role": "system",
                "content": """你是一个智能助手，能够分析用户发送的图片并回答相关问题。
请用详细但简洁的语言描述图片内容，并准确回答用户的问题。"""
            }
        ]

    def encode_image_to_base64(self, image_path):
        """
        将图片文件转换为base64编码

        参数:
            image_path: 图片文件的路径

        返回:
            base64编码的字符串
        """
        path = Path(image_path)

        if not path.exists():
            raise FileNotFoundError(f"图片文件不存在: {image_path}")

        # 支持多种图片格式
        allowed_formats = {'.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp'}
        if path.suffix.lower() not in allowed_formats:
            raise ValueError(f"不支持的图片格式: {path.suffix}")

        with open(path, 'rb') as image_file:
            encoded = base64.b64encode(image_file.read()).decode('utf-8')

        return encoded

    def build_message_content(self, text, image_paths=None):
        """
        构建消息内容，支持文字和图片的混合

        参数:
            text: 文字内容
            image_paths: 图片路径列表（可选）

        返回:
            格式化的消息内容
        """
        content = []

        # 添加图片（如果有）
        if image_paths:
            for path in image_paths:
                base64_image = self.encode_image_to_base64(path)
                content.append({
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/jpeg;base64,{base64_image}"
                    }
                })

        # 添加文字
        content.append({
            "type": "text",
            "text": text
        })

        return content

    def chat(self, text, image_paths=None):
        """
        发送对话请求

        参数:
            text: 用户输入的文字
            image_paths: 要分析的图片路径列表

        返回:
            AI的回复
        """
        # 构建当前用户消息
        user_message = {
            "role": "user",
            "content": self.build_message_content(text, image_paths)
        }

        # 添加到对话历史
        self.conversation_history.append(user_message)

        # 构造API请求
        payload = {
            "model": "gpt-4-vision-preview",  # 使用支持视觉的模型
            "messages": self.conversation_history,
            "max_tokens": 2000
        }

        headers = {
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(
                f"{self.api_base}/chat/completions",
                headers=headers,
                json=payload,
                timeout=120
            )
            response.raise_for_status()

            result = response.json()
            assistant_message = result['choices'][0]['message']

            # 将AI回复添加到历史
            self.conversation_history.append(assistant_message)

            return assistant_message['content']

        except requests.exceptions.RequestException as e:
            return f"请求出错: {str(e)}"

    def clear_history(self):
        """清除对话历史"""
        self.conversation_history = [
            {
                "role": "system", 
                "content": """你是一个智能助手，能够分析用户发送的图片并回答相关问题。
请用详细但简洁的语言描述图片内容，并准确回答用户的问题。"""
            }
        ]
        print("对话历史已清除")

    def show_history(self):
        """显示对话历史"""
        print("\n" + "=" * 60)
        print("对话历史:")
        print("=" * 60)

        for i, msg in enumerate(self.conversation_history):
            role = msg['role']
            content = msg['content']

            if isinstance(content, list):
                # 多模态消息
                texts = [c['text'] for c in content if c['type'] == 'text']
                images = [c['image_url']['url'][:50] + '...' 
                         for c in content if c['type'] == 'image_url']
                content = texts[0] if texts else "[含图片]"
                if images:
                    content += f"\n[含 {len(images)} 张图片]"

            print(f"\n[{role.upper()}]")
            print(content[:500] + "..." if len(str(content)) > 500 else content)

        print("\n" + "=" * 60)

def interactive_demo():
    """
    交互式演示函数
    展示机器人的各种功能
    """
    bot = MultimodalChatBot()

    print("=" * 60)
    print("多模态AI聊天机器人")
    print("=" * 60)
    print("功能说明:")
    print("  - 直接输入文字进行对话")
    print("  - 输入 'image:图片路径' 可以同时发送图片")
    print("  - 输入 'clear' 清除对话历史")
    print("  - 输入 'history' 查看对话历史")
    print("  - 输入 'quit' 退出程序")
    print("=" * 60)
    print()

    while True:
        try:
            user_input = input("你: ").strip()
        except KeyboardInterrupt:
            print("\n\n再见!")
            break

        if not user_input:
            continue

        # 处理特殊命令
        if user_input.lower() in ['quit', 'exit']:
            print("再见!")
            break

        if user_input.lower() == 'clear':
            bot.clear_history()
            continue

        if user_input.lower() == 'history':
            bot.show_history()
            continue

        # 解析图片指令
        text = user_input
        image_paths = None

        if user_input.lower().startswith('image:'):
            parts = user_input.split(':', 1)
            image_paths = [parts[1].strip()]
            text = "请分析这张图片"

        # 发送请求并获取回复
        print("AI: ", end='', flush=True)
        response = bot.chat(text, image_paths)
        print(response)
        print()

def demo_with_sample_image():
    """
    使用示例图片进行演示
    这个函数展示了如何用代码直接调用图片分析功能
    """
    bot = MultimodalChatBot()

    # 假设我们有一张图片
    sample_image = "sample_image.jpg"

    # 检查图片是否存在
    if not Path(sample_image).exists():
        print(f"示例图片不存在: {sample_image}")
        print("跳过图片分析演示")
        return

    questions = [
        "这张图片里有什么？",
        "描述一下图片的主要内容",
        "图片中的文字内容是什么？"
    ]

    for question in questions:
        print(f"\n问题: {question}")
        response = bot.chat(question, [sample_image])
        print(f"回答: {response}")

if __name__ == "__main__":
    import sys

    if len(sys.argv) > 1 and sys.argv[1] == '--demo':
        # 运行演示模式
        demo_with_sample_image()
    else:
        # 运行交互模式
        interactive_demo()

# 交互模式
python multimodal_bot.py

# 演示模式（需要准备示例图片）
python multimodal_bot.py --demo

"""
企业知识库问答系统
基于new-api构建，支持文档检索和智能问答
"""

import os
import json
import requests
from pathlib import Path
from typing import List, Dict, Tuple
from collections import defaultdict

class DocumentChunker:
    """
    文档分块处理器
    将长文档分割成适合处理的较小块
    """

    def __init__(self, chunk_size=500, overlap=50):
        """
        初始化分块器

        参数:
            chunk_size: 每个块的理想大小（字符数）
            overlap: 相邻块之间的重叠字符数
        """
        self.chunk_size = chunk_size
        self.overlap = overlap

    def chunk_text(self, text: str, source: str = "") -> List[Dict]:
        """
        将文本分割成块

        参数:
            text: 要分割的文本
            source: 文本来源（文件名或URL）

        返回:
            包含文本块的字典列表
        """
        chunks = []
        start = 0
        text_length = len(text)

        while start < text_length:
            end = start + self.chunk_size

            # 确保在句子边界分割
            if end < text_length:
                # 向前查找最近的句号或换行符
                for i in range(end, max(start, end - 100), -1):
                    if text[i] in '。！？.!?\n':
                        end = i + 1
                        break

            chunk_text = text[start:end].strip()

            if chunk_text:
                chunks.append({
                    "content": chunk_text,
                    "source": source,
                    "start_pos": start,
                    "end_pos": end
                })

            start = end - self.overlap

        return chunks

    def chunk_file(self, file_path: str) -> List[Dict]:
        """
        读取文件并分块

        参数:
            file_path: 文件路径

        返回:
            分块后的字典列表
        """
        path = Path(file_path)

        if not path.exists():
            raise FileNotFoundError(f"文件不存在: {file_path}")

        # 根据文件类型选择读取方式
        if path.suffix == '.txt':
            with open(path, 'r', encoding='utf-8') as f:
                text = f.read()
        elif path.suffix == '.md':
            with open(path, 'r', encoding='utf-8') as f:
                text = f.read()
        else:
            raise ValueError(f"不支持的文件类型: {path.suffix}")

        return self.chunk_text(text, source=str(path))

class SimpleVectorStore:
    """
    简化的向量存储（基于关键词匹配）

    注意：生产环境中建议使用真正的向量数据库
    如Milvus、Pinecone或ChromaDB
    """

    def __init__(self):
        """初始化向量存储"""
        self.chunks = []
        self.index = defaultdict(list)

    def add_chunks(self, chunks: List[Dict]):
        """
        添加文档块到存储

        参数:
            chunks: 文档块列表
        """
        for i, chunk in enumerate(chunks):
            # 为每个chunk分配ID
            chunk['id'] = len(self.chunks) + i
            self.chunks.append(chunk)

            # 建立简单的关键词索引
            words = self._extract_keywords(chunk['content'])
            for word in words:
                self.index[word].append(chunk['id'])

    def _extract_keywords(self, text: str) -> List[str]:
        """
        从文本中提取关键词

        参数:
            text: 输入文本

        返回:
            关键词列表
        """
        # 简单的中英文分词
        # 实际应用中应使用更专业的分词库
        import re

        # 提取英文单词
        english_words = re.findall(r'[a-zA-Z]+', text.lower())

        # 提取中文词语（简单实现）
        chinese_chars = re.findall(r'[\u4e00-\u9fff]+', text)

        # 过滤停用词
        stop_words = {'的', '了', '在', '是', '我', '有', '和', '就', '不', '人',
                      'the', 'a', 'an', 'is', 'are', 'was', 'were', 'be', 'been'}

        keywords = []
        keywords.extend([w for w in english_words if len(w) > 2 and w not in stop_words])
        keywords.extend([w for w in chinese_chars if len(w) > 1])

        return keywords

    def search(self, query: str, top_k: int = 5) -> List[Dict]:
        """
        搜索最相关的文档块

        参数:
            query: 查询文本
            top_k: 返回的结果数量

        返回:
            相关文档块列表
        """
        # 计算每个chunk的得分
        query_keywords = set(self._extract_keywords(query))

        scores = defaultdict(float)
        for keyword in query_keywords:
            matching_ids = self.index.get(keyword, [])
            for chunk_id in matching_ids:
                # 精确匹配的关键词得分更高
                if keyword in self.chunks[chunk_id]['content']:
                    scores[chunk_id] += 2.0
                else:
                    scores[chunk_id] += 1.0

        # 按得分排序
        sorted_ids = sorted(scores.items(), key=lambda x: x[1], reverse=True)

        # 返回top_k个结果
        results = []
        for chunk_id, score in sorted_ids[:top_k]:
            result = self.chunks[chunk_id].copy()
            result['score'] = score
            results.append(result)

        return results

class KnowledgeBaseQA:
    """
    知识库问答系统

    整合文档处理、检索和AI问答功能
    """

    def __init__(self, api_base="http://localhost:8000/v1"):
        """
        初始化问答系统

        参数:
            api_base: new-api服务地址
        """
        self.api_base = api_base
        self.chunker = DocumentChunker(chunk_size=500)
        self.vector_store = SimpleVectorStore()
        self.conversation_history = []

    def load_documents(self, folder_path: str):
        """
        加载文件夹中的所有文档

        参数:
            folder_path: 文档文件夹路径
        """
        path = Path(folder_path)

        if not path.exists():
            raise FileNotFoundError(f"文件夹不存在: {folder_path}")

        all_chunks = []
        supported_formats = {'.txt', '.md'}

        for file_path in path.rglob('*'):
            if file_path.suffix in supported_formats:
                try:
                    chunks = self.chunker.chunk_file(str(file_path))
                    all_chunks.extend(chunks)
                    print(f"已加载: {file_path} ({len(chunks)} 个块)")
                except Exception as e:
                    print(f"加载失败 {file_path}: {e}")

        if all_chunks:
            self.vector_store.add_chunks(all_chunks)
            print(f"\n总计加载了 {len(all_chunks)} 个文档块")
        else:
            print("未找到任何文档")

    def answer(self, question: str, use_context: bool = True) -> str:
        """
        基于知识库回答问题

        参数:
            question: 用户问题
            use_context: 是否使用检索到的上下文

        返回:
            AI的回答
        """
        # 检索相关文档
        relevant_chunks = self.vector_store.search(question, top_k=3)

        # 构建系统提示
        if use_context and relevant_chunks:
            context_text = "\n\n---\n\n".join([
                f"[来源: {chunk['source']}]\n{chunk['content']}"
                for chunk in relevant_chunks
            ])

            system_prompt = f"""你是一个基于企业知识库的专业助手。
请根据以下参考资料回答用户的问题。如果参考资料中没有相关信息，
请说明你无法从知识库中找到答案。

参考资料:
{context_text}

回答要求:
1. 引用答案的来源
2. 用清晰简洁的语言表达
3. 如果资料不足，如实说明"""
        else:
            system_prompt = "你是一个专业的AI助手，请回答用户的问题。"

        # 构建消息
        messages = [
            {"role": "system", "content": system_prompt}
        ]

        # 添加对话历史（最近3轮）
        messages.extend(self.conversation_history[-6:])

        # 添加当前问题
        messages.append({"role": "user", "content": question})

        # 调用API
        payload = {
            "model": "gpt-4",
            "messages": messages,
            "temperature": 0.3,  # 较低的temperature以获得更稳定的回答
            "max_tokens": 1500
        }

        headers = {"Content-Type": "application/json"}

        try:
            response = requests.post(
                f"{self.api_base}/chat/completions",
                headers=headers,
                json=payload,
                timeout=60
            )
            response.raise_for_status()

            result = response.json()
            answer = result['choices'][0]['message']['content']

            # 更新对话历史
            self.conversation_history.append(
                {"role": "user", "content": question}
            )
            self.conversation_history.append(
                {"role": "assistant", "content": answer}
            )

            return answer, relevant_chunks

        except requests.exceptions.RequestException as e:
            return f"查询失败: {str(e)}", []

    def interactive_mode(self):
        """
        交互式问答模式
        """
        print("=" * 60)
        print("企业知识库问答系统")
        print("=" * 60)
        print("输入问题进行查询，输入 'quit' 退出")
        print("=" * 60)
        print()

        while True:
            try:
                question = input("问题: ").strip()
            except KeyboardInterrupt:
                print("\n\n再见!")
                break

            if not question:
                continue

            if question.lower() in ['quit', 'exit']:
                print("再见!")
                break

            print("\n正在查询知识库...")
            answer, sources = self.answer(question)

            print("\n回答:")
            print("-" * 40)
            print(answer)

            if sources:
                print("\n参考来源:")
                for i, source in enumerate(sources, 1):
                    print(f"  {i}. {source['source']}")

            print()

def setup_demo_knowledge_base(qa_system):
    """
    设置演示用的知识库

    创建一些示例文档用于演示
    """
    demo_folder = Path("demo_knowledge_base")
    demo_folder.mkdir(exist_ok=True)

    # 创建示例文档
    documents = [
        ("公司介绍.md", """
# 公司介绍

我们是一家专注于人工智能技术的科技公司。

## 公司愿景

成为全球领先的AI解决方案提供商。

## 核心价值观

1. 创新：持续探索前沿技术
2. 客户至上：以为客户创造价值为核心
3. 开放协作：与合作伙伴共同成长

## 发展历程

- 2018年：公司成立
- 2020年：完成A轮融资
- 2022年：发布核心产品
- 2024年：用户突破100万
        """),
        ("产品手册.txt", """
产品技术规格

产品名称：AI智能助手
版本：3.0

主要功能：
1. 智能问答：支持多轮对话和上下文理解
2. 文档分析：支持PDF、Word等格式文档解析
3. 代码生成：支持多种编程语言的代码生成
4. 数据可视化：自动生成图表和数据报告

技术指标：
- 响应时间：< 1秒
- 准确率：95%以上
- 支持语言：中文、英文、日文等20种语言

联系方式：
电话：400-123-4567
邮箱：support@example.com
        """),
        ("常见问题.txt", """
常见问题解答

Q: 如何注册账号？
A: 访问官网www.example.com，点击注册按钮，填写手机号和验证码即可完成注册。

Q: 免费额度用完了怎么办？
A: 可以升级到付费套餐，也可以等待下个月额度重置。详细价格请查看官网定价页面。

Q: 支持私有化部署吗？
A: 支持。我们提供标准版和企业版两种部署方案，企业版支持完全私有化部署。

Q: API调用有频率限制吗？
A: 标准版每秒最多10次请求，企业版无限制。

Q: 如何获取技术支持？
A: 可以提交工单，也可以加入官方技术交流群。
        """)
    ]

    # 写入文档
    for filename, content in documents:
        file_path = demo_folder / filename
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(content.strip())

    print("演示知识库创建完成！")
    print(f"文档位置: {demo_folder.absolute()}")
    print()

    # 加载文档
    qa_system.load_documents(str(demo_folder))

if __name__ == "__main__":
    import sys

    qa = KnowledgeBaseQA()

    if len(sys.argv) > 1 and sys.argv[1] == '--demo':
        # 演示模式
        setup_demo_knowledge_base(qa)
        print("\n开始演示问答...\n")

        demo_questions = [
            "你们公司是做什么的？",
            "产品支持哪些语言？",
            "怎么注册账号？",
            "支持私有化部署吗？"
        ]

        for question in demo_questions:
            print(f"问题: {question}")
            answer, sources = qa.answer(question)
            print(f"回答: {answer}")
            if sources:
                print(f"来源: {sources[0]['source']}")
            print()

    else:
        # 交互模式
        if Path("demo_knowledge_base").exists():
            qa.load_documents("demo_knowledge_base")
        else:
            print("未找到知识库文件夹，正在创建演示数据...")
            setup_demo_knowledge_base(qa)

        qa.interactive_mode()

"""
AI模型选择器示例
用户可以自由切换不同的AI模型
"""

import requests
import json

API_BASE = "http://localhost:8000/v1"

# 可用的模型列表
AVAILABLE_MODELS = {
    "1": ("gpt-4", "OpenAI GPT-4", "通用能力最强"),
    "2": ("claude-3-sonnet", "Anthropic Claude-3", "长文本理解好"),
    "3": ("qwen-plus", "通义千问Plus", "中文优化"),
    "4": ("glm-4", "智谱GLM-4", "国产优秀模型"),
}

def display_model_menu():
    """显示模型选择菜单"""
    print("\n可用的AI模型:")
    print("-" * 40)
    for key, (model_id, name, desc) in AVAILABLE_MODELS.items():
        print(f"  {key}. {name} - {desc}")
    print("-" * 40)

def chat_with_model(model_id, message):
    """使用指定模型进行对话"""
    payload = {
        "model": model_id,
        "messages": [{"role": "user", "content": message}],
        "temperature": 0.7
    }

    response = requests.post(
        f"{API_BASE}/chat/completions",
        json=payload,
        timeout=60
    )

    result = response.json()
    return result['choices'][0]['message']['content']

def main():
    """主程序"""
    current_model_key = "1"

    print("=" * 50)
    print("AI 模型选择器")
    print("=" * 50)

    while True:
        display_model_menu()
        print(f"\n当前模型: {AVAILABLE_MODELS[current_model_key][1]}")
        print("\n输入数字切换模型，输入消息进行对话，输入 quit 退出\n")

        user_input = input("请输入: ").strip()

        if user_input.lower() in ['quit', 'exit']:
            break

        if user_input in AVAILABLE_MODELS:
            current_model_key = user_input
            print(f"已切换到: {AVAILABLE_MODELS[current_model_key][1]}")
            continue

        # 发送对话请求
        model_id = AVAILABLE_MODELS[current_model_key][0]
        print(f"\n正在使用 {AVAILABLE_MODELS[current_model_key][1]} 处理...")

        try:
            response = chat_with_model(model_id, user_input)
            print(f"\n回复:\n{response}")
        except Exception as e:
            print(f"出错: {e}")

if __name__ == "__main__":
    main()

"""
模型性能对比工具
对同一问题测试多个模型的回答
"""

import requests
import json
import time
from datetime import datetime

API_BASE = "http://localhost:8000/v1"

# 要对比的模型列表
MODELS_TO_TEST = [
    "gpt-4",
    "gpt-3.5-turbo",
    "claude-3-sonnet",
    "qwen-plus",
    "glm-4"
]

def test_model(model_id, prompt, timeout=60):
    """
    测试单个模型的响应

    返回: (响应内容, 响应时间, 是否成功)
    """
    payload = {
        "model": model_id,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.7,
        "max_tokens": 1000
    }

    start_time = time.time()

    try:
        response = requests.post(
            f"{API_BASE}/chat/completions",
            json=payload,
            timeout=timeout
        )
        elapsed = time.time() - start_time

        if response.status_code == 200:
            result = response.json()
            content = result['choices'][0]['message']['content']
            return content, elapsed, True
        else:
            return f"错误码: {response.status_code}", elapsed, False

    except Exception as e:
        elapsed = time.time() - start_time
        return f"异常: {str(e)}", elapsed, False

def compare_models(test_prompt, models=None):
    """
    对比多个模型的响应

    参数:
        test_prompt: 测试用的提示词
        models: 要测试的模型列表（None表示使用默认列表）

    返回: 测试结果列表
    """
    if models is None:
        models = MODELS_TO_TEST

    print("=" * 70)
    print("模型对比测试")
    print("=" * 70)
    print(f"\n测试问题: {test_prompt}\n")

    results = []

    for model in models:
        print(f"测试 {model}...", end=" ", flush=True)

        response, elapsed, success = test_model(model, test_prompt)

        results.append({
            "model": model,
            "response": response,
            "elapsed": elapsed,
            "success": success
        })

        status = "成功" if success else "失败"
        print(f"[{status}] {elapsed:.2f}秒")

        # 添加延迟避免API限流
        time.sleep(1)

    return results

def display_comparison(results):
    """显示对比结果"""
    print("\n" + "=" * 70)
    print("对比结果汇总")
    print("=" * 70)

    for i, result in enumerate(results, 1):
        status = "✓" if result['success'] else "✗"
        print(f"\n{status} 模型 {i}: {result['model']}")
        print(f"   耗时: {result['elapsed']:.2f} 秒")

        if result['success']:
            # 显示响应的前500个字符
            preview = result['response'][:500]
            if len(result['response']) > 500:
                preview += "..."
            print(f"   响应:\n   {preview}")
        else:
            print(f"   错误: {result['response']}")

def main():
    """主程序"""
    print("模型性能对比工具")
    print("输入要测试的问题，多个问题用|分隔")
    print("输入 quit 退出\n")

    test_prompts = [
        "解释一下什么是机器学习",
        "用Python写一个快速排序算法",
        "推荐5本科适合初学者的编程书籍"
    ]

    for prompt in test_prompts:
        results = compare_models(prompt)
        display_comparison(results)
        print()

if __name__ == "__main__":
    main()

"""
批量内容生成工具
使用AI批量生成各类内容
"""

import requests
import json
import time
from typing import List, Dict
from concurrent.futures import ThreadPoolExecutor, as_completed

API_BASE = "http://localhost:8000/v1"

class BatchContentGenerator:
    """
    批量内容生成器
    支持多种内容类型的批量生成
    """

    def __init__(self, model="gpt-4"):
        self.model = model
        self.api_base = API_BASE

    def generate_single(self, prompt: str, system_prompt: str = None) -> Dict:
        """
        生成单个内容

        参数:
            prompt: 用户提示
            system_prompt: 系统提示（可选）

        返回:
            包含成功状态和内容的字典
        """
        messages = []

        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})

        messages.append({"role": "user", "content": prompt})

        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": 0.8,
            "max_tokens": 1500
        }

        try:
            response = requests.post(
                f"{self.api_base}/chat/completions",
                json=payload,
                timeout=120
            )
            response.raise_for_status()

            result = response.json()
            content = result['choices'][0]['message']['content']

            return {
                "success": True,
                "content": content,
                "prompt": prompt
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "prompt": prompt
            }

    def batch_generate(self, prompts: List[str], 
                       system_prompt: str = None,
                       max_workers: int = 3,
                       delay: float = 1.0) -> List[Dict]:
        """
        批量生成内容

        参数:
            prompts: 提示词列表
            system_prompt: 系统提示
            max_workers: 最大并发数
            delay: 请求间隔（秒）

        返回:
            生成结果列表
        """
        results = []
        total = len(prompts)

        print(f"开始批量生成，共 {total} 个任务")
        print("-" * 40)

        for i, prompt in enumerate(prompts, 1):
            print(f"[{i}/{total}] 正在生成...", end=" ")

            result = self.generate_single(prompt, system_prompt)
            results.append(result)

            if result['success']:
                print("成功")
            else:
                print(f"失败: {result['error']}")

            # 控制请求频率
            if i < total:
                time.sleep(delay)

        # 统计结果
        success_count = sum(1 for r in results if r['success'])
        print("-" * 40)
        print(f"完成！成功: {success_count}/{total}")

        return results

    def export_results(self, results: List[Dict], filename: str):
        """
        导出结果到JSON文件

        参数:
            results: 生成结果
            filename: 输出文件名
        """
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(results, f, ensure_ascii=False, indent=2)

        print(f"结果已保存到: {filename}")

def demo_product_description_generator():
    """演示：产品描述生成器"""
    generator = BatchContentGenerator(model="gpt-4")

    # 产品信息列表
    products = [
        "智能手表 - 防水50米，支持心率监测，续航7天",
        "无线耳机 - 主动降噪，30小时续航，支持无线充电",
        "便携音箱 - 360度环绕音效，IPX7防水，12小时续航",
        "机械键盘 - 青轴手感，RGB灯效，支持宏编程",
        "蓝牙鼠标 - 静音设计，可调DPI，续航6个月"
    ]

    # 生成提示词模板
    def make_prompt(product_info):
        return f"""请为以下产品生成一段营销文案，
要求：语言吸引人，突出产品卖点，适合电商平台展示，
字数控制在100-150字之间。

产品信息：{product_info}"""

    prompts = [make_prompt(p) for p in products]

    # 批量生成
    results = generator.batch_generate(prompts, delay=2)

    # 显示结果
    print("\n生成的产品描述:")
    print("=" * 60)
    for i, result in enumerate(results, 1):
        print(f"\n产品 {i}:")
        print("-" * 40)
        print(result.get('content', f"生成失败: {result.get('error')}"))

    # 导出结果
    generator.export_results(results, "product_descriptions.json")

def demo_article_outline_generator():
    """演示：文章大纲生成器"""
    generator = BatchContentGenerator(model="gpt-4")

    topics = [
        "如何在2024年学习编程",
        "远程工作的最佳实践",
        "人工智能对就业市场的影响",
        "可持续发展的生活方式",
        "数字时代的隐私保护"
    ]

    # 大纲生成提示模板
    def make_prompt(topic):
        return f"""请为以下主题生成一篇文章的大纲，
要求：结构清晰，包含引言、3-5个主要章节、结论，
每个章节包含2-3个子主题。

主题：{topic}"""

    prompts = [make_prompt(t) for t in topics]

    # 批量生成
    results = generator.batch_generate(prompts, delay=2)

    # 显示结果
    print("\n生成的文章大纲:")
    print("=" * 60)
    for i, result in enumerate(results, 1):
        print(f"\n{topics[i-1]}:")
        print("-" * 40)
        print(result.get('content', f"生成失败: {result.get('error')}"))

    # 导出结果
    generator.export_results(
        [
            {"topic": t, **r} 
            for t, r in zip(topics, results)
        ], 
        "article_outlines.json"
    )

if __name__ == "__main__":
    print("批量内容生成工具演示")
    print("=" * 60)

    print("\n1. 产品描述生成")
    print("-" * 40)
    demo_product_description_generator()

    print("\n\n2. 文章大纲生成")
    print("-" * 40)
    demo_article_outline_generator()

"""
成本优化策略示例
展示如何有效控制API调用成本
"""

# 策略一：使用更便宜的模型处理简单任务
def route_request(question: str, use_cheap_model: bool = True):
    """
    根据问题复杂度选择合适的模型

    简单问题使用便宜的模型，复杂问题使用强大模型
    """
    # 简单问题的特征
    simple_patterns = [
        "今天天气怎么样",
        "帮我查一下",
        "定义是什么",
        "翻译成英文"
    ]

    is_simple = any(pattern in question for pattern in simple_patterns)

    if is_simple and use_cheap_model:
        # 使用便宜的模型
        model = "gpt-3.5-turbo"
    else:
        # 使用更强大的模型
        model = "gpt-4"

    return model

# 策略二：减少不必要的token
def optimize_prompt(original_prompt: str, max_length: int = 2000) -> str:
    """
    优化提示词长度，避免发送过多token
    """
    if len(original_prompt) <= max_length:
        return original_prompt

    # 截断到最大长度
    return original_prompt[:max_length] + "\n\n[内容已截断]"

# 策略三：实现响应缓存
class ResponseCache:
    """
    简单的响应缓存
    减少重复请求
    """

    def __init__(self):
        self.cache = {}
        self.hits = 0
        self.misses = 0

    def get_cache_key(self, model: str, prompt: str) -> str:
        """生成缓存键"""
        return f"{model}:{hash(prompt)}"

    def get(self, model: str, prompt: str):
        """获取缓存的响应"""
        key = self.get_cache_key(model, prompt)
        result = self.cache.get(key)

        if result:
            self.hits += 1
        else:
            self.misses += 1

        return result

    def set(self, model: str, prompt: str, response: str):
        """设置缓存"""
        key = self.get_cache_key(model, prompt)
        self.cache[key] = response

    def stats(self):
        """返回缓存统计"""
        total = self.hits + self.misses
        hit_rate = (self.hits / total * 100) if total > 0 else 0
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": f"{hit_rate:.1f}%"
        }

"""
提高AI响应质量的技巧
"""

# 技巧一：清晰的系统提示
GOOD_SYSTEM_PROMPT = """
你是一位资深的Python软件工程师，有10年的开发经验。

回答规范：
1. 代码示例要完整可运行，包含必要的导入
2. 解释时使用简单的语言，避免过度技术化
3. 如果有多种解决方案，展示最优的那个
4. 重要概念用例子说明

格式规范：
- 使用Markdown格式化代码
- 适当使用emoji增加可读性
- 关键点用列表展示
"""

# 技巧二：few-shot示例
FEW_SHOT_EXAMPLES = """
用户：帮我写一个计算器
AI：当然可以！这是用Python写的简单计算器：

```python
def calculator():
    print("简易计算器")
    a = float(input("输入第一个数字: "))
    op = input("选择运算(+-*/): ")
    b = float(input("输入第二个数字: "))

    if op == '+': result = a + b
    elif op == '-': result = a - b
    elif op == '*': result = a * b
    elif op == '/': result = a / b if b != 0 else "错误"

    print(f"结果: {result}")

calculator()

{{
  "summary": "...",
  "key_points": ["...", "..."],
  "recommendation": "..."
}}
```"""

    payload = {
        "model": "gpt-4",
        "messages": [{"role": "user", "content": structured_prompt}],
        "temperature": 0.3  # 降低随机性以获得更一致的JSON
    }

    response = requests.post(f"{api_base}/chat/completions", json=payload)
    result = response.json()

    content = result['choices'][0]['message']['content']

    # 提取JSON部分
    import re
    json_match = re.search(r'\{.*\}', content, re.DOTALL)

    if json_match:
        return json.loads(json_match.group())

    return {"error": "无法解析JSON响应"}

"""
健壮的API调用实现
包含完整的错误处理和重试逻辑
"""

import time
import logging
from functools import wraps
from typing import Callable, Any

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
    """
    装饰器：失败时自动重试

    参数:
        max_retries: 最大重试次数
        delay: 重试间隔（秒）
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            last_exception = None

            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)

                except requests.exceptions.Timeout:
                    last_exception = Exception("请求超时")
                    logger.warning(f"第 {attempt + 1} 次尝试超时")

                except requests.exceptions.ConnectionError:
                    last_exception = Exception("连接错误")
                    logger.warning(f"第 {attempt + 1} 次尝试连接失败")

                except requests.exceptions.HTTPError as e:
                    if e.response.status_code == 429:
                        # 速率限制，等待更长时间
                        wait_time = delay * (2 ** attempt) * 5
                        logger.warning(f"速率限制，等待 {wait_time:.1f} 秒")
                        time.sleep(wait_time)
                        last_exception = e
                    elif e.response.status_code >= 500:
                        last_exception = e
                        logger.warning(f"服务器错误: {e.response.status_code}")
                    else:
                        raise

                if attempt < max_retries - 1:
                    wait_time = delay * (2 ** attempt)
                    logger.info(f"等待 {wait_time:.1f} 秒后重试...")
                    time.sleep(wait_time)

            logger.error(f"达到最大重试次数 {max_retries}")
            raise last_exception

        return wrapper
    return decorator

class RobustAPIClient:
    """
    健壮的API客户端
    提供可靠的大模型调用接口
    """

    def __init__(self, api_base: str, default_model: str = "gpt-4"):
        self.api_base = api_base
        self.default_model = default_model

    @retry_on_failure(max_retries=3, delay=2.0)
    def chat(self, prompt: str, model: str = None, **kwargs) -> str:
        """
        发送聊天请求

        参数:
            prompt: 用户输入
            model: 模型名称（可选）
            **kwargs: 其他参数

        返回:
            AI的回复内容
        """
        model = model or self.default_model

        payload = {
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            **kwargs
        }

        response = requests.post(
            f"{self.api_base}/chat/completions",
            json=payload,
            timeout=120
        )

        # 检查HTTP错误
        if response.status_code == 400:
            error_detail = response.json().get('error', {})
            raise ValueError(f"请求格式错误: {error_detail}")

        response.raise_for_status()

        result = response.json()
        return result['choices'][0]['message']['content']

    def health_check(self) -> bool:
        """
        健康检查

        返回:
            API是否可用
        """
        try:
            # 发送一个简单的测试请求
            test_payload = {
                "model": self.default_model,
                "messages": [{"role": "user", "content": "hi"}],
                "max_tokens": 5
            }

            response = requests.post(
                f"{self.api_base}/chat/completions",
                json=test_payload,
                timeout=10
            )

            return response.status_code == 200

        except Exception:
            return False

# 使用示例
if __name__ == "__main__":
    client = RobustAPIClient(
        api_base="http://localhost:8000/v1",
        default_model="gpt-4"
    )

    # 健康检查
    if client.health_check():
        print("API连接正常")

        try:
            response = client.chat(
                "用一句话介绍Python",
                temperature=0.7
            )
            print(f"AI回复: {response}")
        except Exception as e:
            print(f"请求失败: {e}")
    else:
        print("API连接失败，请检查服务是否运行")

# config.production.yaml
# 生产环境配置文件

server:
  # 服务绑定地址
  host: "0.0.0.0"
  port: 8000

  # 超时设置
  timeout:
    request: 120
    keepalive: 300

  # CORS设置（如果需要从浏览器调用）
  cors:
    enabled: true
    allowed_origins:
      - "https://your-domain.com"
      - "http://localhost:3000"
    allowed_methods:
      - "GET"
      - "POST"
    allowed_headers:
      - "*"

# 日志配置
logging:
  level: "INFO"
  format: "json"
  output: "/var/log/new-api/app.log"
  rotation:
    max_size: "100MB"
    backup_count: 10

# 速率限制
rate_limit:
  enabled: true
  default_rpm: 60  # 每分钟请求数
  per_model:
    gpt-4: 30
    gpt-3.5-turbo: 120

# AI服务商配置
providers:
  openai:
    api_key: "${OPENAI_API_KEY}"
    base_url: "https://api.openai.com/v1"
    models:
      - "gpt-4"
      - "gpt-3.5-turbo"

  anthropic:
    api_key: "${ANTHROPIC_API_KEY}"
    models:
      - "claude-3-opus"
      - "claude-3-sonnet"

  qwen:
    api_key: "${DASHSCOPE_API_KEY}"
    base_url: "https://dashscope.aliyuncs.com/compatible-mode/v1"
    models:
      - "qwen-turbo"
      - "qwen-plus"

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 创建非root用户
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

# 暴露端口
EXPOSE 8000

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
  CMD python -c "import requests; requests.get('http://localhost:8000/health').raise_for_status()"

# 启动命令
CMD ["python", "-m", "new_api.server", "--config", "config.production.yaml"]

# docker-compose.yml
version: '3.8'

services:
  new-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - DASHSCOPE_API_KEY=${DASHSCOPE_API_KEY}
    volumes:
      - ./config.yaml:/app/config.yaml:ro
      - ./logs:/app/logs
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  # 可选：添加Nginx反向代理
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - ./ssl:/etc/nginx/ssl:ro
    depends_on:
      - new-api

QuantumNous/new-api        - 本教程的主角，统一API管理平台
langchain-ai/langchain     - 强大的AI应用开发框架
hwchase17/langchain       - LangChain中文社区版本
chroma-core/chroma        - 向量数据库，用于AI知识库