LangChain 实战指南：构建强大的 LLM 应用

# 安装 LangChain
pip install langchain langchain-core langchain-community

# 安装 LLM 提供商
pip install langchain-openai  # OpenAI
pip install langchain-anthropic  # Anthropic
pip install langchain-ollama  # 本地模型

# 安装向量数据库
pip install chromadb faiss-cpu

# 安装文档加载器
pip install pypdf unstructured

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_ollama import ChatOllama

# OpenAI
llm_openai = ChatOpenAI(
    model="gpt-4o",
    temperature=0.7,
    max_tokens=2000
)

# Anthropic
llm_claude = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    temperature=0.7
)

# 本地模型（Ollama）
llm_local = ChatOllama(
    model="llama3.1",
    temperature=0.7
)

# 调用模型
response = llm_openai.invoke("什么是 LangChain？")
print(response.content)

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 简单提示模板
simple_prompt = ChatPromptTemplate.from_template(
    "请用中文解释什么是{concept}，给出简洁的定义和一个例子。"
)

# 多轮对话模板
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{domain}顾问，请用专业但易懂的语言回答问题。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}")
])

# 带示例的模板
few_shot_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个代码助手。"),
    ("human", "将以下Python代码转换为JavaScript：\n{code}"),
    ("assistant", "以下是等价的JavaScript代码：\n```javascript\n{example_js}\n```"),
    ("human", "将以下Python代码转换为JavaScript：\n{new_code}")
])

# 使用模板
chain = simple_prompt | llm_openai
result = chain.invoke({"concept": "向量数据库"})
print(result.content)

from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field

# 字符串解析器
str_parser = StrOutputParser()

# JSON 解析器
class CodeReview(BaseModel):
    issues: list[str] = Field(description="发现的问题列表")
    suggestions: list[str] = Field(description="改进建议")
    score: int = Field(description="代码质量评分 1-10")

json_parser = JsonOutputParser(pydantic_object=CodeReview)

# 使用解析器
review_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是代码审查专家。{format_instructions}"),
    ("human", "请审查以下代码：\n{code}")
])

chain = review_prompt | llm_openai | json_parser

result = chain.invoke({
    "code": "def add(a, b): return a + b",
    "format_instructions": json_parser.get_format_instructions()
})
print(result)
# {'issues': [], 'suggestions': ['添加类型注解', '添加文档字符串'], 'score': 7}

from langchain_core.runnables import RunnablePassthrough

# 使用 LCEL (LangChain Expression Language)
prompt = ChatPromptTemplate.from_template(
    "将以下文本翻译成{language}：\n{text}"
)

# 构建链
chain = prompt | llm_openai | str_parser

# 调用
result = chain.invoke({
    "language": "英文",
    "text": "LangChain 是一个强大的 LLM 应用开发框架"
})
print(result)

from langchain_core.runnables import RunnableParallel

# 第一步：生成大纲
outline_prompt = ChatPromptTemplate.from_template(
    "请为以下主题生成一个文章大纲：\n{topic}"
)

# 第二步：根据大纲写文章
article_prompt = ChatPromptTemplate.from_template(
    "根据以下大纲写一篇详细的文章：\n{outline}"
)

# 第三步：生成摘要
summary_prompt = ChatPromptTemplate.from_template(
    "为以下文章生成一个简洁的摘要：\n{article}"
)

# 构建顺序链
outline_chain = outline_prompt | llm_openai | str_parser
article_chain = article_prompt | llm_openai | str_parser
summary_chain = summary_prompt | llm_openai | str_parser

# 串联执行
full_chain = (
    {"outline": outline_chain}
    | RunnablePassthrough.assign(article=article_chain)
    | RunnablePassthrough.assign(summary=summary_chain)
)

result = full_chain.invoke({"topic": "人工智能的未来"})
print(result["outline"])
print("---")
print(result["article"])
print("---")
print(result["summary"])

from langchain_core.runnables import RunnableBranch

# 不同的处理分支
simple_prompt = ChatPromptTemplate.from_template("简单回答：{question}")
detailed_prompt = ChatPromptTemplate.from_template("详细解释：{question}")
technical_prompt = ChatPromptTemplate.from_template("技术分析：{question}")

# 路由函数
def route_by_complexity(input):
    question = input["question"]
    if len(question) < 20:
        return simple_prompt
    elif "技术" in question or "代码" in question:
        return technical_prompt
    else:
        return detailed_prompt

# 构建分支链
branch = RunnableBranch(
    (lambda x: len(x["question"]) < 20, simple_prompt | llm_openai),
    (lambda x: "技术" in x["question"], technical_prompt | llm_openai),
    detailed_prompt | llm_openai  # 默认分支
)

result = branch.invoke({"question": "什么是机器学习？"})
print(result.content)

from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    DirectoryLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 加载 PDF 文档
loader = PyPDFLoader("knowledge_base.pdf")
documents = loader.load()

# 加载目录下所有文本文件
loader = DirectoryLoader(
    "./docs",
    glob="**/*.md",
    loader_cls=TextLoader
)
documents = loader.load()

# 文本分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", "。", "！", "？", "，", " "]
)
chunks = text_splitter.split_documents(documents)

print(f"原始文档数: {len(documents)}")
print(f"分割后块数: {len(chunks)}")

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma, FAISS

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Chroma 向量数据库
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# FAISS 向量数据库
vectorstore = FAISS.from_documents(
    documents=chunks,
    embedding=embeddings
)

# 保存和加载
vectorstore.save_local("faiss_index")
vectorstore = FAISS.load_local("faiss_index", embeddings)

# 相似性搜索
results = vectorstore.similarity_search(
    "什么是 RAG？",
    k=3
)
for doc in results:
    print(doc.page_content[:100])
    print("---")

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# RAG 提示模板
rag_prompt = ChatPromptTemplate.from_template("""
你是一个知识助手。请根据以下上下文回答问题。
如果上下文中没有相关信息，请说"我无法根据提供的信息回答这个问题"。

上下文：
{context}

问题：{question}

回答：
""")

# 构建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)

# 构建 RAG 链
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

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

# 使用 RAG
answer = rag_chain.invoke("什么是检索增强生成？")
print(answer)

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# 1. 上下文压缩
compressor = LLMChainExtractor.from_llm(llm_openai)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever()
)

# 2. 混合检索（向量 + BM25）
bm25_retriever = BM25Retriever.from_documents(chunks)
vector_retriever = vectorstore.as_retriever()

ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]
)

# 3. 多查询检索
from langchain.retrievers import MultiQueryRetriever

multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm_openai
)

# 4. 自查询检索（元数据过滤）
from langchain.chains.query_constructor.base import AttributeInfo

metadata_field_info = [
    AttributeInfo(
        name="source",
        type="string",
        description="文档来源"
    ),
    AttributeInfo(
        name="page",
        type="integer",
        description="页码"
    )
]

from langchain.retrievers.self_query.base import SelfQueryRetriever

self_query_retriever = SelfQueryRetriever.from_llm(
    llm=llm_openai,
    vectorstore=vectorstore,
    document_contents="文档内容",
    metadata_field_info=metadata_field_info
)

from langchain.memory import (
    ConversationBufferMemory,
    ConversationSummaryMemory,
    ConversationBufferWindowMemory
)

# 缓冲记忆（保存所有对话）
buffer_memory = ConversationBufferMemory(
    return_messages=True,
    memory_key="chat_history"
)

# 窗口记忆（只保留最近 k 轮）
window_memory = ConversationBufferWindowMemory(
    k=5,
    return_messages=True,
    memory_key="chat_history"
)

# 摘要记忆（压缩历史对话）
summary_memory = ConversationSummaryMemory(
    llm=llm_openai,
    return_messages=True,
    memory_key="chat_history"
)

# 使用记忆
from langchain.chains import ConversationChain

conversation = ConversationChain(
    llm=llm_openai,
    memory=buffer_memory,
    verbose=True
)

conversation.predict(input="你好，我叫张三")
conversation.predict(input="我是一名 Python 开发者")
conversation.predict(input="你还记得我的名字吗？")

from langchain.chains import ConversationalRetrievalChain

# 构建带记忆的 RAG 链
qa_chain = ConversationalRetrievalChain.from_llm(
    llm=llm_openai,
    retriever=vectorstore.as_retriever(),
    memory=ConversationBufferMemory(
        memory_key="chat_history",
        return_messages=True
    ),
    return_source_documents=True
)

# 多轮对话
result1 = qa_chain({"question": "什么是 LangChain？"})
print(result1["answer"])

result2 = qa_chain({"question": "它有哪些主要组件？"})
print(result2["answer"])

from langchain_core.tools import tool
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

# 使用 @tool 装饰器
@tool
def search_web(query: str) -> str:
    """搜索互联网获取信息"""
    # 实现搜索逻辑
    return f"搜索结果：关于 {query} 的信息..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

# 使用 Pydantic 定义结构化工具
class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    num_results: int = Field(default=5, description="返回结果数量")

search_tool = StructuredTool.from_function(
    func=search_web,
    name="web_search",
    description="搜索互联网获取信息",
    args_schema=SearchInput
)

# 工具列表
tools = [search_web, calculate]

from langchain_community.tools import (
    DuckDuckGoSearchRun,
    WikipediaQueryRun,
    ShellTool
)
from langchain_community.utilities import WikipediaAPIWrapper

# DuckDuckGo 搜索
search = DuckDuckGoSearchRun()

# Wikipedia 查询
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

# Shell 命令
shell = ShellTool()

# Python REPL
from langchain_experimental.tools import PythonREPLTool
python_repl = PythonREPLTool()

from langchain_core.messages import HumanMessage

# 将工具绑定到模型
model_with_tools = llm_openai.bind_tools(tools)

# 调用（模型会自动决定是否使用工具）
response = model_with_tools.invoke(
    [HumanMessage(content="帮我搜索一下 LangChain 的最新版本")]
)

print(response.tool_calls)

from langchain.agents import AgentExecutor, create_react_agent
from langchain import hub

# 获取 ReAct 提示模板
prompt = hub.pull("hwchase17/react")

# 创建 ReAct Agent
agent = create_react_agent(
    llm=llm_openai,
    tools=tools,
    prompt=prompt
)

# 创建 Agent 执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    handle_parsing_errors=True
)

# 运行 Agent
result = agent_executor.invoke({
    "input": "帮我搜索 LangChain 的最新版本，并计算 2 的 10 次方"
})
print(result["output"])

from langchain.agents import create_openai_functions_agent

# OpenAI Functions Agent
agent = create_openai_functions_agent(
    llm=llm_openai,
    tools=tools,
    prompt=ChatPromptTemplate.from_messages([
        ("system", "你是一个有用的助手，可以使用工具来回答问题。"),
        MessagesPlaceholder(variable_name="chat_history", optional=True),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad")
    ])
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True
)

from langchain.agents import AgentExecutor
from langchain_core.runnables import chain

@chain
def custom_agent(input):
    """自定义 Agent 逻辑"""
    # 1. 分析输入
    analysis = llm_openai.invoke(
        f"分析以下任务，决定需要使用哪些工具：\n{input}"
    )

    # 2. 执行工具
    results = []
    for tool_call in parse_tool_calls(analysis):
        tool = get_tool(tool_call.name)
        result = tool.invoke(tool_call.args)
        results.append(result)

    # 3. 生成最终回答
    final_answer = llm_openai.invoke(
        f"基于以下结果回答问题：\n问题：{input}\n结果：{results}"
    )

    return final_answer.content

class CustomerServiceAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.memory = ConversationBufferMemory(return_messages=True)
        self.vectorstore = self.load_knowledge_base()
        self.tools = self.create_tools()
        self.agent = self.create_agent()

    def load_knowledge_base(self):
        """加载知识库"""
        loader = DirectoryLoader("./knowledge_base", glob="**/*.md")
        docs = loader.load()
        splitter = RecursiveCharacterTextSplitter(chunk_size=1000)
        chunks = splitter.split_documents(docs)
        return Chroma.from_documents(chunks, OpenAIEmbeddings())

    def create_tools(self):
        """创建工具"""
        @tool
        def search_knowledge_base(query: str) -> str:
            """搜索产品知识库"""
            docs = self.vectorstore.similarity_search(query, k=3)
            return "\n".join(doc.page_content for doc in docs)

        @tool
        def create_ticket(title: str, description: str) -> str:
            """创建工单"""
            # 实际实现中会调用工单系统 API
            return f"工单已创建：{title}"

        @tool
        def get_order_status(order_id: str) -> str:
            """查询订单状态"""
            # 实际实现中会查询订单系统
            return f"订单 {order_id} 状态：已发货"

        return [search_knowledge_base, create_ticket, get_order_status]

    def create_agent(self):
        """创建 Agent"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个专业的客服助手。
            请使用中文回答问题，保持友好和专业的语气。
            如果无法回答，请创建工单转交人工客服。"""),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])

        agent = create_openai_functions_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=prompt
        )

        return AgentExecutor(
            agent=agent,
            tools=self.tools,
            memory=self.memory,
            verbose=True
        )

    def chat(self, message: str) -> str:
        """对话接口"""
        result = self.agent.invoke({"input": message})
        return result["output"]

# 使用示例
cs_agent = CustomerServiceAgent()
response = cs_agent.chat("我的订单 ORD12345 到哪了？")
print(response)

class CodeAssistantAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.tools = self.create_tools()

    def create_tools(self):
        """创建代码相关工具"""
        @tool
        def read_file(file_path: str) -> str:
            """读取文件内容"""
            with open(file_path, 'r') as f:
                return f.read()

        @tool
        def write_file(file_path: str, content: str) -> str:
            """写入文件内容"""
            with open(file_path, 'w') as f:
                f.write(content)
            return f"文件 {file_path} 已写入"

        @tool
        def run_python(code: str) -> str:
            """执行 Python 代码"""
            import subprocess
            result = subprocess.run(
                ["python", "-c", code],
                capture_output=True,
                text=True
            )
            return result.stdout + result.stderr

        @tool
        def search_code(query: str) -> str:
            """搜索代码库"""
            # 实际实现中会使用 ripgrep 或类似工具
            return f"搜索结果：{query}"

        return [read_file, write_file, run_python, search_code]

    def assist(self, task: str) -> str:
        """提供编程帮助"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个高级 Python 开发助手。
            你可以读取、写入和执行代码来帮助用户完成编程任务。
            请确保代码安全，避免执行危险操作。"""),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])

        agent = create_openai_functions_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=prompt
        )

        executor = AgentExecutor(
            agent=agent,
            tools=self.tools,
            verbose=True,
            max_iterations=15
        )

        result = executor.invoke({"input": task})
        return result["output"]

class DataAnalysisAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.tools = self.create_tools()

    def create_tools(self):
        """创建数据分析工具"""
        @tool
        def load_csv(file_path: str) -> str:
            """加载 CSV 文件"""
            import pandas as pd
            df = pd.read_csv(file_path)
            self.df = df
            return f"数据已加载，形状：{df.shape}\n列名：{list(df.columns)}"

        @tool
        def analyze_data(query: str) -> str:
            """执行数据分析"""
            import pandas as pd

            # 安全执行 pandas 操作
            try:
                result = eval(f"self.df.{query}")
                return str(result)
            except Exception as e:
                return f"分析错误：{e}"

        @tool
        def create_visualization(chart_type: str, x: str, y: str) -> str:
            """创建可视化图表"""
            import matplotlib.pyplot as plt

            plt.figure(figsize=(10, 6))

            if chart_type == "bar":
                plt.bar(self.df[x], self.df[y])
            elif chart_type == "line":
                plt.plot(self.df[x], self.df[y])
            elif chart_type == "scatter":
                plt.scatter(self.df[x], self.df[y])

            plt.xlabel(x)
            plt.ylabel(y)
            plt.title(f"{y} by {x}")
            plt.savefig("chart.png")

            return "图表已保存为 chart.png"

        return [load_csv, analyze_data, create_visualization]

    def analyze(self, task: str) -> str:
        """执行数据分析任务"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个数据分析专家。
            你可以加载数据、执行分析和创建可视化。
            请使用 pandas 语法进行数据分析。"""),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])

        agent = create_openai_functions_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=prompt
        )

        executor = AgentExecutor(
            agent=agent,
            tools=self.tools,
            verbose=True
        )

        result = executor.invoke({"input": task})
        return result["output"]

from fastapi import FastAPI
from langserve import add_routes
from langchain_core.runnables import RunnableLambda

app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="LangChain API 服务"
)

# 添加 RAG 路由
add_routes(
    app,
    rag_chain,
    path="/rag"
)

# 添加 Agent 路由
add_routes(
    app,
    agent_executor,
    path="/agent"
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

from langchain_core.runnables import RunnableConfig

def safe_invoke(chain, input_data, max_retries=3):
    """安全调用链"""
    for attempt in range(max_retries):
        try:
            return chain.invoke(input_data)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            print(f"重试 {attempt + 1}/{max_retries}: {e}")

# 流式调用
for chunk in chain.stream({"question": "什么是 AI？"}):
    print(chunk, end="", flush=True)

from langchain_core.runnables import RunnableParallel

# 并行执行多个链
parallel_chain = RunnableParallel(
    summary=summary_chain,
    keywords=keywords_chain,
    translation=translation_chain
)

result = parallel_chain.invoke({"text": "..."})