LangChain 实战入门：构建 AI 应用的瑞士军刀

前言

LangChain 是构建 LLM 应用的框架，它提供了一系列工具和组件，让开发者能够快速构建复杂的 AI 应用。作为一个全栈开发者，我发现 LangChain 大大简化了 RAG、Agent 等复杂系统的开发流程。

今天分享 LangChain 的核心概念和实战经验。

核心概念

1. Chains

Chains 是 LangChain 的核心概念，它允许将多个组件串联起来：

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI

# 创建提示模板
prompt = PromptTemplate(
    input_variables=["topic"],
    template="请解释什么是 {topic}？用简单易懂的语言描述。"
)

# 创建链
chain = LLMChain(llm=OpenAI(temperature=0), prompt=prompt)

# 运行链
result = chain.run("人工智能")
print(result)

2. Agents

Agents 让 LLM 能够自主决定调用哪些工具：

from langchain.agents import initialize_agent, Tool
from langchain.llms import OpenAI

# 定义工具
def search_web(query):
    """搜索网页"""
    return f"搜索结果关于: {query}"

tools = [
    Tool(
        name="WebSearch",
        func=search_web,
        description="用于搜索网络信息"
    )
]

# 初始化 Agent
agent = initialize_agent(
    tools,
    OpenAI(temperature=0),
    agent="zero-shot-react-description"
)

# 运行 Agent
result = agent.run("今天北京天气怎么样？")
print(result)

3. Memory

Memory 让对话具有上下文记忆：

from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

# 创建记忆
memory = ConversationBufferMemory()

# 创建对话链
conversation = ConversationChain(
    llm=OpenAI(temperature=0),
    memory=memory
)

# 多轮对话
result1 = conversation.predict(input="我叫张三")
result2 = conversation.predict(input="我叫什么名字？")
print(result2)  # 应该回答"张三"

RAG 实战

构建简单的 RAG 系统

from langchain.document_loaders import TextLoader
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.chains import RetrievalQA

# 加载文档
loader = TextLoader("knowledge.txt")
documents = loader.load()

# 创建向量数据库
embeddings = OpenAIEmbeddings()
db = FAISS.from_documents(documents, embeddings)

# 创建检索链
qa = RetrievalQA.from_chain_type(
    llm=OpenAI(temperature=0),
    chain_type="stuff",
    retriever=db.as_retriever()
)

# 查询
result = qa.run("文档中提到了什么？")
print(result)

高级 RAG 配置

from langchain.chains import RetrievalQAWithSourcesChain

# 带来源的 RAG
qa_with_sources = RetrievalQAWithSourcesChain.from_chain_type(
    llm=OpenAI(temperature=0),
    chain_type="map_reduce",
    retriever=db.as_retriever(search_kwargs={"k": 5})
)

result = qa_with_sources({"question": "什么是机器学习？"})
print(result["answer"])
print(result["sources"])

自定义 Chains

from langchain.chains.base import Chain
from langchain.callbacks.manager import CallbackManagerForChainRun
from typing import Dict, List

class CustomChain(Chain):
    """自定义链"""
    
    @property
    def input_keys(self) -> List[str]:
        return ["input"]
    
    @property
    def output_keys(self) -> List[str]:
        return ["output"]
    
    def _call(self, inputs: Dict[str, str], run_manager: CallbackManagerForChainRun = None) -> Dict[str, str]:
        input_text = inputs["input"]
        
        # 处理逻辑
        result = self.process(input_text)
        
        return {"output": result}
    
    def process(self, text: str) -> str:
        """实际处理逻辑"""
        return f"处理后的结果: {text.upper()}"

实战：构建智能客服

class SmartCustomerService:
    """智能客服系统"""
    
    def __init__(self):
        self.knowledge_base = self._load_knowledge()
        self.conversation_memory = ConversationBufferMemory()
    
    def _load_knowledge(self):
        """加载知识库"""
        loader = TextLoader("faq.txt")
        documents = loader.load()
        db = FAISS.from_documents(documents, OpenAIEmbeddings())
        return db.as_retriever()
    
    def chat(self, user_input: str) -> str:
        """处理用户查询"""
        # 先尝试从知识库回答
        qa = RetrievalQA.from_chain_type(
            llm=OpenAI(temperature=0),
            chain_type="stuff",
            retriever=self.knowledge_base,
            memory=self.conversation_memory
        )
        
        result = qa.run(user_input)
        
        return result

性能优化

缓存机制

from langchain.cache import InMemoryCache
from langchain.callbacks import get_openai_callback

# 设置缓存
langchain.llm_cache = InMemoryCache()

# 使用缓存
with get_openai_callback() as cb:
    # 第一次调用
    result1 = chain.run("人工智能")
    print(f"第一次调用: {cb.total_tokens} tokens")
    
    # 第二次调用（相同输入）
    result2 = chain.run("人工智能")
    print(f"第二次调用: {cb.total_tokens} tokens")  # 应该为 0

异步处理

from langchain.chains import LLMChain
from langchain.chat_models import ChatOpenAI

# 异步链
async def async_chain():
    llm = ChatOpenAI(temperature=0)
    prompt = PromptTemplate(
        input_variables=["topic"],
        template="解释什么是 {topic}？"
    )
    
    chain = LLMChain(llm=llm, prompt=prompt)
    
    # 异步调用
    result = await chain.arun("机器学习")
    return result

总结

LangChain 是构建 AI 应用的强大工具：

1. Chains：串联组件构建复杂逻辑
2. Agents：让模型自主决策和调用工具
3. Memory：保持对话上下文
4. RAG：结合外部知识库

关键要点：

• 从简单的 Chain 开始
• 利用 Memory 保持对话连续性
• RAG 需要高质量的知识库
• 缓存可以显著降低成本