Python + AI 时代来了！这 3 个大模型应用开发方向，薪资翻倍

pip install langchain langchain-openai langchain-community python-dotenv serpapi

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import Tool
from langchain_community.tools import SerpAPIWrapper
import math

# 加载环境变量（需提前在 .env 文件配置 OPENAI_API_KEY 和 SERPAPI_API_KEY）
load_dotenv()

# 1. 定义工具：搜索引擎
search = SerpAPIWrapper(serpapi_api_key=os.getenv("SERPAPI_API_KEY"))

# 2. 定义工具：计算器（计算平方根）
def calculate_square_root(number: float) -> str:
    """计算一个数字的平方根，仅接受数字输入"""
    try:
        result = math.sqrt(number)
        return f"平方根计算结果：{result:.2f}"
    except ValueError:
        return "错误：无法计算负数的平方根"

# 3. 整合工具列表
tools = [
    Tool(
        name="Search",
        func=search.run,
        description="当需要获取最新信息、实时数据、未知事实时使用，比如2025年北京GDP、最新政策等"
    ),
    Tool(
        name="Calculator",
        func=calculate_square_root,
        description="当需要计算平方根时使用，输入必须是数字"
    )
]

# 4. 配置 LLM（使用 GPT-4o，也可替换为 Ollama 部署的 Llama 3）
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY")
)

# 5. 定义 Agent 提示词（ReAct 模式）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手，能使用工具解决问题。请按照以下步骤操作：\n1. 分析问题，判断是否需要使用工具；\n2. 如果需要，选择合适的工具；\n3. 执行工具调用，获取结果；\n4. 基于结果回答原始问题。"),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),  # 保存思考过程
])

# 6. 创建 Agent 并执行
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 测试：问一个需要搜索+计算的问题
if __name__ == "__main__":
    response = agent_executor.invoke({
        "input": "2025年北京市人均GDP是多少万元？请计算这个数值的平方根（保留2位小数）"
    })
    print("\n最终回答：", response["output"])

pip install langchain langchain-openai pypdf chromadb python-dotenv

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA

# 加载环境变量（配置 OPENAI_API_KEY）
load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")

# 1. 加载并解析 PDF 文件
def load_pdf_document(pdf_path):
    """加载 PDF 文件并返回文档对象列表"""
    if not os.path.exists(pdf_path):
        raise FileNotFoundError(f"PDF 文件不存在：{pdf_path}")
    
    # 加载 PDF
    loader = PyPDFLoader(pdf_path)
    documents = loader.load()
    
    # 分割文本（避免超过 LLM 上下文窗口）
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,  # 每个文本块大小
        chunk_overlap=200,  # 块之间重叠，保证上下文连贯
        separators=["\n\n", "\n", "。", "！", "？", "，", ""]  # 中文分割符
    )
    split_docs = text_splitter.split_documents(documents)
    print(f"PDF 解析完成，分割为 {len(split_docs)} 个文本块")
    return split_docs

# 2. 创建向量数据库并存储文本
def create_vector_db(documents, persist_directory="./chroma_db"):
    """将文档存入 Chroma 向量数据库"""
    # 初始化嵌入模型（将文本转为向量）
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    
    # 创建向量库
    vectordb = Chroma.from_documents(
        documents=documents,
        embedding=embeddings,
        persist_directory=persist_directory  # 持久化存储
    )
    vectordb.persist()
    print(f"向量数据库创建完成，存储路径：{persist_directory}")
    return vectordb

# 3. 创建 RAG 问答链
def create_rag_chain(vectordb):
    """创建检索增强生成的问答链"""
    # 初始化 LLM
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    
    # 创建检索器（返回最相关的 3 个文本块）
    retriever = vectordb.as_retriever(search_kwargs={"k": 3})
    
    # 创建 RAG 链
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",  # 将检索到的文本全部传入 LLM
        retriever=retriever,
        return_source_documents=True,  # 返回答案来源
        chain_type_kwargs={
            "prompt": """基于以下提供的上下文回答问题，只使用上下文里的信息，不要编造内容。
如果上下文里没有相关信息，直接说“无法从提供的文档中找到相关答案”。

上下文：
{context}

问题：
{question}

回答："""
        }
    )
    return qa_chain

# 主函数：运行 RAG 问答
if __name__ == "__main__":
    # 替换为你的 PDF 文件路径
    PDF_PATH = "./company_product_manual.pdf"
    
    try:
        # 步骤1：加载并分割 PDF
        docs = load_pdf_document(PDF_PATH)
        
        # 步骤2：创建向量库
        vectordb = create_vector_db(docs)
        
        # 步骤3：创建问答链
        qa_chain = create_rag_chain(vectordb)
        
        # 步骤4：提问并获取答案
        question = "产品A的核心功能有哪些？"
        result = qa_chain.invoke({"query": question})
        
        # 输出结果
        print("\n=== 问答结果 ===")
        print(f"问题：{question}")
        print(f"答案：{result['result']}")
        
        # 输出答案来源（可选）
        print("\n=== 答案来源 ===")
        for i, doc in enumerate(result["source_documents"]):
            print(f"来源 {i+1}：页码 {doc.metadata['page']}，内容片段：{doc.page_content[:100]}...")
    
    except Exception as e:
        print(f"运行出错：{str(e)}")

pip install fastapi uvicorn langchain-openai python-dotenv pydantic

import os
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate

# 加载环境变量
load_dotenv()

# 初始化 FastAPI 应用
app = FastAPI(title="AI 文案生成 SaaS API", version="1.0")

# 解决跨域问题（前端调用接口需要）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境需替换为具体域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 初始化 LLM
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.7,  # 创意性调参，越高越灵活
    api_key=os.getenv("OPENAI_API_KEY")
)

# 定义请求体模型（校验输入参数）
class CopywritingRequest(BaseModel):
    topic: str  # 文案主题
    scene: str  # 场景：wechat（朋友圈）、xiaohongshu（小红书）、gongzhonghao（公众号）
    tone: str = "活泼"  # 语气：活泼/正式/文艺/搞笑
    length: int = 100  # 字数

# 定义不同场景的提示词模板
prompt_templates = {
    "wechat": PromptTemplate(
        template="""请为我生成一条朋友圈文案，要求如下：
主题：{topic}
语气：{tone}
字数：{length}字左右
风格：符合朋友圈社交场景，简洁有吸引力，避免广告感太重。
""",
        input_variables=["topic", "tone", "length"]
    ),
    "xiaohongshu": PromptTemplate(
        template="""请为我生成一篇小红书文案，要求如下：
主题：{topic}
语气：{tone}
字数：{length}字左右
风格：符合小红书平台特性，包含合适的emoji，结尾带相关话题标签（3-5个）。
""",
        input_variables=["topic", "tone", "length"]
    ),
    "gongzhonghao": PromptTemplate(
        template="""请为我生成一篇公众号文章开头文案，要求如下：
主题：{topic}
语气：{tone}
字数：{length}字左右
风格：符合公众号阅读场景，有吸引力，能引导读者继续阅读。
""",
        input_variables=["topic", "tone", "length"]
    )
}

# 定义文案生成接口
@app.post("/api/generate-copywriting", summary="生成AI文案")
async def generate_copywriting(request: CopywritingRequest):
    try:
        # 校验场景参数
        if request.scene not in prompt_templates:
            raise HTTPException(
                status_code=400,
                detail=f"场景参数错误，支持的场景：{list(prompt_templates.keys())}"
            )
        
        # 获取对应场景的提示词模板
        prompt = prompt_templates[request.scene]
        
        # 生成提示词
        prompt_text = prompt.format(
            topic=request.topic,
            tone=request.tone,
            length=request.length
        )
        
        # 调用 LLM 生成文案
        response = llm.invoke(prompt_text)
        
        # 返回结果
        return {
            "code": 200,
            "message": "success",
            "data": {
                "topic": request.topic,
                "scene": request.scene,
                "tone": request.tone,
                "copywriting": response.content
            }
        }
    
    except HTTPException as e:
        raise e
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"服务器内部错误：{str(e)}"
        )

# 健康检查接口
@app.get("/health", summary="健康检查")
async def health_check():
    return {"status": "healthy"}

# 运行服务
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        app="main:app",
        host="0.0.0.0",
        port=8000,
        reload=True  # 开发模式自动重载
    )

{
  "topic": "春季新品咖啡上市",
  "scene": "xiaohongshu",
  "tone": "活泼",
  "length": 200
}