大模型应用开发 · 动手实验室

前沿技术 · 动手实验室

上一篇《大模型应用开发》讲了方法论，这一篇把它变成能跑的代码。四个实验：最小对话、Function Calling、最小 RAG、最小 ReAct Agent。标注「需 API Key」的实验用 OpenAI 兼容接口（通义/DeepSeek 同理，换 BASE 和 model 即可）；最小 RAG 的检索部分纯标准库、可离线跑。先把一个跑通，再回头看概念，比只读快得多。

一

实验一：最小对话

Minimal Chat · 需 API Key

一个 HTTP 请求就能调大模型。下面用 OpenAI 兼容接口；换通义/DeepSeek 只改 BASE 和 model。Key 从环境变量读，别硬编码。

lab1_chat.py

import os, requests

API_KEY = os.environ["OPENAI_API_KEY"]        # 或通义/DeepSeek 的 key
BASE    = "https://api.openai.com/v1"        # 兼容端点换对应网关

resp = requests.post(
    f"{BASE}/chat/completions",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={
        "model": "gpt-4o-mini",                # 换成你用的模型名
        "messages": [
            {"role": "system", "content": "你是严谨的助手，只输出 JSON。"},
            {"role": "user",  "content": "订单 ORD-1001 状态是什么？"},
        ],
        "temperature": 0.2,
    },
    timeout=30,
)
print(resp.json()["choices"][0]["message"]["content"])

运行前

export OPENAI_API_KEY="sk-xxx"，再 pip install requests。网络/配额/模型名报错优先查这三项。

二

实验二：Function Calling

Let the Model Call Your Function · 需 API Key

模型返回 tool_calls 时不执行，你把结果作为 role:"tool" 消息回灌，再请求一次拿最终答案。执行权永远在你本地。

lab2_tools.py

import os, requests, json

API_KEY = os.environ["OPENAI_API_KEY"]
BASE    = "https://api.openai.com/v1"

tools = [{
    "type": "function",
    "function": {
        "name": "get_order",
        "description": "按订单号查询金额与状态",
        "parameters": {"type": "object",
            "properties": {"order_id": {"type": "string"}},
            "required": ["order_id"]},
    },
}]

def run_local(name, args):
    # 执行权在你手里；这里用假数据演示，生产查库/调内部 API
    if name == "get_order":
        return {"order_id": args["order_id"], "amount": 128.5, "status": "paid"}
    return {"error": "unknown tool"}

messages = [{"role": "user", "content": "查一下订单 ORD-1001 多少钱？"}]
r = requests.post(f"{BASE}/chat/completions",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={"model": "gpt-4o-mini", "messages": messages, "tools": tools, "temperature": 0},
    timeout=30).json()

msg = r["choices"][0]["message"]
if msg.get("tool_calls"):
    for tc in msg["tool_calls"]:
        fn = tc["function"]
        result = run_local(fn["name"], json.loads(fn["arguments"]))
        messages.append(msg)                        # 模型原样回传
        messages.append({"role": "tool", "tool_call_id": tc["id"],
                         "content": json.dumps(result, ensure_ascii=False)})  # 回灌结果
    final = requests.post(f"{BASE}/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"model": "gpt-4o-mini", "messages": messages}, timeout=30).json()
    print(final["choices"][0]["message"]["content"])

论记住三步

① 带 tools 请求；② 若返回 tool_calls，本地执行并把结果以 role:"tool" 回灌；③ 再请求一次拿最终答案。工具执行务必校验参数与权限。

三

实验三：最小 RAG（检索可离线跑）

Minimal RAG · 检索离线 / 生成需 Key

RAG 的"检索"部分纯标准库即可跑通。下面用玩具 embedding（词频向量）演示"切片→向量化→余弦检索"的机制；真实场景把 toy_embed 换成 Embedding 模型即可，检索逻辑不变。

lab3_rag.py（前两段可离线运行）

import os, requests

docs = [
    "退款需要在订单支付后 7 天内申请。",
    "会员等级根据累计消费金额自动升级。",
    "物流异常时由仓库侧发起补发。",
]

def toy_embed(text):
    # 教学占位：真实用 Embedding 模型（维度 768/1024/1536）。
    # 词频向量仅用于跑通"检索"机制，语义能力很弱。
    vec = [0.0] * 50
    for ch in text:
        vec[ord(ch) % 50] += 1.0
    norm = sum(v * v for v in vec) ** 0.5
    return [v / (norm + 1e-9) for v in vec]

def cosine(a, b):
    return sum(x * y for x, y in zip(a, b))

doc_vecs = [toy_embed(d) for d in docs]

def retrieve(query, k=1):
    q = toy_embed(query)
    sims = [cosine(q, v) for v in doc_vecs]
    idx = sorted(range(len(sims)), key=lambda i: sims[i], reverse=True)[:k]
    return [docs[i] for i in idx]

top = retrieve("怎么退款")
print("检索到的片段:", top)     # 离线即可看到结果

# ---- 需要 Key：把片段拼进 prompt 让模型基于材料回答 ----
API_KEY = os.environ.get("OPENAI_API_KEY")
if API_KEY:
    prompt = f"只依据材料回答并标注来源：\n材料：{top[0]}\n问题：怎么退款？"
    r = requests.post("https://api.openai.com/v1/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": prompt}]},
        timeout=30).json()
    print(r["choices"][0]["message"]["content"])
else:
    print("(未设 OPENAI_API_KEY，跳过模型调用；检索部分已离线跑通)")

玩具 embedding 不是真语义

词频向量只能演示流程，分不清"退款"和"存货"的细微语义。生产换成真实 Embedding 模型（如 text-embedding-3-small / bge / m3e），检索质量才够用。检索逻辑（余弦 + Top-K）保持不变。

四

实验四：最小 ReAct Agent 循环

Minimal ReAct Loop · 离线可跑

Agent = 循环（思考→调工具→看结果）。下面用"脚本化假模型"演示循环结构，无需 API Key 即可跑通机制；生产环境把 script 换成真实模型调用，解析它返回的 thought/action，并加 max_steps 防止死循环。

lab4_agent.py

def calculator(expr):
    return str(eval(expr, {"__builtins__": {}}, {}))   # 演示用；生产要更严的沙箱

tools = {"calculator": calculator}

# 脚本化的"模型"输出（生产换成真实 LLM 的 thought/action 解析）
script = [
    "thought: 先算 23*17\naction: calculator\naction_input: 23 * 17",
    "observation: 391\nthought: 已有结果\nfinal_answer: 391",
]

scratchpad = "问题：23 乘以 17 是多少？\n"
for i, out in enumerate(script):
    print(f"[step {i}] {out}")
    scratchpad += out + "\n"
    if out.startswith("final_answer"):
        print("答案:", out.split(":", 1)[1].strip()); break
    if out.startswith("action:"):
        name = out.split(":", 1)[1].strip()
        arg  = script[i].split("action_input:", 1)[1].strip()
        print(f"  -> 执行 {name}({arg}) = {tools[name](arg)}")

论真 Agent 长这样

把 script 换成真实模型调用：每轮让模型输出 thought + 是否 action；有 action 就本地执行工具、把 observation 写回 scratchpad；循环直到 final_answer 或达到 max_steps。多工具、多步、human-in-the-loop 都是在這个骨架上加。

四个实验的关系

实验一打底对话；实验二让模型接真实世界；实验三喂私有知识；实验四把"多步自主"串成循环。跑通这四个，你就握住了大模型应用开发的全部骨架——框架只是帮你省胶水代码。

结

小结与下一步

Wrap-up

方法论（见 tech-llm-app.html）讲"为什么"，本篇讲"怎么跑"。下一步建议：把实验三的玩具 embedding 换成真实模型、把实验四的脚本换成真 LLM、给工具加参数校验与超时，再用评测集测忠实度。一行行跑通，比看十篇文章管用。

自测

Q1. 哪个实验能完全离线跑？

参考答案

实验三的"检索"和实验四的"循环"都可离线；实验一/二需要 API Key 才能拿到模型回答。

Q2. RAG 里 toy_embed 能直接上生产吗？

参考答案

不能。它只演示流程，没有语义能力。生产必须换成真实 Embedding 模型，检索逻辑（余弦+Top-K）不变。