大模型应用开发：从 Prompt 工程到 Agent 系统的架构跃迁

“你必须只返回一个合法的 JSON 对象，不要包含任何解释、前后缀或 Markdown 标记。”

import outlines

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer", "minimum": 0, "maximum": 120},
        "hobbies": {"type": "array", "items": {"type": "string"}}
    },
    "required": ["name", "age"]
}

model = outlines.models.transformers("Qwen/Qwen2-7B-Instruct")
generator = outlines.generate.json(model, schema)

result = generator("提取以下文本中的姓名、年龄和爱好：张三，25岁，喜欢编程和游泳。")
print(result)  # {'name': '张三', 'age': 25, 'hobbies': ['编程', '游泳']}

import json
import re

def safe_json_parse(text: str) -> dict:
    # 1. 尝试直接解析
    try:
        return json.loads(text)
    except:
        pass
    # 2. 移除 Markdown 代码块
    cleaned = re.sub(r'```json\s*|\s*```', '', text)
    # 3. 使用正则提取第一个完整 JSON 对象
    match = re.search(r'\{.*\}', cleaned, re.DOTALL)
    if match:
        try:
            return json.loads(match.group())
        except:
            pass
    # 4. 利用 json-repair 修复常见错误（如末尾逗号）
    from json_repair import repair_json
    return json.loads(repair_json(cleaned))

from elasticsearch import Elasticsearch
import faiss
import numpy as np

def multi_recall(query: str, top_k: int = 10):
    # 1. BM25 召回（ES）
    es_body = {
        "query": {
            "multi_match": {"query": query, "fields": ["title^3", "content"], "operator": "or"}
        },
        "size": top_k * 2
    }
    es_hits = es.search(index="docs", body=es_body)["hits"]["hits"]
    
    # 2. 向量召回（FAISS）
    query_vec = embed_model.encode(query).reshape(1, -1)
    distances, indices = faiss_index.search(query_vec, top_k * 2)
    vec_docs = [all_docs[i] for i in indices[0]]
    
    # 3. 合并去重 + RRF（Reciprocal Rank Fusion）重排
    scores = {}
    for rank, hit in enumerate(es_hits):
        doc_id = hit["_id"]
        scores[doc_id] = scores.get(doc_id, 0) + 1 / (60 + rank)
    for rank, doc in enumerate(vec_docs):
        doc_id = doc["id"]
        scores[doc_id] = scores.get(doc_id, 0) + 1 / (60 + rank)
    
    sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
    return [doc_map[d] for d, _ in sorted_docs]

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('BAAI/bge-reranker-v2-m3', max_length=512)

def rerank_docs(query: str, docs: list, top_k: int = 5):
    pairs = [[query, d["content"]] for d in docs]
    scores = reranker.predict(pairs)  # 返回相关度分数
    sorted_idx = np.argsort(scores)[::-1]
    return [docs[i] for i in sorted_idx[:top_k]]

def compress_docs(docs, max_tokens=2000):
    # 将每个文档用 LLM 生成一句关键句（利用较小的模型，如 GPT-3.5-turbo）
    summaries = []
    for d in docs:
        prompt = f"用一句话概括以下内容：{d['content'][:300]}"
        summary = small_llm.generate(prompt)
        summaries.append(summary)
    # 将摘要拼接，并保证总长度不超过 max_tokens
    return " ".join(summaries)[:max_tokens]

tools = [
    {
        "type": "function",
        "function": {
            "name": "search_flights",
            "description": "查询航班信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "origin": {"type": "string"},
                    "destination": {"type": "string"},
                    "date": {"type": "string", "format": "date"}
                },
                "required": ["origin", "destination"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询目的地天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"}
                },
                "required": ["city"]
            }
        }
    }
]

class SimpleAgent:
    def __init__(self, model, tools, max_iter=5):
        self.model = model
        self.tools = tools
        self.max_iter = max_iter
        self.memory = []  # 对话历史

    def run(self, user_query):
        self.memory.append({"role": "user", "content": user_query})
        for _ in range(self.max_iter):
            response = self.model.chat(self.memory, tools=self.tools)
            # 检查是否要求调用工具
            if response.get("tool_calls"):
                for call in response["tool_calls"]:
                    tool_name = call["function"]["name"]
                    args = json.loads(call["function"]["arguments"])
                    result = self.execute_tool(tool_name, args)
                    self.memory.append({
                        "role": "tool",
                        "tool_call_id": call["id"],
                        "content": json.dumps(result)
                    })
            else:
                # 没有工具调用，直接返回最终答案
                self.memory.append({"role": "assistant", "content": response["content"]})
                return response["content"]
        return "任务执行超时，请简化需求。"

user_memories = vector_db.search(user_id, top_k=5)  # 返回偏好条目
memory_prompt = "该用户的历史偏好：" + "；".join([m["content"] for m in user_memories])
system_prompt = f"你是一个智能助手。{memory_prompt}。请根据用户当前需求提供个性化服务。"

class Router:
    def __init__(self):
        self.classifier = pipeline("text-classification", model="small-intent-model")
    
    def route(self, query):
        label = self.classifier(query)[0]["label"]
        if label in ["问候", "常识", "翻译"]:
            return "qwen2-7b"
        else:
            return "qwen2-72b"

# vLLM 支持推测解码
llm = LLM(model="meta-llama/Llama-2-70b-chat",
          speculative_model="meta-llama/Llama-2-7b-chat",
          num_speculative_tokens=5)  # 每次预测 5 个候选

def evaluate_answer(query, answer, reference=None):
    judge_prompt = f"""你是一个严格的评审专家。请对以下回答进行评分（1-10分）：
    问题：{query}
    回答：{answer}
    参考：{reference or "无"}
    请返回 JSON：{{"score": 8, "reason": "..."}}"""
    return llm_generate(judge_prompt, model="gpt-4")