当 Vibe Coding 遇上 AI 应用定制化：一场全栈开发范式的革命

┌─────────────┐     ┌─────────────────┐     ┌──────────────┐
│  用户输入   │────▶│ FastAPI 网关    │────▶│ 检索链       │
│  (query)    │     │ + 身份认证      │     │ (LCEL)       │
└─────────────┘     └─────────────────┘     └──────┬───────┘
                                                    │
                                                    ▼
                                         ┌──────────────────┐
                                         │  向量检索(Qdrant)│
                                         │  + 重排序       │
                                         └────────┬─────────┘
                                                  │
                                                  ▼
                                         ┌──────────────────┐
                                         │  LLM 生成       │
                                         │  + 引用标注     │
                                         └────────┬─────────┘
                                                  │
                                                  ▼
                                         ┌──────────────────┐
                                         │  流式响应       │
                                         └──────────────────┘

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore
from qdrant_client import QdrantClient
import os

# 1. 加载
loader = PyPDFLoader("service_manual.pdf")
docs = loader.load()

# 2. 智能分块（保留标题层级）
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=150,
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
    keep_separator=True,
)
chunks = text_splitter.split_documents(docs)

# 3. 添加元数据（用于后续过滤）
for i, chunk in enumerate(chunks):
    chunk.metadata["chunk_id"] = i
    chunk.metadata["source"] = "service_manual_v2"

# 4. 初始化 Qdrant（本地运行）
client = QdrantClient(host="localhost", port=6333)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vector_store = QdrantVectorStore.from_documents(
    documents=chunks,
    embedding=embeddings,
    client=client,
    collection_name="service_kb",
    batch_size=64,
)
print(f"已索引 {len(chunks)} 个片段")

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from operator import itemgetter

# 定制化 prompt 模板（强调引用来源）
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一位专业的售后顾问。请基于以下上下文片段回答用户问题。
    如果上下文不足，请明确告知“根据现有资料无法回答”。
    在回答末尾，必须列出所引用的片段编号（如 [1][3]）。
    上下文：
    {context}"""),
    ("human", "{question}")
])

# 检索器（返回 top-5，且过滤 product_line="A系列"）
retriever = vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5, "filter": {"product_line": "A系列"}}
)

# 重排序函数（使用交叉编码器）
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")

def rerank_documents(query, docs):
    pairs = [[query, doc.page_content] for doc in docs]
    scores = reranker.predict(pairs)
    sorted_docs = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, _ in sorted_docs[:3]]  # 保留前3

# LCEL 链
def format_docs(docs):
    return "\n\n".join(f"[{i+1}] {doc.page_content}" for i, doc in enumerate(docs))

chain = (
    {
        "context": (itemgetter("question") 
                    | retriever 
                    | RunnableLambda(lambda docs: rerank_documents(query, docs))
                    | format_docs),
        "question": itemgetter("question")
    }
    | prompt
    | ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
    | JsonOutputParser()  # 强制输出 { "answer": "...", "citations": [...] }
)

# 异步调用
async def answer_question(question: str):
    result = await chain.ainvoke({"question": question})
    return result

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sse_starlette.sse import EventSourceResponse
import json

app = FastAPI(title="定制化知识库问答")

class QueryRequest(BaseModel):
    question: str
    stream: bool = False

@app.post("/ask")
async def ask_endpoint(req: QueryRequest):
    try:
        if req.stream:
            # 流式生成（使用 async for 迭代 token）
            async def event_generator():
                async for chunk in chain.astream({"question": req.question}):
                    yield {"data": json.dumps(chunk)}
            return EventSourceResponse(event_generator())
        else:
            result = await chain.ainvoke({"question": req.question})
            return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

import streamlit as st
import requests

st.set_page_config(page_title="售后智能问答")
st.title("📚 定制化知识库")

if "messages" not in st.session_state:
    st.session_state.messages = []

for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

if prompt := st.chat_input("请输入您的问题"):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)
    with st.chat_message("assistant"):
        resp = requests.post("http://localhost:8000/ask", json={"question": prompt})
        data = resp.json()
        st.markdown(data["answer"])
        st.caption(f"引用: {', '.join(data.get('citations', []))}")
    st.session_state.messages.append({"role": "assistant", "content": data["answer"]})

from langchain.globals import set_llm_cache
from langchain.cache import RedisCache
import redis

redis_client = redis.Redis(host="localhost", port=6379)
set_llm_cache(RedisCache(redis_client))

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def rerank_documents_async(query, docs):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, rerank_documents, query, docs)

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_relevancy

dataset = ...  # 包含 question, answer, contexts, ground_truth
result = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_relevancy])
print(result)

version: '3.8'
services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
    volumes:
      - qdrant_storage:/qdrant/storage

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  api:
    build: ./api
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - QDRANT_HOST=qdrant
      - REDIS_HOST=redis
    depends_on:
      - qdrant
      - redis
    command: uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

  streamlit:
    build: ./frontend
    ports:
      - "8501:8501"
    depends_on:
      - api