基于LangChain/LlamaIndex搭建企业级AI应用

# LangChain的数据加载
from langchain_community.document_loaders import PyPDFLoader, TextLoader
loader = PyPDFLoader("document.pdf")
docs = loader.load()

# LlamaIndex的数据加载
from llama_index.core import SimpleDirectoryReader
documents = SimpleDirectoryReader("./data").load_data()

# LangChain RAG
from langchain_chroma import Chroma
from langchain.chains import RetrievalQA

vector_store = Chroma.from_documents(docs, embeddings)
chain = RetrievalQA.from_chain_type(llm, retriever=vector_store.as_retriever())

# LlamaIndex RAG
from llama_index.core import VectorStoreIndex

index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

# LangChain Agent
from langchain.agents import create_tool_calling_agent

tools = [search_tool, calculator_tool, database_tool]
agent = create_tool_calling_agent(llm, tools, prompt)

# LlamaIndex Agent
from llama_index.core.agent import ReActAgent

agent = ReActAgent.from_tools(tools, llm=llm, verbose=True)

# 使用LlamaIndex处理核心RAG逻辑
from llama_index.core import (
    VectorStoreIndex, 
    SimpleDirectoryReader,
    StorageContext,
)
from llama_index.core.postprocessor import SentenceTransformerRerank
from llama_index.vector_stores.milvus import MilvusVectorStore

# 知识库构建
documents = SimpleDirectoryReader("./knowledge_base").load_data()
index = VectorStoreIndex.from_documents(
    documents,
    vector_store=MilvusVectorStore(uri="./milvus.db"),
)

# 查询引擎（带重排序）
query_engine = index.as_query_engine(
    similarity_top_k=10,
    node_postprocessors=[
        SentenceTransformerRerank(top_k=3, model="BAAI/bge-reranker-v2-m3")
    ],
)

# 使用LangChain补充Agent能力
from langchain.agents import tool

@tool
def query_knowledge_base(question: str) -> str:
    """查询内部技术知识库"""
    response = query_engine.query(question)
    return str(response)

# 结合Agent实现更复杂的查询

用户输入
    │
    ▼
┌──────────────────┐
│  LlamaIndex      │
│  ┌──────────────┐│
│  │ 查询重写      ││
│  │ 混合检索      ││  ←── 核心RAG能力
│  │ 重排序       ││
│  │ 引用生成      ││
│  └──────────────┘│
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│  LangChain       │
│  ┌──────────────┐│
│  │ 多轮对话管理  ││  ←── 补充能力
│  │ 工具调度      ││
│  │ 输出格式化    ││
│  └──────────────┘│
└────────┬─────────┘
         │
         ▼
   用户看到回答

┌────────────────────────────────────────┐
│          应用层（FastAPI）               │
│   REST API / WebSocket / Admin Panel   │
├────────────────────────────────────────┤
│          编排层（LangChain）             │
│   Agent编排 / 多轮对话 / Tool调度       │
├────────────────────────────────────────┤
│          知识层（LlamaIndex）            │
│   索引构建 / 检索 / 重排序 / 引用       │
├────────────────────────────────────────┤
│          基础设施层                      │
│   向量库(Milvus) / 文档存储(MinIO) / LLM│
└────────────────────────────────────────┘

# 缓存策略
from functools import lru_cache

class QueryCache:
    def __init__(self, ttl=3600):
        self.cache = {}
        self.ttl = ttl
    
    async def get_or_query(self, question, query_func):
        cache_key = hashlib.md5(question.encode()).hexdigest()
        if cache_key in self.cache:
            entry = self.cache[cache_key]
            if time.time() - entry['time'] < self.ttl:
                return entry['result']
        
        result = await query_func(question)
        self.cache[cache_key] = {'result': result, 'time': time.time()}
        return result

# LangChain的Callback实现监控
from langchain.callbacks.base import BaseCallbackHandler

class MonitoringCallback(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        logger.info(f"LLM请求开始: {prompts[:100]}...")
    
    def on_llm_end(self, response, **kwargs):
        logger.info(f"LLM请求完成: {len(response.generations[0][0].text)} tokens")
    
    def on_chain_error(self, error, **kwargs):
        logger.error(f"链执行错误: {error}")
        metrics.increment('chain_errors')

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

# 预加载模型和索引
RUN python -c "from llama_index.core import VectorStoreIndex; \
    VectorStoreIndex.load_from_disk('./index')"

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]