AI 应用开发实战：从 Prompt 调用到 RAG、Agent 与可观测工程

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ.get("LLM_BASE_URL", "https://api.openai.com/v1"),
)

def chat(prompt: str, system: str = "你是一名严谨的助手。",
         model: str = "gpt-4o-mini", temperature: float = 0.3) -> str:
    resp = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": prompt},
        ],
        temperature=temperature,
        stream=False,
    )
    return resp.choices[0].message.content

SYSTEM = """你是资深技术编辑。
规则：
1. 只根据用户提供的材料回答；
2. 材料没有依据时回答“未找到相关依据”；
3. 输出 Markdown，包含结论、要点、风险三部分。"""

def summarize(text: str) -> str:
    return chat(f"材料如下：\n{text}", system=SYSTEM)

import json

SCHEMA = """只输出 JSON，格式：
{"title": "", "tags": [], "risk": "low|medium|high"}"""

def extract(text: str) -> dict:
    content = chat(text, system=SCHEMA)
    return json.loads(content)  # 生产环境务必用 Pydantic 校验

def rag_answer(question: str, top_k: int = 4) -> str:
    docs = vector_search(question, top_k=top_k)
    context = "\n\n".join(d["content"] for d in docs)
    prompt = f"""只根据以下资料回答问题。
资料没有依据时回答“未找到相关依据”，不要猜测。
资料：
{context}
问题：{question}"""
    return chat(prompt)

def agent_loop(task: str, tools: dict, max_steps: int = 8) -> str:
    messages = [{"role": "user", "content": task}]
    seen = set()

    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=[t["schema"] for t in tools.values()],
        )
        msg = resp.choices[0].message

        if not msg.tool_calls:
            return msg.content

        for call in msg.tool_calls:
            name = call.function.name
            args = json.loads(call.function.arguments)

            sig = (name, json.dumps(args, sort_keys=True))
            if sig in seen:                 # 循环检测
                return "检测到重复调用，已中止。"
            seen.add(sig)

            result = tools[name]["fn"](**args)  # 执行工具
            messages.append(msg)
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": json.dumps(result, ensure_ascii=False),
            })

    return "达到最大步数，任务未完成。"

import time
from dataclasses import dataclass

@dataclass
class Metric:
    model: str
    prompt_tokens: int
    completion_tokens: int
    latency_ms: float
    success: bool

def call_with_metric(prompt: str) -> tuple[str, Metric]:
    start = time.time()
    try:
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
        )
        u = resp.usage
        m = Metric("gpt-4o-mini", u.prompt_tokens,
                   u.completion_tokens, (time.time()-start)*1000, True)
        return resp.choices[0].message.content, m
    except Exception:
        m = Metric("gpt-4o-mini", 0, 0, (time.time()-start)*1000, False)
        raise