Python + LLM 实战：OpenAI API 集成与 Agent 开发

pip install openai python-dotenv

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# 基础调用
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个 helpful 助手"},
        {"role": "user", "content": "解释 Python 的 GIL 是什么"}
    ],
    temperature=0.7,
    max_tokens=500
)
print(response.choices[0].message.content)

# 流式响应（实时输出，适合聊天界面）
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "写一首关于编程的诗"}],
    stream=True
)
for chunk in stream:
    content = chunk.choices[0].delta.content
    if content:
        print(content, end="", flush=True)

# 阿里云通义千问
from openai import OpenAI
qwen = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

# 百度文心
# 火山引擎豆包
# 智谱 GLM
# 统一使用 OpenAI SDK 兼容接口

import json

def get_weather(city: str) -> str:
    """查询城市天气（模拟）"""
    weather_data = {
        "北京": "晴天 25°C",
        "上海": "多云 22°C",
        "深圳": "小雨 28°C"
    }
    return weather_data.get(city, "未知城市")

# 定义工具
functions = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "获取指定城市的天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称"}
            },
            "required": ["city"]
        }
    }
}]

# 调用
messages = [{"role": "user", "content": "北京今天天气怎么样？"}]
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=functions
)

# 处理工具调用
message = response.choices[0].message
if message.tool_calls:
    tool_call = message.tool_calls[0]
    function_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)
    
    # 执行函数
    result = get_weather(**arguments)
    
    # 将结果返回给 LLM
    messages.append(message)  # AI 的 tool_call
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": result
    })
    
    final = client.chat.completions.create(
        model="gpt-4o", messages=messages
    )
    print(final.choices[0].message.content)
    # 输出: "北京今天是晴天，气温 25°C，适合外出！"

# 简化版 RAG 实现
from typing import List
import numpy as np

class SimpleRAG:
    def __init__(self, client):
        self.client = client
        self.documents: List[str] = []
        self.embeddings: List[List[float]] = []
    
    def add_document(self, text: str):
        """添加文档并计算嵌入"""
        self.documents.append(text)
        resp = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        self.embeddings.append(resp.data[0].embedding)
    
    def search(self, query: str, top_k: int = 3) -> List[str]:
        """语义搜索最相关文档"""
        resp = self.client.embeddings.create(
            model="text-embedding-3-small", input=query
        )
        query_emb = np.array(resp.data[0].embedding)
        
        # 计算余弦相似度
        scores = []
        for emb in self.embeddings:
            sim = np.dot(query_emb, emb) / (np.linalg.norm(query_emb) * np.linalg.norm(emb))
            scores.append(sim)
        
        top_indices = np.argsort(scores)[-top_k:][::-1]
        return [self.documents[i] for i in top_indices]
    
    def ask(self, query: str) -> str:
        """RAG 问答"""
        contexts = self.search(query)
        context_text = "\n".join(contexts)
        
        prompt = f"""基于以下上下文回答问题：

上下文：
{context_text}

问题：{query}

请基于上下文回答，如果无法回答请说明。"""
        
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content

# 使用
rag = SimpleRAG(client)
rag.add_document("Python 3.12 于 2023 年 10 月发布，引入了改进的错误消息、f-string 调试语法等")
rag.add_document("FastAPI 是一个现代 Python Web 框架，基于 Starlette 和 Pydantic")
rag.add_document("Docker 容器化可以将 Python 应用打包成镜像，实现环境一致性")

print(rag.ask("FastAPI 是什么？"))

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.text_splitter import RecursiveCharacterTextSplitter

# LLM
llm = ChatOpenAI(model="gpt-4o")

# 文档处理
text = """Python 是一种解释型、高级、通用编程语言。Python 的设计哲学强调代码的可读性和简洁的语法。
它的语言结构以及面向对象的方法旨在帮助程序员为小型和大型项目编写清晰、合乎逻辑的代码。"""

splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20)
docs = splitter.create_documents([text])

# 向量化存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(docs, embeddings)

# RAG 链
qa = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(),
    chain_type="stuff"
)

result = qa.invoke({"query": "Python 是什么类型的语言？"})
print(result["result"])