AI开发~OpenAI连接各大公共模型：从入门到精通（进阶）

ftpeak

420人浏览 · 2026-05-22 14:25:28

ftpeak · 2026-05-22 14:25:28 发布

写在前面

你已经学会了如何用OpenAI的API连接各大模型，知道怎么发请求、怎么处理响应。但是，当你真正要开发一个AI应用时，你会发现这还不够。

为什么AI有时候答非所问？
如何让AI基于你的私有数据回答？
怎样让AI自动完成复杂任务？
如何降低API调用成本？

这篇教程就是为了回答这些问题而写的。我们会深入到RAG、Agent、Embeddings、微调等核心概念，让你从"会用API"进阶到"能构建AI应用"。

第一部分：Embeddings（向量化）——AI理解文本的秘密武器

1.1 什么是Embeddings？

大白话解释： 想象你要教一个外国人认识"苹果"这个词。你可以告诉他：苹果是红色的、圆的、甜的、可以吃的、长在树上的。这些特征组合起来，就是"苹果"的向量表示。

Embeddings就是把文字转换成数字列表（向量），让计算机能够理解文字的含义和关系。

深入理解： Embeddings是一种将离散的符号（如单词、句子）映射到连续向量空间的技术。它的核心思想是：

语义相似的文本，向量距离更近
可以计算文本之间的相似度
支持语义搜索、聚类、分类等任务

向量维度通常是1536（OpenAI的text-embedding-3-small）或3072（text-embedding-3-large）。

1.2 如何使用Embeddings？

from openai import OpenAI
import numpy as np

client = OpenAI(api_key="your-api-key")

def get_embedding(text: str, model: str = "text-embedding-3-small"):
    """获取文本的向量表示"""
    text = text.replace("\n", " ")
    response = client.embeddings.create(
        input=[text],
        model=model
    )
    return response.data[0].embedding

def cosine_similarity(vec1: list, vec2: list) -> float:
    """计算两个向量的余弦相似度"""
    vec1 = np.array(vec1)
    vec2 = np.array(vec2)

    dot_product = np.dot(vec1, vec2)
    norm1 = np.linalg.norm(vec1)
    norm2 = np.linalg.norm(vec2)

    return dot_product / (norm1 * norm2)

text1 = "我喜欢吃苹果"
text2 = "苹果是我的最爱"
text3 = "今天天气很好"

emb1 = get_embedding(text1)
emb2 = get_embedding(text2)
emb3 = get_embedding(text3)

print(f"'{text1}' 和 '{text2}' 的相似度: {cosine_similarity(emb1, emb2):.4f}")
print(f"'{text1}' 和 '{text3}' 的相似度: {cosine_similarity(emb1, emb3):.4f}")

输出示例：

'我喜欢吃苹果' 和 '苹果是我的最爱' 的相似度: 0.8765
'我喜欢吃苹果' 和 '今天天气很好' 的相似度: 0.3421

1.3 实战：构建语义搜索引擎

import json
from typing import List, Dict
import pickle

class SemanticSearchEngine:
    """语义搜索引擎"""

    def __init__(self, model: str = "text-embedding-3-small"):
        self.client = OpenAI(api_key="your-api-key")
        self.model = model
        self.documents = []
        self.embeddings = []

    def add_document(self, doc: str, metadata: dict = None):
        """添加文档"""
        self.documents.append({
            "text": doc,
            "metadata": metadata or {}
        })

        embedding = self._get_embedding(doc)
        self.embeddings.append(embedding)

    def _get_embedding(self, text: str) -> list:
        """获取向量"""
        response = self.client.embeddings.create(
            input=[text],
            model=self.model
        )
        return response.data[0].embedding

    def search(self, query: str, top_k: int = 5) -> List[Dict]:
        """语义搜索"""
        query_embedding = self._get_embedding(query)

        similarities = []
        for i, doc_embedding in enumerate(self.embeddings):
            similarity = self._cosine_similarity(query_embedding, doc_embedding)
            similarities.append((i, similarity))

        similarities.sort(key=lambda x: x[1], reverse=True)

        results = []
        for i, similarity in similarities[:top_k]:
            results.append({
                "document": self.documents[i],
                "similarity": similarity
            })

        return results

    def _cosine_similarity(self, vec1: list, vec2: list) -> float:
        """计算余弦相似度"""
        vec1 = np.array(vec1)
        vec2 = np.array(vec2)
        return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

    def save(self, filepath: str):
        """保存索引"""
        with open(filepath, 'wb') as f:
            pickle.dump({
                'documents': self.documents,
                'embeddings': self.embeddings
            }, f)

    def load(self, filepath: str):
        """加载索引"""
        with open(filepath, 'rb') as f:
            data = pickle.load(f)
            self.documents = data['documents']
            self.embeddings = data['embeddings']

engine = SemanticSearchEngine()

documents = [
    "Python是一种流行的编程语言，适合初学者学习",
    "机器学习是人工智能的一个分支，让计算机从数据中学习",
    "深度学习使用神经网络来处理复杂的数据模式",
    "自然语言处理让计算机能够理解和生成人类语言",
    "计算机视觉让机器能够理解和分析图像和视频"
]

for doc in documents:
    engine.add_document(doc)

results = engine.search("如何让计算机理解语言？")
for result in results:
    print(f"相似度: {result['similarity']:.4f}")
    print(f"文档: {result['document']['text']}\n")

1.4 Embeddings的应用场景

应用场景	说明	示例
语义搜索	根据含义而非关键词搜索	搜索"如何学习编程"能找到"Python入门教程"
文本聚类	将相似文本分组	新闻分类、用户评论分组
推荐系统	基于相似度推荐	"喜欢这篇文章的人也喜欢..."
重复检测	识别相似内容	检测重复问题、抄袭检测
RAG	检索相关文档供AI参考	企业知识库问答

第二部分：RAG（检索增强生成）——让AI拥有你的知识

2.1 什么是RAG？

大白话解释： 想象你在考试，可以带一本参考书。遇到不会的题，你先翻书找到相关内容，再结合自己的理解回答。RAG就是让AI也能"翻书"——先从知识库中检索相关信息，再基于这些信息回答问题。

深入理解： RAG（Retrieval-Augmented Generation）是一种结合检索和生成的技术：

检索阶段：从知识库中找到与问题相关的文档片段
增强阶段：将检索到的内容作为上下文提供给模型
生成阶段：模型基于上下文生成答案

为什么需要RAG？

知识时效性：模型训练有截止日期，RAG可以提供最新信息
私有数据：模型不知道你的企业内部文档、产品手册
减少幻觉：有据可依，降低AI胡编乱造的概率
可追溯性：可以知道答案来自哪些文档

2.2 RAG系统架构

用户问题
    ↓
[问题向量化]
    ↓
[向量数据库检索] ← 预先存储的文档向量
    ↓
[获取相关文档片段]
    ↓
[构建Prompt：问题 + 相关文档]
    ↓
[LLM生成答案]
    ↓
最终答案（带引用来源）

2.3 实战：构建完整的RAG系统

from openai import OpenAI
import numpy as np
from typing import List, Dict, Optional
import re

class SimpleVectorDB:
    """简单的向量数据库（生产环境建议使用Pinecone、Weaviate等）"""

    def __init__(self):
        self.documents = []
        self.embeddings = []
        self.metadatas = []

    def add(self, docs: List[str], embeddings: List[List[float]], metadatas: List[Dict] = None):
        """添加文档"""
        self.documents.extend(docs)
        self.embeddings.extend(embeddings)
        self.metadatas.extend(metadatas or [{}] * len(docs))

    def search(self, query_embedding: List[float], top_k: int = 5) -> List[Dict]:
        """搜索最相似的文档"""
        similarities = []
        for i, emb in enumerate(self.embeddings):
            sim = self._cosine_similarity(query_embedding, emb)
            similarities.append((i, sim))

        similarities.sort(key=lambda x: x[1], reverse=True)

        results = []
        for i, sim in similarities[:top_k]:
            results.append({
                "text": self.documents[i],
                "metadata": self.metadatas[i],
                "score": sim
            })

        return results

    def _cosine_similarity(self, vec1, vec2):
        vec1 = np.array(vec1)
        vec2 = np.array(vec2)
        return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

class TextSplitter:
    """文本分割器"""

    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    def split_text(self, text: str) -> List[str]:
        """将长文本分割成小块"""
        sentences = re.split(r'[。！？\n]', text)
        sentences = [s.strip() for s in sentences if s.strip()]

        chunks = []
        current_chunk = []
        current_size = 0

        for sentence in sentences:
            sentence_size = len(sentence)

            if current_size + sentence_size > self.chunk_size and current_chunk:
                chunks.append(''.join(current_chunk))

                overlap_sentences = []
                overlap_size = 0
                for s in reversed(current_chunk):
                    if overlap_size + len(s) <= self.chunk_overlap:
                        overlap_sentences.insert(0, s)
                        overlap_size += len(s)
                    else:
                        break

                current_chunk = overlap_sentences
                current_size = overlap_size

            current_chunk.append(sentence)
            current_size += sentence_size

        if current_chunk:
            chunks.append(''.join(current_chunk))

        return chunks

class RAGSystem:
    """RAG系统"""

    def __init__(
        self,
        embedding_model: str = "text-embedding-3-small",
        generation_model: str = "gpt-4-turbo",
        top_k: int = 3
    ):
        self.client = OpenAI(api_key="your-api-key")
        self.embedding_model = embedding_model
        self.generation_model = generation_model
        self.top_k = top_k

        self.vector_db = SimpleVectorDB()
        self.text_splitter = TextSplitter()

    def add_documents(self, documents: List[str], metadatas: List[Dict] = None):
        """添加文档到知识库"""
        all_chunks = []
        all_embeddings = []
        all_metadatas = []

        for i, doc in enumerate(documents):
            chunks = self.text_splitter.split_text(doc)
            all_chunks.extend(chunks)

            embeddings = self._get_embeddings(chunks)
            all_embeddings.extend(embeddings)

            metadata = metadatas[i] if metadatas else {}
            all_metadatas.extend([metadata] * len(chunks))

        self.vector_db.add(all_chunks, all_embeddings, all_metadatas)
        print(f"已添加 {len(all_chunks)} 个文档片段到知识库")

    def _get_embeddings(self, texts: List[str]) -> List[List[float]]:
        """批量获取向量"""
        response = self.client.embeddings.create(
            input=texts,
            model=self.embedding_model
        )
        return [item.embedding for item in response.data]

    def query(
        self,
        question: str,
        system_prompt: Optional[str] = None,
        show_sources: bool = True
    ) -> Dict:
        """查询RAG系统"""
        question_embedding = self._get_embeddings([question])[0]

        relevant_docs = self.vector_db.search(question_embedding, top_k=self.top_k)

        context = "\n\n---\n\n".join([doc["text"] for doc in relevant_docs])

        if system_prompt is None:
            system_prompt = """你是一个专业的问答助手。请基于提供的参考资料回答用户问题。

要求：
1. 优先使用参考资料中的信息
2. 如果参考资料中没有相关信息，请明确说明
3. 回答要准确、简洁、有条理
4. 如果需要，可以引用参考资料的具体内容"""

        messages = [
            {"role": "system", "content": system_prompt},
            {
                "role": "user",
                "content": f"""参考资料：
{context}

---
用户问题：{question}

请基于以上参考资料回答问题。"""
            }
        ]

        response = self.client.chat.completions.create(
            model=self.generation_model,
            messages=messages,
            temperature=0.3
        )

        answer = response.choices[0].message.content

        result = {
            "question": question,
            "answer": answer,
            "sources": relevant_docs if show_sources else None
        }

        return result

    def chat(self, question: str, stream: bool = False) -> str:
        """简化的对话接口"""
        result = self.query(question, show_sources=False)
        return result["answer"]

knowledge_base = [
    """
    公司产品手册 - Python编程课程

    我们的Python课程分为三个级别：
    1. 入门级（Python基础）：适合零基础学员，学习Python语法、数据类型、控制流程等基础知识。课程时长8周，学费2999元。
    2. 进阶级（Python实战）：适合有基础的学员，学习面向对象编程、文件操作、数据库连接等。课程时长10周，学费4999元。
    3. 高级（Python数据分析）：学习NumPy、Pandas、Matplotlib等数据分析库。课程时长12周，学费6999元。

    所有课程都提供：
    - 在线直播授课
    - 录播视频回放
    - 课后作业批改
    - 班主任全程跟踪
    - 就业推荐服务

    报名方式：官网注册或联系客服400-123-4567
    """,
    """
    公司退款政策

    1. 开课前7天以上申请退款：全额退款
    2. 开课前3-7天申请退款：退还80%学费
    3. 开课前3天内申请退款：退还50%学费
    4. 开课后不支持退款，但可以申请延期学习

    退款流程：
    1. 登录官网个人中心
    2. 点击"我的订单"
    3. 选择需要退款的课程
    4. 填写退款原因
    5. 提交申请，3-5个工作日内处理

    联系方式：退款专线400-123-4568
    """
]

rag = RAGSystem()
rag.add_documents(knowledge_base)

print("=" * 50)
print("问题1：Python入门课程多少钱？")
print("=" * 50)
result1 = rag.query("Python入门课程多少钱？")
print(f"回答：{result1['answer']}\n")

print("=" * 50)
print("问题2：如果我想退款，什么时候可以全额退？")
print("=" * 50)
result2 = rag.query("如果我想退款，什么时候可以全额退？")
print(f"回答：{result2['answer']}\n")

print("=" * 50)
print("问题3：课程都包含哪些服务？")
print("=" * 50)
result3 = rag.query("课程都包含哪些服务？")
print(f"回答：{result3['answer']}\n")

if result3['sources']:
    print("\n参考来源：")
    for i, source in enumerate(result3['sources'], 1):
        print(f"{i}. 相似度: {source['score']:.4f}")
        print(f"   内容片段: {source['text'][:100]}...")

2.4 RAG的优化技巧

2.4.1 文本分割策略

class AdvancedTextSplitter:
    """高级文本分割器"""

    def __init__(
        self,
        chunk_size: int = 500,
        chunk_overlap: int = 50,
        separators: List[str] = None
    ):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.separators = separators or ["\n\n", "\n", "。", "！", "？", " "]

    def split_text(self, text: str) -> List[str]:
        """递归分割文本"""
        final_chunks = []

        separator = self.separators[-1]
        for sep in self.separators:
            if sep in text:
                separator = sep
                break

        if separator:
            splits = text.split(separator)
        else:
            splits = [text]

        current_chunk = []
        current_length = 0

        for split in splits:
            split_length = len(split)

            if current_length + split_length > self.chunk_size:
                if current_chunk:
                    final_chunks.append(separator.join(current_chunk))

                overlap = self._get_overlap(current_chunk, separator)
                current_chunk = overlap
                current_length = sum(len(s) for s in current_chunk)

            if split_length > self.chunk_size:
                sub_chunks = self.split_text(split)
                final_chunks.extend(sub_chunks)
            else:
                current_chunk.append(split)
                current_length += split_length

        if current_chunk:
            final_chunks.append(separator.join(current_chunk))

        return final_chunks

    def _get_overlap(self, chunks: List[str], separator: str) -> List[str]:
        """获取重叠部分"""
        if not chunks:
            return []

        overlap_chunks = []
        overlap_length = 0

        for chunk in reversed(chunks):
            if overlap_length + len(chunk) <= self.chunk_overlap:
                overlap_chunks.insert(0, chunk)
                overlap_length += len(chunk)
            else:
                break

        return overlap_chunks

2.4.2 重排序（Reranking）

class Reranker:
    """重排序器 - 提高检索精度"""

    def __init__(self, client: OpenAI):
        self.client = client

    def rerank(
        self,
        query: str,
        documents: List[Dict],
        top_k: int = 3
    ) -> List[Dict]:
        """使用LLM重新排序文档"""
        prompt = f"""你是一个文档相关性评估专家。

用户问题：{query}

请对以下文档片段与问题的相关性进行评分（1-10分）：

"""
        for i, doc in enumerate(documents, 1):
            prompt += f"文档{i}：\n{doc['text'][:200]}...\n\n"

        prompt += """
请以JSON格式返回评分结果，格式如下：
{"scores": [分数1, 分数2, ...]}
只返回JSON，不要其他内容。"""

        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0
        )

        import json
        scores = json.loads(response.choices[0].message.content)["scores"]

        for i, doc in enumerate(documents):
            doc["rerank_score"] = scores[i]

        documents.sort(key=lambda x: x["rerank_score"], reverse=True)

        return documents[:top_k]

第三部分：Prompt Engineering（提示词工程）——与AI高效沟通的艺术

3.1 为什么Prompt Engineering很重要？

大白话解释： 同样一个问题，不同的问法会得到完全不同的答案。Prompt Engineering就是学习如何"正确地问问题"，让AI给出你想要的答案。

深入理解： Prompt Engineering是一门通过设计和优化提示词来引导大语言模型生成高质量输出的技术。它涉及：

指令设计：明确告诉AI要做什么
上下文管理：提供必要的背景信息
示例学习：通过例子教会AI你的期望
思维链：引导AI一步步推理

3.2 Prompt设计的核心原则

原则1：明确具体

# ❌ 模糊的提示词
prompt1 = "写一篇文章"

# ✅ 明确的提示词
prompt2 = """请写一篇关于人工智能在教育领域应用的文章。

要求：
1. 字数：800-1000字
2. 结构：引言、现状分析、应用案例、未来展望、结论
3. 风格：专业但不晦涩，适合教育工作者阅读
4. 包含至少2个具体的应用案例
5. 使用数据和统计来支持观点"""

原则2：提供上下文

# ❌ 缺少上下文
prompt1 = "如何优化这段代码？"

# ✅ 提供完整上下文
prompt2 = """我正在开发一个Python Web应用，使用Flask框架。

代码片段：
```python
def get_users():
    users = []
    for i in range(1000):
        user = db.query(f"SELECT * FROM users WHERE id = {i}")
        users.append(user)
    return users

问题：这个函数响应时间太长（超过5秒）。

请帮我优化这段代码，要求：

解释性能问题的原因
提供优化后的代码
说明优化后的预期性能提升"""

原则3：使用示例（Few-shot Learning）

def few_shot_prompt():
    """少样本学习示例"""
    prompt = """任务：将产品描述转换为结构化的产品信息。

示例1：
输入：这款iPhone 15 Pro Max拥有256GB存储空间，钛金属边框，A17 Pro芯片，售价999美元。
输出：
{
  "product": "iPhone 15 Pro Max",
  "storage": "256GB",
  "material": "钛金属",
  "chip": "A17 Pro",
  "price": "$999"
}

示例2：
输入：MacBook Air M3，13.6英寸屏幕，8GB内存，256GB SSD，星光色，售价1099美元。
输出：
{
  "product": "MacBook Air M3",
  "screen": "13.6英寸",
  "memory": "8GB",
  "storage": "256GB SSD",
  "color": "星光色",
  "price": "$1099"
}

现在请处理：
输入：戴尔XPS 15，Intel i7处理器，16GB内存，512GB SSD，OLED 3.5K显示屏，售价1499美元。
输出："""

    return prompt

3.3 高级Prompt技术

3.3.1 思维链（Chain of Thought）

def chain_of_thought_prompt():
    """思维链提示 - 让AI展示推理过程"""
    prompt = """问题：小明有5个苹果，他给了小红2个，又从妈妈那里得到了3个，然后吃掉了1个。请问小明现在有几个苹果？

请一步步思考：
1. 首先，小明有5个苹果
2. 他给了小红2个，所以现在有 5 - 2 = 3个
3. 从妈妈那里得到3个，所以现在有 3 + 3 = 6个
4. 吃掉了1个，所以现在有 6 - 1 = 5个

答案：小明现在有5个苹果。

---

现在请用同样的方式解决这个问题：
问题：一本书有240页，小红第一天读了全书的1/4，第二天读了剩下的1/3，第三天读了40页。还剩多少页没读？"""

    return prompt

3.3.2 角色扮演

def role_play_prompt():
    """角色扮演提示"""
    prompts = {
        "expert": """你是一位拥有20年经验的资深Python开发者，专注于性能优化和代码质量。
你的回答应该：
1. 提供专业的技术见解
2. 指出潜在的性能瓶颈
3. 给出最佳实践建议
4. 引用相关的PEP规范或设计模式

用户问题：{question}""",

        "teacher": """你是一位耐心的编程老师，擅长用简单的语言解释复杂的概念。
你的教学风格：
1. 用生活中的类比来解释技术概念
2. 循序渐进，从简单到复杂
3. 鼓励学生提问
4. 提供可运行的代码示例

学生问题：{question}""",

        "reviewer": """你是一位严格的代码审查员，关注代码的可读性、可维护性和安全性。
审查标准：
1. 代码风格是否符合PEP 8
2. 是否有潜在的安全问题
3. 是否有性能问题
4. 是否有更好的实现方式

请审查以下代码：{question}"""
    }

    return prompts

3.3.3 结构化输出

import json
from pydantic import BaseModel
from typing import List

class ProductInfo(BaseModel):
    name: str
    price: float
    features: List[str]
    pros: List[str]
    cons: List[str]

def structured_output_prompt():
    """结构化输出提示"""
    prompt = """分析以下产品描述，提取结构化信息。

产品描述：
这款索尼WH-1000XM5无线降噪耳机，售价349美元，拥有业界领先的降噪技术，30小时续航，支持多点连接。优点包括出色的降噪效果、舒适的佩戴体验、清晰的音质。缺点是价格较高、体积较大、不支持折叠。

请以JSON格式输出，包含以下字段：
- name: 产品名称
- price: 价格（数字）
- features: 主要特性列表
- pros: 优点列表
- cons: 缺点列表

只输出JSON，不要其他内容。"""

    return prompt

def get_structured_response(client: OpenAI, prompt: str, model: str = "gpt-4-turbo"):
    """获取结构化响应"""
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"}
    )

    return json.loads(response.choices[0].message.content)

3.4 Prompt模板管理

from string import Template

class PromptTemplateManager:
    """Prompt模板管理器"""

    def __init__(self):
        self.templates = {}

    def register(self, name: str, template: str):
        """注册模板"""
        self.templates[name] = Template(template)

    def render(self, name: str, **kwargs) -> str:
        """渲染模板"""
        if name not in self.templates:
            raise ValueError(f"模板 '{name}' 不存在")

        return self.templates[name].safe_substitute(**kwargs)

manager = PromptTemplateManager()

manager.register(
    "summarize",
    """请总结以下文章：

标题：$title
内容：$content

要求：
1. 总结长度：$length 字左右
2. 重点突出：$focus
3. 语言风格：$style"""
)

manager.register(
    "translate",
    """请将以下文本从 $source_lang 翻译成 $target_lang：

原文：
$text

翻译要求：
1. 保持原文的语气和风格
2. 专业术语保持准确
3. 适当添加注释说明文化差异"""
)

summary_prompt = manager.render(
    "summarize",
    title="人工智能的未来",
    content="..." * 100,
    length="200",
    focus="技术发展趋势",
    style="专业客观"
)

print(summary_prompt)

第四部分：Agent（智能体）——让AI学会自主行动

4.1 什么是Agent？

大白话解释： 普通的AI对话就像一个只会回答问题的顾问。Agent则像一个能干活的员工——你给它一个目标，它会自己规划步骤、调用工具、执行任务，直到完成目标。

深入理解： Agent是一个能够自主感知、决策和行动的AI系统。它包含以下核心组件：

感知模块：理解环境和任务
规划模块：制定行动计划
工具模块：调用外部工具和API
记忆模块：存储和检索信息
执行模块：执行具体操作

4.2 Agent的工作流程

用户目标
    ↓
[理解目标] → 分析任务需求
    ↓
[制定计划] → 分解为子任务
    ↓
[执行步骤]
    ├→ 需要工具？ → 调用工具 → 获取结果
    ├→ 需要信息？ → 搜索/查询 → 获取信息
    └→ 需要决策？ → 推理判断 → 做出决策
    ↓
[检查结果] → 是否完成？
    ├→ 是 → 返回最终答案
    └→ 否 → 调整计划 → 继续执行

4.3 实战：构建一个简单的Agent

from openai import OpenAI
import json
from typing import List, Dict, Callable, Any
import re

class Tool:
    """工具基类"""

    def __init__(self, name: str, description: str, func: Callable):
        self.name = name
        self.description = description
        self.func = func
        self.parameters = self._extract_parameters()

    def _extract_parameters(self) -> Dict:
        """提取参数信息（简化版）"""
        import inspect
        sig = inspect.signature(self.func)

        parameters = {
            "type": "object",
            "properties": {},
            "required": []
        }

        for param_name, param in sig.parameters.items():
            param_info = {"type": "string"}

            if param.default == inspect.Parameter.empty:
                parameters["required"].append(param_name)

            parameters["properties"][param_name] = param_info

        return parameters

    def to_openai_tool(self) -> Dict:
        """转换为OpenAI工具格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters
            }
        }

    def execute(self, **kwargs) -> Any:
        """执行工具"""
        return self.func(**kwargs)

class SimpleAgent:
    """简单的Agent实现"""

    def __init__(
        self,
        client: OpenAI,
        model: str = "gpt-4-turbo",
        system_prompt: str = None
    ):
        self.client = client
        self.model = model
        self.tools: Dict[str, Tool] = {}
        self.conversation_history = []

        self.system_prompt = system_prompt or """你是一个智能助手，能够使用工具来帮助用户完成任务。

当需要使用工具时，请严格按照以下格式：
<thought>你的思考过程</thought>
<tool>工具名称</tool>
<arguments>{"参数名": "参数值"}</arguments>

当任务完成时，直接回答用户。"""

    def register_tool(self, tool: Tool):
        """注册工具"""
        self.tools[tool.name] = tool

    def run(self, user_input: str, max_iterations: int = 10) -> str:
        """运行Agent"""
        self.conversation_history.append({
            "role": "user",
            "content": user_input
        })

        for iteration in range(max_iterations):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": self.system_prompt},
                    *self.conversation_history
                ],
                tools=[tool.to_openai_tool() for tool in self.tools.values()],
                tool_choice="auto"
            )

            message = response.choices[0].message

            if message.tool_calls:
                self.conversation_history.append(message)

                for tool_call in message.tool_calls:
                    tool_name = tool_call.function.name
                    tool_args = json.loads(tool_call.function.arguments)

                    print(f"[调用工具] {tool_name}({tool_args})")

                    if tool_name in self.tools:
                        result = self.tools[tool_name].execute(**tool_args)

                        self.conversation_history.append({
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "name": tool_name,
                            "content": str(result)
                        })
                    else:
                        self.conversation_history.append({
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "name": tool_name,
                            "content": f"错误：工具 '{tool_name}' 不存在"
                        })
            else:
                final_response = message.content
                self.conversation_history.append({
                    "role": "assistant",
                    "content": final_response
                })
                return final_response

        return "达到最大迭代次数，任务未完成"

def get_weather(city: str) -> str:
    """获取天气信息（模拟）"""
    weather_data = {
        "北京": "晴天，温度25°C，空气质量良好",
        "上海": "多云，温度28°C，有轻微雾霾",
        "广州": "雷阵雨，温度32°C，湿度高",
        "深圳": "晴天，温度30°C，紫外线强"
    }
    return weather_data.get(city, f"未找到{city}的天气信息")

def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

def search_web(query: str) -> str:
    """搜索网络（模拟）"""
    search_results = {
        "Python": "Python是一种流行的编程语言，广泛用于Web开发、数据科学、人工智能等领域。",
        "机器学习": "机器学习是人工智能的一个分支，让计算机从数据中学习模式和规律。",
        "深度学习": "深度学习使用多层神经网络来学习数据的复杂表示。"
    }

    for keyword, result in search_results.items():
        if keyword in query:
            return result

    return f"搜索'{query}'的结果：未找到相关信息"

client = OpenAI(api_key="your-api-key")
agent = SimpleAgent(client)

agent.register_tool(Tool(
    name="get_weather",
    description="获取指定城市的天气信息",
    func=get_weather
))

agent.register_tool(Tool(
    name="calculate",
    description="计算数学表达式，如'2+3*4'",
    func=calculate
))

agent.register_tool(Tool(
    name="search_web",
    description="搜索网络获取信息",
    func=search_web
))

print("=" * 50)
print("测试1：天气查询")
print("=" * 50)
result1 = agent.run("北京今天天气怎么样？")
print(f"回答：{result1}\n")

print("=" * 50)
print("测试2：数学计算")
print("=" * 50)
result2 = agent.run("帮我计算一下 (25 + 35) * 2 等于多少")
print(f"回答：{result2}\n")

print("=" * 50)
print("测试3：复杂任务")
print("=" * 50)
result3 = agent.run("北京和上海哪个城市温度更高？高多少度？")
print(f"回答：{result3}\n")

4.4 ReAct模式：推理+行动

class ReActAgent:
    """ReAct模式的Agent：Reasoning + Acting"""

    def __init__(self, client: OpenAI, model: str = "gpt-4-turbo"):
        self.client = client
        self.model = model
        self.tools = {}

        self.system_prompt = """你是一个遵循ReAct模式的智能助手。

对于每个任务，你需要：
1. Thought（思考）：分析当前情况，决定下一步行动
2. Action（行动）：选择并执行工具
3. Observation（观察）：查看行动结果
4. 重复以上步骤直到任务完成

输出格式：
Thought: [你的思考]
Action: [工具名称]
Action Input: [工具参数]

或者当任务完成时：
Thought: [最终思考]
Final Answer: [最终答案]"""

    def register_tool(self, name: str, func: Callable, description: str):
        """注册工具"""
        self.tools[name] = {
            "func": func,
            "description": description
        }

    def run(self, task: str, max_iterations: int = 5) -> str:
        """运行ReAct循环"""
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": task}
        ]

        for _ in range(max_iterations):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0
            )

            output = response.choices[0].message.content
            messages.append({"role": "assistant", "content": output})

            print(output)
            print("-" * 50)

            if "Final Answer:" in output:
                final_answer = output.split("Final Answer:")[-1].strip()
                return final_answer

            if "Action:" in output and "Action Input:" in output:
                action_match = re.search(r"Action:\s*(\w+)", output)
                input_match = re.search(r"Action Input:\s*(.+)", output)

                if action_match and input_match:
                    action = action_match.group(1)
                    action_input = input_match.group(1).strip()

                    if action in self.tools:
                        result = self.tools[action]["func"](action_input)
                        observation = f"Observation: {result}"
                        messages.append({"role": "user", "content": observation})
                        print(observation)
                        print("-" * 50)

        return "任务未能在限定步骤内完成"

react_agent = ReActAgent(client)
react_agent.register_tool("search", search_web, "搜索网络")
react_agent.register_tool("calculate", calculate, "计算表达式")

result = react_agent.run("搜索Python的相关信息，然后告诉我它主要用于哪些领域")
print(f"\n最终答案：{result}")

第五部分：Fine-tuning（微调）——打造专属模型

5.1 什么是微调？

大白话解释： 预训练模型就像一个大学毕业生，知识面广但不够专业。微调就像岗前培训，让模型在特定领域变得更专业。

深入理解： Fine-tuning是在预训练模型的基础上，使用特定领域的数据继续训练，使模型适应特定任务或风格的过程。

微调 vs RAG： | 对比维度 | 微调 | RAG | |---------|------|-----| | 知识更新 | 需要重新训练 | 实时更新 | | 成本 | 较高（训练费用） | 较低（只需存储） | | 适用场景 | 特定风格、格式 | 知识密集型任务 | | 数据隐私 | 数据上传到模型 | 数据保留在本地 | | 灵活性 | 较低 | 较高 |

5.2 微调的适用场景

✅ 适合微调：

需要特定的输出风格（如品牌调性）
特定的格式要求（如JSON、代码）
专业术语和领域知识
降低推理成本（小模型微调替代大模型）

❌ 不适合微调：

知识频繁更新
数据量小
需要可追溯的知识来源
预算有限

5.3 准备微调数据

import json
from typing import List, Dict

def prepare_fine_tuning_data(
    conversations: List[Dict[str, str]],
    output_file: str = "fine_tuning_data.jsonl"
):
    """
    准备微调数据

    Args:
        conversations: 对话列表，格式为 [{"system": "...", "user": "...", "assistant": "..."}]
        output_file: 输出文件路径
    """
    with open(output_file, 'w', encoding='utf-8') as f:
        for conv in conversations:
            messages = []

            if "system" in conv:
                messages.append({
                    "role": "system",
                    "content": conv["system"]
                })

            messages.append({
                "role": "user",
                "content": conv["user"]
            })

            messages.append({
                "role": "assistant",
                "content": conv["assistant"]
            })

            line = json.dumps({"messages": messages}, ensure_ascii=False)
            f.write(line + '\n')

    print(f"已生成 {len(conversations)} 条训练数据")

training_data = [
    {
        "system": "你是某电商平台的客服助手，语气要亲切友好。",
        "user": "我的订单怎么还没发货？",
        "assistant": "亲，非常抱歉让您久等了！我帮您查了一下，您的订单正在打包中，预计今天下午就会发货。发货后您会收到短信通知的。如果比较着急，我也可以帮您催促一下仓库哦~"
    },
    {
        "system": "你是某电商平台的客服助手，语气要亲切友好。",
        "user": "这个商品支持退货吗？",
        "assistant": "亲，这款商品是支持7天无理由退货的哦！只要商品保持原样、不影响二次销售，您可以在收到货后的7天内申请退货。退货流程也很简单，在订单详情页点击'申请退货'就可以了。有什么不清楚的随时问我哈~"
    },
    {
        "system": "你是某电商平台的客服助手，语气要亲切友好。",
        "user": "运费多少钱？",
        "assistant": "亲，我们店铺全场包邮哦！不管您买多少，都是免运费的。部分地区（如偏远地区）可能需要额外补一点运费差价，但我们会提前和您确认的。放心购买吧~"
    }
]

prepare_fine_tuning_data(training_data, "customer_service_fine_tune.jsonl")

def validate_fine_tuning_data(filepath: str):
    """验证微调数据格式"""
    with open(filepath, 'r', encoding='utf-8') as f:
        for i, line in enumerate(f, 1):
            try:
                data = json.loads(line)

                if "messages" not in data:
                    print(f"行 {i}: 缺少 'messages' 字段")
                    continue

                for msg in data["messages"]:
                    if "role" not in msg or "content" not in msg:
                        print(f"行 {i}: 消息格式错误")
                        break

                print(f"行 {i}: ✓ 格式正确")

            except json.JSONDecodeError:
                print(f"行 {i}: JSON解析错误")

validate_fine_tuning_data("customer_service_fine_tune.jsonl")

5.4 OpenAI微调流程

from openai import OpenAI
import time

class FineTuningManager:
    """微调管理器"""

    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)

    def upload_file(self, filepath: str) -> str:
        """上传训练文件"""
        with open(filepath, 'rb') as f:
            response = self.client.files.create(
                file=f,
                purpose='fine-tune'
            )

        print(f"文件上传成功，ID: {response.id}")
        return response.id

    def create_fine_tuning_job(
        self,
        training_file_id: str,
        model: str = "gpt-3.5-turbo-0125",
        hyperparameters: Dict = None
    ) -> str:
        """创建微调任务"""
        if hyperparameters is None:
            hyperparameters = {
                "n_epochs": 3,
                "batch_size": 4,
                "learning_rate_multiplier": 0.1
            }

        response = self.client.fine_tuning.jobs.create(
            training_file=training_file_id,
            model=model,
            hyperparameters=hyperparameters
        )

        print(f"微调任务已创建，ID: {response.id}")
        return response.id

    def check_job_status(self, job_id: str) -> Dict:
        """检查任务状态"""
        response = self.client.fine_tuning.jobs.retrieve(job_id)

        status_info = {
            "id": response.id,
            "status": response.status,
            "model": response.model,
            "created_at": response.created_at,
            "finished_at": response.finished_at,
            "fine_tuned_model": response.fine_tuned_model
        }

        return status_info

    def wait_for_completion(self, job_id: str, check_interval: int = 60) -> str:
        """等待任务完成"""
        print(f"开始监控微调任务 {job_id}...")

        while True:
            status = self.check_job_status(job_id)
            print(f"状态: {status['status']}")

            if status['status'] == 'succeeded':
                print(f"微调成功！模型ID: {status['fine_tuned_model']}")
                return status['fine_tuned_model']
            elif status['status'] == 'failed':
                print("微调失败！")
                return None

            time.sleep(check_interval)

    def list_fine_tuned_models(self):
        """列出所有微调模型"""
        response = self.client.fine_tuning.jobs.list()

        print("已创建的微调模型：")
        for job in response.data:
            print(f"- ID: {job.id}")
            print(f"  状态: {job.status}")
            print(f"  基础模型: {job.model}")
            print(f"  微调模型: {job.fine_tuned_model}")
            print()

manager = FineTuningManager(api_key="your-api-key")

file_id = manager.upload_file("customer_service_fine_tune.jsonl")

job_id = manager.create_fine_tuning_job(file_id)

# fine_tuned_model = manager.wait_for_completion(job_id)

# if fine_tuned_model:
#     response = manager.client.chat.completions.create(
#         model=fine_tuned_model,
#         messages=[
#             {"role": "user", "content": "我的包裹什么时候能到？"}
#         ]
#     )
#     print(response.choices[0].message.content)

5.5 微调最佳实践

class FineTuningBestPractices:
    """微调最佳实践指南"""

    @staticmethod
    def estimate_cost(num_tokens: int, num_epochs: int = 3) -> float:
        """
        估算微调成本

        OpenAI微调定价（2024年）：
        - GPT-3.5-Turbo: $0.008 / 1K tokens (训练)
        - GPT-4: $0.025 / 1K tokens (训练)
        """
        cost_per_1k_tokens = 0.008
        total_cost = (num_tokens / 1000) * cost_per_1k_tokens * num_epochs
        return total_cost

    @staticmethod
    def calculate_tokens_estimate(num_examples: int, avg_tokens_per_example: int) -> int:
        """估算总token数"""
        return num_examples * avg_tokens_per_example

    @staticmethod
    def get_minimum_examples():
        """获取最少训练样本建议"""
        return {
            "简单任务（风格调整）": "100-500条",
            "中等任务（特定格式）": "500-1000条",
            "复杂任务（领域知识）": "1000-10000条",
            "推荐数量": "至少500条，越多越好"
        }

    @staticmethod
    def data_quality_checklist():
        """数据质量检查清单"""
        return [
            "✓ 数据格式正确（JSONL）",
            "✓ 每条数据包含完整的对话",
            "✓ 用户输入和助手回复匹配",
            "✓ 没有敏感信息（密码、密钥等）",
            "✓ 语气和风格一致",
            "✓ 覆盖了主要的使用场景",
            "✓ 数据分布合理（避免类别不平衡）",
            "✓ 去除了重复数据"
        ]

print("微调成本估算：")
num_examples = 1000
avg_tokens = 200
total_tokens = FineTuningBestPractices.calculate_tokens_estimate(num_examples, avg_tokens)
cost = FineTuningBestPractices.estimate_cost(total_tokens)
print(f"样本数: {num_examples}")
print(f"平均tokens: {avg_tokens}")
print(f"总tokens: {total_tokens}")
print(f"预估成本: ${cost:.2f}")

print("\n最少样本建议：")
for task, num in FineTuningBestPractices.get_minimum_examples().items():
    print(f"  {task}: {num}")

print("\n数据质量检查清单：")
for item in FineTuningBestPractices.data_quality_checklist():
    print(f"  {item}")

第六部分：高级应用场景

6.1 多模态文档理解

import base64
from pathlib import Path

class DocumentAnalyzer:
    """多模态文档分析器"""

    def __init__(self, client: OpenAI):
        self.client = client

    def analyze_image_document(
        self,
        image_path: str,
        task: str = "提取文档中的所有文字信息"
    ) -> str:
        """分析图像文档（OCR + 理解）"""
        with open(image_path, 'rb') as f:
            image_data = base64.b64encode(f.read()).decode('utf-8')

        response = self.client.chat.completions.create(
            model="gpt-4-vision-preview",
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": task},
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/jpeg;base64,{image_data}"
                            }
                        }
                    ]
                }
            ],
            max_tokens=2000
        )

        return response.choices[0].message.content

    def analyze_pdf_with_images(
        self,
        pdf_path: str,
        questions: List[str]
    ) -> Dict[str, str]:
        """分析包含图像的PDF（需要先转换为图像）"""
        pass

analyzer = DocumentAnalyzer(client)

result = analyzer.analyze_image_document(
    "invoice.jpg",
    task="""请分析这张发票图像，提取以下信息：
1. 发票编号
2. 开票日期
3. 购买方名称
4. 商品明细（名称、数量、单价、金额）
5. 总金额
6. 税额

请以JSON格式输出。"""
)

6.2 实时对话系统

import asyncio
from openai import AsyncOpenAI
from typing import AsyncGenerator

class RealTimeChat:
    """实时对话系统"""

    def __init__(self, api_key: str, model: str = "gpt-4-turbo"):
        self.client = AsyncOpenAI(api_key=api_key)
        self.model = model
        self.conversation_history = []

    async def stream_chat(self, user_input: str) -> AsyncGenerator[str, None]:
        """流式对话"""
        self.conversation_history.append({
            "role": "user",
            "content": user_input
        })

        stream = await self.client.chat.completions.create(
            model=self.model,
            messages=self.conversation_history,
            stream=True
        )

        full_response = ""
        async for chunk in stream:
            if chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                full_response += content
                yield content

        self.conversation_history.append({
            "role": "assistant",
            "content": full_response
        })

    async def chat_with_interrupt(
        self,
        user_input: str,
        interrupt_check: callable = None
    ):
        """支持中断的对话"""
        self.conversation_history.append({
            "role": "user",
            "content": user_input
        })

        stream = await self.client.chat.completions.create(
            model=self.model,
            messages=self.conversation_history,
            stream=True
        )

        full_response = ""
        async for chunk in stream:
            if interrupt_check and interrupt_check():
                print("\n[用户中断]")
                break

            if chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                full_response += content
                print(content, end="", flush=True)

        print()

        self.conversation_history.append({
            "role": "assistant",
            "content": full_response
        })

async def main():
    chat = RealTimeChat(api_key="your-api-key")

    print("用户: 你好")
    print("AI: ", end="")
    async for chunk in chat.stream_chat("你好"):
        print(chunk, end="", flush=True)
    print()

    print("\n用户: 请介绍一下Python")
    print("AI: ", end="")
    async for chunk in chat.stream_chat("请介绍一下Python"):
        print(chunk, end="", flush=True)
    print()

asyncio.run(main())

6.3 批量处理与并发控制

import asyncio
from openai import AsyncOpenAI
from typing import List, Dict
import time

class BatchProcessor:
    """批量处理器"""

    def __init__(
        self,
        api_key: str,
        model: str = "gpt-3.5-turbo",
        max_concurrent: int = 10,
        requests_per_minute: int = 500
    ):
        self.client = AsyncOpenAI(api_key=api_key)
        self.model = model
        self.max_concurrent = max_concurrent
        self.rpm = requests_per_minute
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.request_times = []

    async def process_single(
        self,
        prompt: str,
        temperature: float = 0.7
    ) -> Dict:
        """处理单个请求"""
        async with self.semaphore:
            await self._rate_limit()

            try:
                response = await self.client.chat.completions.create(
                    model=self.model,
                    messages=[{"role": "user", "content": prompt}],
                    temperature=temperature
                )

                return {
                    "prompt": prompt,
                    "response": response.choices[0].message.content,
                    "success": True
                }
            except Exception as e:
                return {
                    "prompt": prompt,
                    "error": str(e),
                    "success": False
                }

    async def _rate_limit(self):
        """速率限制"""
        now = time.time()

        self.request_times = [t for t in self.request_times if now - t < 60]

        if len(self.request_times) >= self.rpm:
            sleep_time = 60 - (now - self.request_times[0])
            await asyncio.sleep(sleep_time)

        self.request_times.append(now)

    async def process_batch(
        self,
        prompts: List[str],
        temperature: float = 0.7
    ) -> List[Dict]:
        """批量处理"""
        tasks = [
            self.process_single(prompt, temperature)
            for prompt in prompts
        ]

        results = await asyncio.gather(*tasks)

        success_count = sum(1 for r in results if r["success"])
        print(f"处理完成: {success_count}/{len(prompts)} 成功")

        return results

async def batch_example():
    processor = BatchProcessor(
        api_key="your-api-key",
        max_concurrent=5,
        requests_per_minute=60
    )

    prompts = [
        "用一句话介绍Python",
        "用一句话介绍JavaScript",
        "用一句话介绍Go",
        "用一句话介绍Rust",
        "用一句话介绍Java"
    ]

    results = await processor.process_batch(prompts)

    for result in results:
        if result["success"]:
            print(f"问题: {result['prompt']}")
            print(f"回答: {result['response']}\n")
        else:
            print(f"错误: {result['error']}\n")

asyncio.run(batch_example())

第七部分：生产环境最佳实践

7.1 错误处理与重试

import time
from functools import wraps
from openai import (
    APIError,
    APIConnectionError,
    RateLimitError,
    APITimeoutError
)

def retry_with_exponential_backoff(
    max_retries: int = 3,
    initial_delay: float = 1.0,
    exponential_base: float = 2.0,
    errors: tuple = (RateLimitError, APIConnectionError, APITimeoutError)
):
    """指数退避重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            delay = initial_delay

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

                except errors as e:
                    if attempt == max_retries:
                        raise

                    print(f"第 {attempt + 1} 次重试，错误: {type(e).__name__}")
                    time.sleep(delay)
                    delay *= exponential_base

                except Exception as e:
                    raise

            return None

        return wrapper
    return decorator

class RobustLLMClient:
    """健壮的LLM客户端"""

    def __init__(self, client: OpenAI, model: str = "gpt-4-turbo"):
        self.client = client
        self.model = model

    @retry_with_exponential_backoff(max_retries=5)
    def chat(self, messages: list, **kwargs) -> str:
        """带重试的对话"""
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            **kwargs
        )
        return response.choices[0].message.content

    def chat_with_fallback(
        self,
        messages: list,
        fallback_model: str = "gpt-3.5-turbo",
        **kwargs
    ) -> str:
        """带降级的对话"""
        try:
            return self.chat(messages, **kwargs)
        except Exception as e:
            print(f"主模型失败: {e}")
            print(f"降级到 {fallback_model}")

            response = self.client.chat.completions.create(
                model=fallback_model,
                messages=messages,
                **kwargs
            )
            return response.choices[0].message.content

7.2 监控与日志

import logging
from datetime import datetime
from typing import Dict, Any
import json

class LLMLogger:
    """LLM调用日志记录器"""

    def __init__(self, log_file: str = "llm_calls.log"):
        self.log_file = log_file

        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(log_file),
                logging.StreamHandler()
            ]
        )

        self.logger = logging.getLogger(__name__)

    def log_request(
        self,
        model: str,
        messages: list,
        parameters: Dict[str, Any]
    ):
        """记录请求"""
        log_data = {
            "timestamp": datetime.now().isoformat(),
            "type": "request",
            "model": model,
            "message_count": len(messages),
            "parameters": parameters
        }
        self.logger.info(json.dumps(log_data, ensure_ascii=False))

    def log_response(
        self,
        model: str,
        response: str,
        usage: Dict[str, int],
        latency: float
    ):
        """记录响应"""
        log_data = {
            "timestamp": datetime.now().isoformat(),
            "type": "response",
            "model": model,
            "response_length": len(response),
            "usage": usage,
            "latency_seconds": latency
        }
        self.logger.info(json.dumps(log_data, ensure_ascii=False))

    def log_error(self, error: Exception, context: Dict[str, Any] = None):
        """记录错误"""
        log_data = {
            "timestamp": datetime.now().isoformat(),
            "type": "error",
            "error_type": type(error).__name__,
            "error_message": str(error),
            "context": context
        }
        self.logger.error(json.dumps(log_data, ensure_ascii=False))

class MonitoredLLMClient:
    """带监控的LLM客户端"""

    def __init__(self, client: OpenAI, model: str = "gpt-4-turbo"):
        self.client = client
        self.model = model
        self.logger = LLMLogger()
        self.call_count = 0
        self.total_tokens = 0
        self.total_cost = 0.0

    def chat(self, messages: list, **kwargs) -> str:
        """带监控的对话"""
        start_time = time.time()

        self.logger.log_request(self.model, messages, kwargs)

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                **kwargs
            )

            latency = time.time() - start_time
            content = response.choices[0].message.content
            usage = {
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens
            }

            self.logger.log_response(self.model, content, usage, latency)

            self.call_count += 1
            self.total_tokens += usage["total_tokens"]

            return content

        except Exception as e:
            self.logger.log_error(e, {"model": self.model})
            raise

    def get_stats(self) -> Dict[str, Any]:
        """获取统计信息"""
        return {
            "total_calls": self.call_count,
            "total_tokens": self.total_tokens,
            "average_tokens_per_call": self.total_tokens / self.call_count if self.call_count > 0 else 0
        }

7.3 成本优化策略

class CostOptimizer:
    """成本优化器"""

    MODEL_COSTS = {
        "gpt-4-turbo": {"input": 0.01 / 1000, "output": 0.03 / 1000},
        "gpt-4": {"input": 0.03 / 1000, "output": 0.06 / 1000},
        "gpt-3.5-turbo": {"input": 0.0005 / 1000, "output": 0.0015 / 1000},
        "gpt-4o": {"input": 0.005 / 1000, "output": 0.015 / 1000}
    }

    @staticmethod
    def estimate_cost(
        model: str,
        input_tokens: int,
        output_tokens: int
    ) -> float:
        """估算成本"""
        if model not in CostOptimizer.MODEL_COSTS:
            return 0.0

        costs = CostOptimizer.MODEL_COSTS[model]
        return (
            input_tokens * costs["input"] +
            output_tokens * costs["output"]
        )

    @staticmethod
    def optimize_model_selection(
        task_complexity: str,
        estimated_input_tokens: int,
        estimated_output_tokens: int
    ) -> Dict[str, Any]:
        """优化模型选择"""
        recommendations = {
            "simple": ["gpt-3.5-turbo", "gpt-4o-mini"],
            "medium": ["gpt-4o", "gpt-4-turbo"],
            "complex": ["gpt-4", "gpt-4-turbo"]
        }

        models = recommendations.get(task_complexity, recommendations["medium"])

        results = []
        for model in models:
            if model in CostOptimizer.MODEL_COSTS:
                cost = CostOptimizer.estimate_cost(
                    model,
                    estimated_input_tokens,
                    estimated_output_tokens
                )
                results.append({
                    "model": model,
                    "estimated_cost": cost,
                    "cost_per_1k_requests": cost * 1000
                })

        results.sort(key=lambda x: x["estimated_cost"])

        return {
            "recommended": results[0] if results else None,
            "all_options": results
        }

    @staticmethod
    def reduce_token_usage(text: str) -> str:
        """减少token使用（简化文本）"""
        import re

        text = re.sub(r'\s+', ' ', text)

        text = text.strip()

        return text

print("成本优化示例：")
optimizer = CostOptimizer()

result = optimizer.optimize_model_selection(
    task_complexity="medium",
    estimated_input_tokens=500,
    estimated_output_tokens=300
)

print(f"推荐模型: {result['recommended']['model']}")
print(f"预估成本: ${result['recommended']['estimated_cost']:.6f}")
print(f"每1000次请求成本: ${result['recommended']['cost_per_1k_requests']:.2f}")

第八部分：学习路线图

8.1 基础阶段（1-2个月）

目标： 掌握API调用和基本概念

学习内容：

✅ OpenAI API基础（已完成）
✅ 不同模型的连接方式（已完成）
✅ 文本、图像、语音处理（已完成）
⬜ Python异步编程
⬜ API错误处理和重试机制

实践项目：

构建一个简单的聊天机器人
实现一个文档摘要工具
开发一个图像描述生成器

8.2 进阶阶段（2-3个月）

目标： 掌握RAG和Prompt Engineering

学习内容：

⬜ Embeddings深入理解
⬜ 向量数据库（Pinecone、Weaviate、Chroma）
⬜ RAG系统架构
⬜ Prompt Engineering技巧
⬜ Few-shot和Zero-shot学习

实践项目：

构建企业知识库问答系统
开发一个智能客服机器人
实现多文档摘要和问答

8.3 高级阶段（3-6个月）

目标： 掌握Agent开发和模型微调

学习内容：

⬜ Agent架构设计
⬜ 工具调用和Function Calling
⬜ ReAct、Plan-and-Execute等模式
⬜ 模型微调流程
⬜ 数据准备和质量控制

实践项目：

开发一个自动化研究助手
构建一个代码生成和调试Agent
微调一个特定领域的模型

8.4 专家阶段（6个月以上）

目标： 生产级应用和架构设计

学习内容：

⬜ 大规模并发处理
⬜ 成本优化策略
⬜ 监控和日志系统
⬜ 安全和隐私保护
⬜ 多模态高级应用

实践项目：

构建一个企业级AI平台
开发一个多模态内容生成系统
实现一个AI驱动的自动化工作流

第九部分：推荐学习资源

9.1 官方文档

9.2 开源项目

LangChain: LLM应用开发框架
LlamaIndex: 数据框架，专注于RAG
AutoGPT: 自主AI Agent
PrivateGPT: 私有化RAG系统
Chroma: 开源向量数据库

9.3 学习课程

DeepLearning.AI: Andrew Ng的AI课程
Fast.ai: 实用深度学习课程
Stanford CS224N: NLP课程
Hugging Face Course: Transformers课程

9.4 社区资源

OpenAI社区论坛: 官方讨论区
Reddit r/OpenAI: Reddit社区
Discord: 各大开源项目的Discord服务器
GitHub Discussions: 项目讨论区

总结

恭喜你完成了这篇进阶教程！现在你应该已经掌握了：

✅ Embeddings：理解向量表示和语义搜索 ✅ RAG：构建知识库问答系统 ✅ Prompt Engineering：设计高效的提示词 ✅ Agent：开发自主行动的AI助手 ✅ Fine-tuning：定制专属模型 ✅ 生产实践：错误处理、监控、成本优化

下一步建议：