从Prompt到生产：AI应用定制化开发与Vibe Coding工程实战

from pydantic import BaseModel, Field
from openai import OpenAI
from typing import List, Optional
import json

# 定义业务定制化Schema（AI必须严格遵循此结构生成）
class APIEndpoint(BaseModel):
    path: str = Field(description="RESTful路径，如 /api/v1/orders")
    method: str = Field(description="HTTP方法，仅允许 GET/POST/PUT/DELETE")
    query_params: Optional[List[str]] = Field(default=[], description="查询参数列表")
    request_model: str = Field(description="请求体的Pydantic类名")
    response_model: str = Field(description="响应体的Pydantic类名")
    business_logic: str = Field(description="该接口的核心业务逻辑伪代码（中文）")

class APISpec(BaseModel):
    service_name: str
    endpoints: List[APIEndpoint]
    db_connection_string: str = Field(..., regex=r"^postgresql://.*")

# 调用LLM生成定制化代码
client = OpenAI(api_key="xxx", base_url="https://api.tencent.com/v1") # 示例网关
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "system", "content": "你是后端架构师，输出必须符合APISpec的JSON格式"},
              {"role": "user", "content": "生成一个订单管理微服务的API设计，包含创建、取消和按用户ID查询"}],
    response_format={"type": "json_object"}  # 强制JSON输出
)
spec = APISpec(**json.loads(response.choices[0].message.content))
print(f"生成 {len(spec.endpoints)} 个接口")

from langchain.text_splitter import TextSplitter
from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticChunkSplitter(TextSplitter):
    def __init__(self, model_name="BAAI/bge-small-zh", threshold=0.65):
        self.model = SentenceTransformer(model_name)
        self.threshold = threshold
        
    def split_text(self, text: str) -> List[str]:
        sentences = text.replace("\n", "。").split("。")
        if len(sentences) < 2:
            return [text]
            
        # 计算句向量
        embeddings = self.model.encode(sentences)
        chunks = []
        current_chunk = [sentences[0]]
        
        for i in range(1, len(sentences)):
            # 计算当前累积块与下一句的语义相似度
            cum_emb = np.mean([embeddings[j] for j in range(len(current_chunk))], axis=0)
            sim = np.dot(cum_emb, embeddings[i]) / (np.linalg.norm(cum_emb) * np.linalg.norm(embeddings[i]))
            
            if sim > self.threshold:  # 语义高度相关，合并
                current_chunk.append(sentences[i])
            else:  # 语义断裂，开启新块
                chunks.append("。".join(current_chunk))
                current_chunk = [sentences[i]]
        chunks.append("。".join(current_chunk))
        return chunks

from tcvectordb import VectorDBClient
from tcvectordb.model.enum import FieldType, IndexType

# 创建向量索引时同时构建稀疏向量索引（BM25）
index = Index(
    FilterIndex(name="id", field_type=FieldType.String, index_type=IndexType.PRIMARY_KEY),
    VectorIndex(name="dense", dimension=768, index_type=IndexType.HNSW, metric_type=MetricType.COSINE),
    SparseIndex(name="sparse", field_type=FieldType.SparseVector, index_type=IndexType.INVERTED)  # 稀疏向量
)

# 检索时执行RRF（倒数秩融合）加权
def hybrid_search(query: str, dense_weight=0.7, sparse_weight=0.3):
    dense_res = client.search(collection, vectors=encode(query), limit=50)
    sparse_res = client.search(collection, sparse_vectors=bm25_encode(query), limit=50)
    # RRF融合算法
    scores = {}
    for idx, doc in enumerate(dense_res + sparse_res):
        doc_id = doc["id"]
        rank = idx % 50 + 1
        scores[doc_id] = scores.get(doc_id, 0) + (dense_weight if idx < 50 else sparse_weight) / (60 + rank)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)[:10]

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    code: str
    error_log: str
    iteration: int
    max_iter: int = 5

def code_writer(state: AgentState):
    # 调用LLM根据业务需求生成初始代码
    prompt = f"生成Python函数，需求：{state['business']}。若存在历史错误，请修正：{state['error_log']}"
    return {"code": llm.invoke(prompt)}

def code_executor(state: AgentState):
    # 在沙箱中动态执行生成的代码（捕获异常）
    try:
        exec_globals = {"np": np, "pd": pd}  # 受限环境
        exec(state["code"], exec_globals)
        return {"error_log": "", "iteration": state["iteration"] + 1}
    except Exception as e:
        return {"error_log": str(e), "iteration": state["iteration"] + 1}

def should_continue(state: AgentState):
    if state["error_log"] == "" or state["iteration"] >= state["max_iter"]:
        return "end"
    else:
        return "rewrite"

# 构建图
builder = StateGraph(AgentState)
builder.add_node("writer", code_writer)
builder.add_node("executor", code_executor)
builder.add_edge("writer", "executor")
builder.add_conditional_edges("executor", should_continue, {"rewrite": "writer", "end": END})
builder.set_entry_point("writer")
app = builder.compile()

# 使用GPT-4生成1000条“业务规则->自然语言SQL”的样本
seed_rules = ["查询昨日活跃用户数", "计算渠道ROI大于5%的广告组"]
for rule in seed_rules:
    # 调用LLM生成 指令-输入-输出 三元组
    response = llm.generate(f"生成Alpaca格式：指令为'{rule}'，输出对应的SQL查询及解释")
    save_to_jsonl(response)

from transformers import AutoTokenizer, AutoModelForCausalLM, TrainingArguments
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
import torch

model = AutoModelForCausalLM.from_pretrained("meta-llama/Meta-Llama-3-8B", load_in_4bit=True)
tokenizer = AutoTokenizer.from_pretrained("meta-llama/Meta-Llama-3-8B")
model = prepare_model_for_kbit_training(model)

# 配置LoRA超参
lora_config = LoraConfig(
    r=64,                      # 秩
    lora_alpha=16,
    target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM"
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()  # 仅训练约 0.5% 参数

# 训练参数（适配腾讯云HAI的A100）
args = TrainingArguments(
    output_dir="./sql-lora",
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    num_train_epochs=3,
    learning_rate=2e-4,
    fp16=True,
    logging_steps=10,
    save_total_limit=2
)
# trainer.train()  # 实际运行

import ast
import importlib.util

def dynamic_load_function(code_str: str, func_name: str):
    # 1. 语法检查
    try:
        tree = ast.parse(code_str)
    except SyntaxError as e:
        raise ValueError(f"生成的代码包含语法错误: {e}")
    
    # 2. 安全审计：禁止危险内置函数（如 eval, exec, __import__）
    for node in ast.walk(tree):
        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
            if node.func.id in ["eval", "exec", "__import__", "open"]:
                raise PermissionError(f"非法函数调用: {node.func.id}")
    
    # 3. 编译并执行
    exec_globals = {"__builtins__": None, "sum": sum, "len": len}  # 白名单
    exec(code_str, exec_globals)
    return exec_globals.get(func_name)

# 实际业务场景：AI生成一个定制化数据清洗函数
ai_generated_code = """
def clean_customer_name(raw: str) -> str:
    import re  # 即使import也被拦截，需在白名单加入re
    return re.sub(r'[0-9]', '', raw).strip()
"""
my_func = dynamic_load_function(ai_generated_code, "clean_customer_name")
print(my_func("张三_2024"))  # 输出: 张三_

from gptcache import Cache
from gptcache.adapter import openai
from gptcache.embedding import Onnx
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation

cache = Cache()
# 使用ONNX轻量级嵌入模型进行语义去重
embedding = Onnx(model_name="all-MiniLM-L6-v2")
cache.init(
    pre_embedding_func=embedding.to_embeddings,
    similarity_evaluation=SearchDistanceEvaluation(max_distance=0.08),  # 阈值
)

# 替换openai.ChatCompletion的调用方式
response = openai.ChatCompletion.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "查询昨日销售额"}],
    cache_obj=cache
)
# 第二次相同语义查询将直接命中缓存，平均延迟从1.2s降至50ms

# 使用llama.cpp将PEFT权重合并并量化
python scripts/merge_lora.py --model_name meta-llama/Llama-3-8B --lora_path ./sql-lora
./llama.cpp/quantize ./merged.gguf ./quantized-q4_0.gguf q4_0

FROM nvidia/cuda:12.1-runtime-ubuntu22.04
RUN apt update && apt install -y python3-pip
COPY requirements.txt .
RUN pip install -r requirements.txt -i https://mirrors.cloud.tencent.com/pypi/simple
COPY ./quantized-q4_0.gguf /app/model/
COPY ./fastapi_server.py /app/
CMD ["uvicorn", "fastapi_server:app", "--host", "0.0.0.0", "--port", "8080"]

from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
import llama_cpp

app = FastAPI()
llm = llama_cpp.Llama(model_path="/app/model/quantized-q4_0.gguf", n_gpu_layers=-1)

@app.post("/v1/custom/chat")
async def custom_chat(prompt: str):
    def event_gen():
        for chunk in llm.create_chat_completion(messages=[{"role": "user", "content": prompt}], stream=True):
            yield {"data": chunk["choices"][0]["delta"].get("content", "")}
    return EventSourceResponse(event_gen())