大模型应用开发实战：基于LangChain与腾讯云混元构建智能问答系统

用户Query → 嵌入检索（FAISS）→ 上下文压缩 → 混元生成 → 流式返回

# 创建虚拟环境
python3.9 -m venv llm-venv
source llm-venv/bin/activate

# 安装核心依赖（版本锁定）
pip install langchain==0.2.1 langchain-community==0.2.1 \
            faiss-cpu==1.8.0 sentence-transformers==2.2.2 \
            fastapi==0.111.0 uvicorn[standard]==0.30.1 \
            python-dotenv==1.0.1 tenacity==8.2.3

TENCENT_SECRET_ID=AKIDxxxxxxxx
TENCENT_SECRET_KEY=xxxxxxxx
HUNYUAN_MODEL=hunyuan-pro
HUNYUAN_BASE_URL=https://hunyuan.tencentcloudapi.com

# hunyuan_wrapper.py
import json
import hashlib
import hmac
import time
from datetime import datetime
import requests
from langchain.schema import BaseMessage, AIMessage, HumanMessage
from langchain.callbacks.manager import CallbackManagerForLLMRun
from langchain.llms.base import LLM
from typing import List, Optional, Dict, Any

class HunyuanLLM(LLM):
    secret_id: str
    secret_key: str
    model: str = "hunyuan-pro"
    temperature: float = 0.7
    max_tokens: int = 2000

    @property
    def _llm_type(self) -> str:
        return "hunyuan"

    def _call(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs,
    ) -> str:
        # 简化的API调用，实际生产需实现TC3签名
        # 这里使用requests直接调用腾讯云API网关（示例）
        url = "https://hunyuan.tencentcloudapi.com"
        payload = {
            "Model": self.model,
            "Messages": [{"Role": "user", "Content": prompt}],
            "Temperature": self.temperature,
            "TopP": 0.8,
            "Stream": False,
        }
        headers = self._sign_request(payload)
        resp = requests.post(url, json=payload, headers=headers)
        data = resp.json()
        return data["Response"]["Choices"][0]["Message"]["Content"]

    def _sign_request(self, payload: Dict) -> Dict:
        # 省略具体签名实现（参考腾讯云官方文档）
        # 生产环境建议使用tencentcloud-sdk-python
        pass

export OPENAI_API_KEY="sk-xxx"  # 实际为混元的APIKey（通过控制台获取）
export OPENAI_BASE_URL="https://api.hunyuan.cloud.tencent.com/v1"

SYSTEM_TEMPLATE = """你是一个专业的技术顾问，擅长从给定文档中提取信息并给出清晰解答。
你必须遵循以下规则：
1. 仅基于下方「文档片段」回答，若信息不足，明确说“未找到相关信息”。
2. 回答需分点列出依据，每条依据后注明引用来源（文件名+页码）。
3. 若问题涉及代码，请提供可执行的代码示例，并标注语言。
4. 回答长度控制在200-500字之间，避免冗余。

文档片段：
{context}

问题：{question}
"""

# 少样本示例（Few-shot）提升复杂推理
FEW_SHOT_EXAMPLES = [
    {"question": "如何优化MySQL查询？", "answer": "1. 使用EXPLAIN分析执行计划...（引用：MySQL调优手册 P23）"},
    # ...
]

from langchain.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings

# 加载本地文档（支持txt, pdf, markdown等）
loader = DirectoryLoader("./docs/", glob="**/*.txt", loader_cls=TextLoader)
docs = loader.load()

# 智能分割：按段落、句子，保持重叠
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
)
chunks = splitter.split_documents(docs)
print(f"分割为 {len(chunks)} 个片段")

embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
    model_kwargs={'device': 'cpu'},
    encode_kwargs={'normalize_embeddings': True}
)

# 构建FAISS索引
from langchain.vectorstores import FAISS
vectorstore = FAISS.from_documents(chunks, embeddings)
vectorstore.save_local("faiss_index")  # 持久化

# 加载已有索引
# vectorstore = FAISS.load_local("faiss_index", embeddings)

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 5,
        "fetch_k": 20,
        "lambda_mult": 0.5,
        "score_threshold": 0.6  # 低于此值的文档丢弃
    }
)

from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI(
    model="hunyuan-pro",
    temperature=0.3,
    max_tokens=2000,
    streaming=True  # 启用流式
)

prompt = ChatPromptTemplate.from_template(SYSTEM_TEMPLATE)

def format_docs(docs):
    return "\n\n".join([f"【{doc.metadata.get('source', '未知')}】\n{doc.page_content}" for doc in docs])

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 同步调用示例
answer = rag_chain.invoke("什么是RAG？")
print(answer)

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio

app = FastAPI()

class QueryRequest(BaseModel):
    question: str
    session_id: str = "default"

async def stream_generate(question: str):
    # 构造异步迭代器
    async for chunk in rag_chain.astream(question):
        yield chunk
    yield "[DONE]"

@app.post("/rag/stream")
async def rag_stream(req: QueryRequest):
    return StreamingResponse(stream_generate(req.question), media_type="text/plain")

# 启动：uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

from langchain.cache import InMemoryCache
import langchain
langchain.llm_cache = InMemoryCache()

# 对相同或高度相似的query直接返回缓存结果（需结合嵌入相似度）
# 可自定义缓存键生成器

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_llm_with_retry(prompt):
    return llm.invoke(prompt)

[Unit]
Description=RAG API Service
After=network.target

[Service]
User=ubuntu
WorkingDirectory=/home/ubuntu/rag-app
ExecStart=/home/ubuntu/llm-venv/bin/uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

from ragas import evaluate
from ragas.metrics import context_precision, context_recall, faithfulness, answer_relevancy

# 准备测试数据集（question, answer, contexts）
eval_data = [...]
result = evaluate(eval_data, metrics=[context_precision, context_recall, faithfulness, answer_relevancy])
print(result)