大模型 Python 工程：从 API 调用到生产级 LLM 应用的系统实践

原创

资源大佬 jzit-top

发布于 2026-09-17 19:19:55

1270

摘要

大模型（LLM）正在重塑软件架构：从确定性函数调用，转向概率性生成、上下文工程、检索增强与智能体编排。Python 凭借生态优势，成为 LLM 工程的事实标准语言。但“会调 API”与“能上生产”之间，隔着提示词工程、结构化输出、RAG、微调、推理优化、评估、可观测性、成本与安全治理。本文从专业角度，系统梳理大模型 Python 工程的完整链路，并给出可直接运行的代码示例。

关键词：大模型；Python；RAG；LoRA；vLLM；结构化输出；Agent；评估；可观测性

1. LLM 工程的技术栈分层

一个生产级 LLM 应用通常包含：

应用层：对话、搜索、写作、Agent、Copilot
编排层：Prompt、Chain、Graph、Tool Calling、Memory
能力层：RAG、微调、函数调用、多模态
推理层：OpenAI API / vLLM / TGI / llama.cpp / Ollama
数据层：向量库、文档解析、Embedding、缓存
工程层：评估、监控、限流、成本、安全、灰度

Python 在每个层次都有成熟库：

API 与编排：openai、anthropic、langchain、llama-index、dspy；
本地推理：transformers、vllm、llama-cpp-python、ollama；
微调：peft、trl、accelerate、bitsandbytes；
向量与检索：faiss、chromadb、qdrant-client、pgvector；
评估：ragas、deepeval、promptfoo；
服务化：FastAPI、uvicorn、streamlit、gradio。

核心原则：先做对，再做快，最后做便宜。

2. 最基础的调用：从同步到异步

2.1 同步调用

from openai import OpenAI

client = OpenAI(api_key="sk-...", base_url="https://api.openai.com/v1")

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "你是一个严谨的中文技术助手。"},
        {"role": "user", "content": "用一句话解释什么是向量数据库。"},
    ],
    temperature=0.2,
    max_tokens=256,
)

print(resp.choices[0].message.content)

2.2 异步与并发

LLM 调用是 I/O 密集型，异步能显著提升吞吐。

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def ask(prompt: str) -> str:
    resp = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
    )
    return resp.choices[0].message.content

async def main():
    prompts = ["解释 RAG", "解释 LoRA", "解释 KV Cache", "解释量化"]
    results = await asyncio.gather(*(ask(p) for p in prompts))
    for p, r in zip(prompts, results):
        print(p, "=>", r)

asyncio.run(main())

2.3 流式输出

def stream_chat(prompt: str):
    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

for token in stream_chat("写一首关于编译器的五言绝句"):
    print(token, end="", flush=True)

工程要点：

统一超时、重试、退避；
记录 token 用量与成本；
并发上限受速率限制约束；
流式场景要处理中断与拼接。

3. 结构化输出：让 LLM 返回可解析数据

LLM 输出是自然语言，业务需要结构化数据。三种主流方案：

JSON mode；
Function Calling / Tool Calling；
Pydantic 校验与修复。

from pydantic import BaseModel, Field, ValidationError
import json

class Invoice(BaseModel):
    invoice_no: str = Field(description="发票号")
    amount: float = Field(description="金额，单位元")
    currency: str = Field(default="CNY")
    items: list[str] = Field(default_factory=list)

def extract_invoice(text: str) -> Invoice:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": "你只输出 JSON，字段与 schema 一致。"},
            {"role": "user", "content": f"抽取发票信息：\n{text}\n"
                                        f"schema={Invoice.model_json_schema()}"},
        ],
        temperature=0,
    )
    raw = resp.choices[0].message.content
    try:
        return Invoice.model_validate_json(raw)
    except ValidationError as e:
        # 修复策略：让模型自我修正一次
        fix = client.chat.completions.create(
            model="gpt-4o-mini",
            response_format={"type": "json_object"},
            messages=[
                {"role": "system", "content": "修正以下 JSON 使其符合 schema。"},
                {"role": "user", "content": f"schema={Invoice.model_json_schema()}\n"
                                            f"json={raw}\nerror={e}"},
            ],
            temperature=0,
        )
        return Invoice.model_validate_json(fix.choices[0].message.content)

Function Calling 更适合需要调用外部工具的场景：

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询城市天气",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "unit": {"type": "string", "enum": ["c", "f"]}
            },
            "required": ["city"]
        }
    }
}]

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
    tools=tools,
    tool_choice="auto",
)

生产建议：永远不要让 LLM 直接执行 SQL、Shell 或文件删除。工具调用必须经过白名单、参数校验和权限控制。

4. 提示词工程：从技巧到工程

提示词不是“咒语”，而是上下文工程。可复用结构：

角色：你是谁
任务：要做什么
约束：不能做什么
输入：数据与格式
输出：格式与示例
评判：如何判断好坏

SYSTEM = """你是电商客服助手。
规则：
1. 只基于给定知识库回答，不知道就说“我需要转人工”。
2. 不承诺价格、库存、时效以外的内容。
3. 输出使用简体中文，不超过 120 字。
"""

def build_prompt(question: str, context: str) -> list[dict]:
    return [
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": f"知识库：\n{context}\n\n用户问题：{question}"},
    ]

进阶方法：

Few-shot：给出输入输出示例；
Chain-of-Thought：让模型分步推理；
Self-Consistency：多路采样投票；
ReAct：推理与行动交替；
Reflexion：自我反思与修正；
DSPy：将提示词作为可优化参数。

提示词应纳入版本管理，并配合评估集回归。

5. RAG：检索增强生成

RAG 是当前最实用的 LLM 落地范式：用检索提供事实，用生成组织语言。

5.1 最小 RAG 实现

import numpy as np
from openai import OpenAI

client = OpenAI()

def embed(texts: list[str]) -> np.ndarray:
    resp = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return np.array([d.embedding for d in resp.data], dtype=np.float32)

class VectorStore:
    def __init__(self):
        self.docs: list[str] = []
        self.mat: np.ndarray | None = None

    def add(self, docs: list[str]):
        self.docs.extend(docs)
        vecs = embed(docs)
        self.mat = vecs if self.mat is None else np.vstack([self.mat, vecs])

    def search(self, query: str, top_k: int = 3) -> list[str]:
        q = embed([query])[0]
        sims = self.mat @ q / (np.linalg.norm(self.mat, axis=1) * np.linalg.norm(q) + 1e-8)
        idx = np.argsort(-sims)[:top_k]
        return [self.docs[i] for i in idx]

store = VectorStore()
store.add([
    "退货政策：签收后 7 天内可无理由退货，商品需不影响二次销售。",
    "发货时效：现货商品 48 小时内发出，预售以详情页为准。",
    "发票：支持电子普票，下单时勾选，发货后 3 个工作日开出。",
])

def rag_answer(question: str) -> str:
    ctx = "\n".join(store.search(question, top_k=3))
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=build_prompt(question, ctx),
        temperature=0.2,
    )
    return resp.choices[0].message.content

print(rag_answer("我买的衣服不合适能退吗？"))

5.2 生产级 RAG 的关键环节

文档解析：PDF、HTML、Word、表格、图片 OCR；
切分策略：固定长度、递归、语义、按标题层级；
元数据：来源、时间、权限、版本；
混合检索：向量 + BM25 + 关键词；
重排序：Cross-Encoder Rerank；
查询改写：HyDE、多查询、子问题分解；
上下文压缩：只保留相关片段；
引用与溯源：回答附带来源；
权限过滤：不同用户检索不同文档；
评估：命中率、忠实度、答案相关性。

# 伪代码：混合检索 + 重排
candidates = vector_search(q, top_k=20) + bm25_search(q, top_k=20)
reranked = rerank(q, candidates, top_k=5)
answer = generate(q, reranked)

RAG 失败模式：

检索不到：切分、Embedding、查询改写问题；
检索到但答错：上下文噪声、模型幻觉；
答非所问：提示词与格式约束不足；
权限泄露：缺少元数据过滤。

6. 微调：LoRA 与指令微调

当 RAG 无法满足风格、格式、领域术语或成本要求时，考虑微调。

6.1 LoRA 微调示例

# pip install transformers peft trl datasets accelerate bitsandbytes
import torch
from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model
from trl import SFTTrainer

model_name = "Qwen/Qwen2.5-1.5B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    trust_remote_code=True,
)

lora = LoraConfig(
    r=16, lora_alpha=32, lora_dropout=0.05,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
    task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora)

dataset = load_dataset("json", data_files="sft.jsonl", split="train")

args = TrainingArguments(
    output_dir="./out",
    per_device_train_batch_size=2,
    gradient_accumulation_steps=8,
    learning_rate=1e-4,
    num_train_epochs=3,
    bf16=True,
    logging_steps=10,
    save_strategy="epoch",
)

trainer = SFTTrainer(
    model=model,
    args=args,
    train_dataset=dataset,
    tokenizer=tokenizer,
    dataset_text_field="text",
    max_seq_length=1024,
)
trainer.train()
model.save_pretrained("./lora-adapter")

6.2 微调决策

场景	优先方案
知识更新频繁	RAG
格式与风格固定	微调或提示词
领域术语多	RAG + 微调
成本敏感	小模型微调
数据不足	提示词 + RAG
需要引用溯源	RAG

微调不是知识注入的首选，RAG 更适合事实更新。微调适合“行为塑造”。

7. 本地推理与部署

7.1 vLLM 高吞吐推理

# pip install vllm
from vllm import LLM, SamplingParams

llm = LLM(model="Qwen/Qwen2.5-7B-Instruct", tensor_parallel_size=1)
params = SamplingParams(temperature=0.2, max_tokens=256)

outputs = llm.generate(["用一句话解释 KV Cache。"], params)
for o in outputs:
    print(o.outputs[0].text)

7.2 FastAPI 服务化

from fastapi import FastAPI
from pydantic import BaseModel
from vllm import LLM, SamplingParams

app = FastAPI()
llm = LLM(model="Qwen/Qwen2.5-7B-Instruct")
params = SamplingParams(temperature=0.2, max_tokens=512)

class ChatReq(BaseModel):
    prompt: str

@app.post("/chat")
def chat(req: ChatReq):
    out = llm.generate([req.prompt], params)[0]
    return {"text": out.outputs[0].text}

7.3 量化

GPTQ / AWQ：权重量化，适合 GPU；
GGUF：CPU/Metal，适合 llama.cpp；
bitsandbytes：4bit/8bit 加载；
FP8：Hopper 及以后架构。

量化降低显存与成本，但可能损失质量，需评估。

7.4 推理优化要点

KV Cache：避免重复计算；
PagedAttention：vLLM 核心；
Continuous Batching：动态批处理；
Prefix Caching：复用系统提示；
Speculative Decoding：小模型草稿；
Tensor Parallel：多卡切分。

8. Agent：工具调用与编排

Agent 的本质是 LLM + 工具 + 记忆 + 循环。

import json

TOOLS = {
    "search": lambda q: f"[搜索结果] {q} 的相关资料...",
    "calculator": lambda expr: str(eval(expr, {"__builtins__": {}}, {})),
}

def run_agent(question: str, max_steps: int = 5) -> str:
    messages = [
        {"role": "system", "content":
            "你可以调用工具：search(query), calculator(expr)。"
            "需要时输出 JSON：{\"tool\":\"...\",\"args\":{...}}，否则直接回答。"},
        {"role": "user", "content": question},
    ]
    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o-mini", messages=messages, temperature=0
        )
        content = resp.choices[0].message.content.strip()
        try:
            call = json.loads(content)
            if "tool" in call:
                fn = TOOLS[call["tool"]]
                result = fn(**call["args"])
                messages.append({"role": "assistant", "content": content})
                messages.append({"role": "user", "content": f"工具结果：{result}"})
                continue
        except json.JSONDecodeError:
            pass
        return content
    return "达到最大步数限制"

生产级 Agent 需要：

工具白名单与权限；
参数校验；
超时与重试；
最大步数与成本上限；
人工确认高风险操作；
全链路日志与回放。

LangGraph、AutoGen、CrewAI 等框架提供更完整的编排能力，但核心仍是状态机与工具协议。

9. 评估：LLM 应用的生死线

没有评估，就没有优化。LLM 评估分三层：

单元评估：单条提示、单个函数；
组件评估：检索、重排、生成；
端到端评估：任务成功率、用户满意度。

常用指标：

忠实度 Faithfulness；
答案相关性 Answer Relevancy；
上下文精确率/召回率；
格式合规率；
延迟、成本、token 用量。

# 伪代码：RAG 评估
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

result = evaluate(
    dataset=eval_dataset,
    metrics=[faithfulness, answer_relevancy, context_precision],
)
print(result)

评估集构建：

真实用户问题；
人工标注答案；
边界与对抗样本；
回归集持续积累。

LLM-as-Judge 可扩展评估，但需校准偏差，最好与人工评估结合。

10. 可观测性与成本治理

生产级 LLM 应用必须回答：

每次调用用了多少 token、多少钱；
哪个环节慢；
失败率与重试率；
用户满意度；
是否有安全事件。

import time, logging

logger = logging.getLogger("llm")

def traced_chat(messages, **kwargs):
    start = time.time()
    resp = client.chat.completions.create(messages=messages, **kwargs)
    usage = resp.usage
    logger.info({
        "model": kwargs.get("model"),
        "prompt_tokens": usage.prompt_tokens,
        "completion_tokens": usage.completion_tokens,
        "latency_ms": int((time.time() - start) * 1000),
    })
    return resp

成本优化：

缓存：精确缓存 + 语义缓存；
模型路由：简单问题用小模型；
提示压缩：减少上下文；
批处理；
流式输出提升感知速度；
限制 max_tokens。

安全治理：

输入输出审核；
提示注入防护；
敏感信息脱敏；
工具调用权限；
审计日志。

11. 常见反模式

把 LLM 当数据库，事实靠模型记忆；
不做评估就上线；
提示词硬编码在业务代码；
无 token 与成本监控；
无超时、无重试、无降级；
让模型直接执行危险操作；
微调解决知识更新；
RAG 只做向量检索，不做重排与过滤；
忽略权限与数据隔离；
追求最大模型，忽略成本与延迟。

12. 结论

大模型 Python 工程，不是简单的 API 调用，而是一套围绕概率性组件的系统工程。它要求我们掌握异步调用与流式输出，用结构化输出连接业务，用提示词工程控制行为，用 RAG 提供事实，用微调塑造风格，用 vLLM 与量化优化推理，用 Agent 扩展能力，用评估驱动迭代，用可观测性与成本治理保障生产。真正专业的 LLM 应用，是在质量、延迟、成本与安全之间持续权衡的结果。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python