Chenyj Space

教程目标

本教程将带你从零开始，一步步构建一个完整的 LangChain 应用。每个步骤都有可直接运行的代码。

关于本教程使用的模型：

本教程统一使用 mimo-v2.5-pro 作为 LLM 模型。如果你使用其他模型（如 OpenAI 的 gpt-4o-mini、Anthropic 的 claude-3-5-sonnet 等），只需将代码中的 model="mimo-v2.5-pro" 替换为你使用的模型名称即可。

说明：mimo-v2.5-pro 是小米大模型团队开发的模型，通过 OpenAI 兼容接口调用。你需要在 .env 中配置对应的 API Key 和 Base URL。

完成本教程后，你将掌握：

LangChain 环境搭建
基础对话链构建
RAG 知识库问答系统
工具调用 Agent
完整项目部署

第一步：环境搭建

1.1 创建项目目录

bash

# 创建项目目录
mkdir langchain-tutorial
cd langchain-tutorial

# 创建虚拟环境（Windows 使用 py，macOS/Linux 使用 python）
py -m venv venv

# 激活虚拟环境
# Windows
venv\Scripts\activate
# macOS/Linux
source venv/bin/activate

1.2 安装依赖

在项目目录下创建 requirements.txt 文件，复制以下内容：

txt

langchain>=1.0.0
langchain-core>=1.0.0
langchain-text-splitters>=1.0.0
langchain-openai>=1.0.0
langchain-huggingface>=1.0.0
langchain-chroma>=0.2.0
langchain-anthropic>=0.3.0
langgraph-checkpoint>=1.0.0
sentence-transformers>=3.0.0
python-dotenv>=1.0.0
pypdf>=4.0.0

然后执行安装命令：

bash

pip install -r requirements.txt

1.3 配置环境变量

在项目目录下创建 .env 文件，根据你使用的 API 提供商选择对应配置：

OpenAI 官方：

code

OPENAI_API_KEY=sk-xxx

OpenRouter（Token Plan）：

code

OPENAI_API_KEY=sk-or-xxx
OPENAI_BASE_URL=https://openrouter.ai/api/v1

Anthropic（Claude）：

code

ANTHROPIC_API_KEY=sk-ant-xxx

小米大模型（mimo-v2.5-pro）：

code

OPENAI_API_KEY=your-api-key
OPENAI_BASE_URL=https://api.xiaomi.com/v1

国内代理示例：

code

OPENAI_API_KEY=your-api-key
OPENAI_BASE_URL=https://api.chatanywhere.tech/v1

说明：LangChain 会自动读取 OPENAI_API_KEY 和 ANTHROPIC_API_KEY 等环境变量。如果使用代理或非 OpenAI 提供商，需要设置 OPENAI_BASE_URL。本教程使用 mimo-v2.5-pro 模型，请根据你的 API 提供商配置对应的 Key 和 URL。

1.4 验证安装

python

# test_install.py
from dotenv import load_dotenv
import os

load_dotenv()

print("环境变量加载成功")
print(f"API Key 已配置: {'OPENAI_API_KEY' in os.environ}")

bash

py test_install.py

第二步：基础对话链

2.1 最简单的对话

python

# step1_simple_chat.py
from dotenv import load_dotenv
import httpx
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

load_dotenv()

# 创建自定义 httpx 客户端（禁用 SSL 验证以解决某些环境的连接问题）
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 初始化模型
llm = ChatOpenAI(
    model="mimo-v2.5-pro",
    temperature=0.7,
    http_client=custom_client
)

# 直接调用
response = llm.invoke("你好，请用一句话介绍 LangChain")
print(response.content)

bash

py step1_simple_chat.py

2.2 使用提示模板

python

# step2_prompt_template.py
from dotenv import load_dotenv
import httpx
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 初始化模型
llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0.7, http_client=custom_client)

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{role}，请用简洁的语言回答问题。"),
    ("human", "{question}")
])

# 构建链
chain = prompt | llm

# 调用链
response = chain.invoke({
    "role": "Python 教师",
    "question": "什么是列表推导式？"
})

print(response.content)

bash

py step2_prompt_template.py

2.3 添加输出解析

python

# step3_output_parser.py
from dotenv import load_dotenv
import httpx
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from pydantic import BaseModel, Field

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0, http_client=custom_client)

# 1. 字符串解析器
str_prompt = ChatPromptTemplate.from_template(
    "用一句话解释什么是{concept}"
)
str_chain = str_prompt | llm | StrOutputParser()

result = str_chain.invoke({"concept": "机器学习"})
print("字符串输出:", result)

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

json_parser = JsonOutputParser(pydantic_object=CodeReview)

json_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是代码审查专家。{format_instructions}"),
    ("human", "审查这段代码：\n{code}")
])

json_chain = json_prompt | llm | json_parser

result = json_chain.invoke({
    "code": "def add(a, b): return a + b",
    "format_instructions": json_parser.get_format_instructions()
})
print("\nJSON 输出:", result)

bash

py step3_output_parser.py

2.4 流式输出

python

# step4_streaming.py
from dotenv import load_dotenv
import httpx
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0.7, streaming=True, http_client=custom_client)

prompt = ChatPromptTemplate.from_template(
    "请详细解释什么是{topic}，至少 200 字"
)

chain = prompt | llm | StrOutputParser()

# 流式输出
print("回答：", end="")
for chunk in chain.stream({"topic": "深度学习"}):
    print(chunk, end="", flush=True)
print()

bash

py step4_streaming.py

第三步：对话记忆

3.1 基础对话记忆

python

# step5_conversation_memory.py
from dotenv import load_dotenv
import httpx
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0.7, http_client=custom_client)

# 带记忆的提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手。请根据对话历史回答问题。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}")
])

chain = prompt | llm | StrOutputParser()

# 对话历史
chat_history = []

def chat(user_input):
    """带记忆的对话函数"""
    response = chain.invoke({
        "chat_history": chat_history,
        "input": user_input
    })

    # 更新历史
    chat_history.append(HumanMessage(content=user_input))
    chat_history.append(AIMessage(content=response))

    return response

# 测试对话
print("助手:", chat("你好，我叫张三"))
print("助手:", chat("我是一名 Python 开发者"))
print("助手:", chat("你还记得我的名字吗？"))

bash

py step5_conversation_memory.py

3.2 使用 RunnableWithMessageHistory

python

# step6_buffer_memory.py
from dotenv import load_dotenv
import httpx
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0.7, http_client=custom_client)

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

# 创建链
chain = prompt | llm

# 创建消息历史存储
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 创建带历史的链
with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 测试对话
config = {"configurable": {"session_id": "user_1"}}

print("=== 对话测试 ===")
response1 = with_history.invoke({"input": "你好，我叫李四"}, config=config)
print(f"助手: {response1.content}")

response2 = with_history.invoke({"input": "我喜欢 Python"}, config=config)
print(f"助手: {response2.content}")

response3 = with_history.invoke({"input": "你还记得我叫什么吗？"}, config=config)
print(f"助手: {response3.content}")

bash

py step6_buffer_memory.py

3.3 带摘要的长对话

python

# step7_summary_memory.py
from dotenv import load_dotenv
import httpx
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0.7, http_client=custom_client)

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手。请根据对话历史回答问题。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

chain = prompt | llm

# 消息历史存储
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 模拟长对话
config = {"configurable": {"session_id": "user_2"}}

messages = [
    "你好，我是王五，一名数据科学家",
    "我正在做一个客户流失预测项目",
    "数据集有 10 万条记录，50 个特征",
    "目前使用 XGBoost 模型，AUC 是 0.85",
    "我想尝试用深度学习来提升效果",
    "你觉得我应该用什么模型？"
]

for msg in messages:
    response = with_history.invoke({"input": msg}, config=config)
    print(f"用户: {msg}")
    print(f"助手: {response.content[:100]}...\n")

# 查看历史消息数量
history = get_session_history("user_2")
print(f"=== 共 {len(history.messages)} 条消息 ===")

bash

py step7_summary_memory.py

第四步：RAG 知识库问答

4.1 文档加载与分割

python

# step8_document_loading.py
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

# 1. 创建示例文档
sample_text = """
LangChain 是一个用于开发由大语言模型（LLM）驱动的应用程序的框架。

它提供了一套工具、组件和接口，简化了构建基于 LLM 的应用程序的过程。

LangChain 的核心概念包括：
1. Models（模型）：支持多种 LLM 提供商的统一接口
2. Prompts（提示）：管理和优化提示模板
3. Chains（链）：将多个组件串联成工作流
4. Memory（记忆）：维护对话上下文
5. Agents（智能体）：让 LLM 自主决策使用哪些工具
6. Retrieval（检索）：从外部数据源获取信息

LangChain 支持 Python 和 JavaScript 两种语言版本。
Python 版本更加成熟，拥有更丰富的功能。
"""

# 2. 创建文档对象
documents = [Document(page_content=sample_text, metadata={"source": "sample"})]
print(f"加载了 {len(documents)} 个文档")

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

chunks = text_splitter.split_documents(documents)
print(f"分割成 {len(chunks)} 个块")

# 查看分割结果
for i, chunk in enumerate(chunks):
    print(f"\n--- 块 {i+1} ---")
    print(chunk.page_content)
    print(f"元数据: {chunk.metadata}")

bash

py step8_document_loading.py

4.2 创建向量数据库

安装依赖：

使用本地嵌入模型需要额外安装 sentence-transformers：

bash

pip install sentence-transformers

说明：langchain-huggingface 已在 requirements.txt 中包含，但 sentence-transformers 是其运行时依赖，需要单独安装。首次运行会自动下载模型文件（约 100MB）。

python

# step9_vector_store.py
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from langchain_huggingface import HuggingFaceEmbeddings

# 1. 创建文档
sample_text = """
LangChain 是一个用于开发由大语言模型（LLM）驱动的应用程序的框架。

它提供了一套工具、组件和接口，简化了构建基于 LLM 的应用程序的过程。

LangChain 的核心概念包括：
1. Models（模型）：支持多种 LLM 提供商的统一接口
2. Prompts（提示）：管理和优化提示模板
3. Chains（链）：将多个组件串联成工作流
4. Memory（记忆）：维护对话上下文
5. Agents（智能体）：让 LLM 自主决策使用哪些工具
6. Retrieval（检索）：从外部数据源获取信息
"""

documents = [Document(page_content=sample_text, metadata={"source": "sample"})]

# 2. 分割文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=50
)
chunks = text_splitter.split_documents(documents)

# 3. 创建本地嵌入模型（首次运行会自动下载模型）
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5"
)

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

print("向量数据库创建成功")

# 5. 相似性搜索测试
query = "LangChain 的核心概念有哪些？"
results = vectorstore.similarity_search(query, k=3)

print(f"\n查询: {query}")
print("相关结果:")
for i, doc in enumerate(results):
    print(f"\n结果 {i+1}:")
    print(doc.page_content)

bash

py step9_vector_store.py

4.3 构建 RAG 链

python

# step10_rag_chain.py
from dotenv import load_dotenv
import httpx
from langchain_openai import ChatOpenAI
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

load_dotenv()

# 创建自定义 httpx 客户端（禁用 SSL 验证以解决某些环境的连接问题）
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 1. 加载向量数据库（必须使用与创建时相同的嵌入模型）
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5")
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

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

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

上下文：
{context}

问题：{question}

回答：
""")

# 4. 创建 LLM（使用自定义客户端解决 SSL 连接问题）
llm = ChatOpenAI(
    model="mimo-v2.5-pro",
    temperature=0,
    http_client=custom_client
)

# 5. 格式化文档函数
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 6. 构建 RAG 链
rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | prompt
    | llm
    | StrOutputParser()
)

# 7. 测试问答
questions = [
    "LangChain 是什么？",
    "LangChain 有哪些核心概念？",
    "LangChain 支持哪些编程语言？"
]

for question in questions:
    print(f"\n问题: {question}")
    answer = rag_chain.invoke(question)
    print(f"回答: {answer}")

bash

py step10_rag_chain.py

4.4 带记忆的 RAG

python

# step11_rag_with_memory.py
from dotenv import load_dotenv
import httpx
from langchain_openai import ChatOpenAI
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

load_dotenv()

# 创建自定义 httpx 客户端（禁用 SSL 验证以解决某些环境的连接问题）
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 1. 初始化组件（必须使用与创建时相同的嵌入模型）
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5")
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
llm = ChatOpenAI(
    model="mimo-v2.5-pro",
    temperature=0,
    http_client=custom_client
)

# 2. 带记忆的提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个知识助手。根据上下文和对话历史回答问题。
    如果没有相关信息，请说"我无法根据提供的信息回答这个问题"。

    上下文：{context}"""),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}")
])

# 3. 格式化函数
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 4. 构建链
chain = (
    {
        "context": (lambda x: x["question"]) | retriever | format_docs,
        "chat_history": lambda x: x["chat_history"],
        "question": lambda x: x["question"]
    }
    | prompt
    | llm
    | StrOutputParser()
)

# 5. 对话历史
chat_history = []

def rag_chat(question):
    """带记忆的 RAG 对话"""
    response = chain.invoke({
        "chat_history": chat_history,
        "question": question
    })

    # 更新历史
    chat_history.append(HumanMessage(content=question))
    chat_history.append(AIMessage(content=response))

    return response

# 6. 测试多轮对话
print("用户: LangChain 是什么？")
print("助手:", rag_chat("LangChain 是什么？"))

print("\n用户: 它有哪些核心概念？")
print("助手:", rag_chat("它有哪些核心概念？"))

print("\n用户: 你刚才提到的记忆系统是怎么工作的？")
print("助手:", rag_chat("你刚才提到的记忆系统是怎么工作的？"))

bash

py step11_rag_with_memory.py

第五步：工具调用 Agent

5.1 定义工具

python

# step12_tools.py
from langchain_core.tools import tool
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

# 1. 使用 @tool 装饰器定义工具
@tool
def search_web(query: str) -> str:
    """搜索互联网获取信息"""
    # 这里是模拟实现，实际应用中接入搜索 API
    results = {
        "LangChain": "LangChain 是一个 LLM 应用开发框架",
        "Python": "Python 是一种流行的编程语言",
        "AI": "人工智能是计算机科学的一个分支"
    }

    for key, value in results.items():
        if key.lower() in query.lower():
            return value

    return f"未找到关于 '{query}' 的信息"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        # 安全的数学计算
        allowed_chars = set("0123456789+-*/.() ")
        if all(c in allowed_chars for c in expression):
            result = eval(expression)
            return str(result)
        else:
            return "错误：表达式包含不允许的字符"
    except Exception as e:
        return f"计算错误: {e}"

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

def advanced_search(query: str, num_results: int = 3) -> str:
    """高级搜索功能"""
    return f"搜索 '{query}'，返回 {num_results} 个结果"

search_tool = StructuredTool.from_function(
    func=advanced_search,
    name="advanced_search",
    description="执行高级搜索",
    args_schema=SearchInput
)

# 测试工具
print("测试 search_web:", search_web.invoke("LangChain"))
print("测试 calculate:", calculate.invoke("2 + 3 * 4"))
print("测试 advanced_search:", search_tool.invoke({"query": "Python", "num_results": 5}))

bash

py step12_tools.py

5.2 创建 Agent

python

# step13_agent.py
from dotenv import load_dotenv
import httpx
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

load_dotenv()

# 创建自定义 httpx 客户端（禁用 SSL 验证以解决某些环境的连接问题）
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 创建 LLM
llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0, http_client=custom_client)

# 1. 定义工具
@tool
def search_web(query: str) -> str:
    """搜索互联网获取信息"""
    results = {
        "天气": "今天北京天气晴朗，气温 25°C",
        "新闻": "最新科技新闻：AI 技术取得重大突破",
        "股票": "苹果公司股价今日上涨 2%"
    }

    for key, value in results.items():
        if key in query:
            return value

    return f"搜索结果：关于 '{query}' 的信息"

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

# 2. 创建 Agent
agent = create_agent(
    model=llm,
    tools=[search_web, calculate],
    system_prompt="你是一个有用的助手，可以使用工具来回答问题。"
)

# 3. 测试 Agent
print("=== 测试 1: 天气查询 ===")
result1 = agent.invoke({"messages": [{"role": "user", "content": "今天北京天气怎么样？"}]})
print("回答:", result1["messages"][-1].content)

print("\n=== 测试 2: 计算 ===")
result2 = agent.invoke({"messages": [{"role": "user", "content": "计算 (15 + 27) * 3"}]})
print("回答:", result2["messages"][-1].content)

print("\n=== 测试 3: 综合 ===")
result3 = agent.invoke({"messages": [{"role": "user", "content": "帮我搜索一下最新的 AI 新闻"}]})
print("回答:", result3["messages"][-1].content)

bash

py step13_agent.py

5.3 带记忆的 Agent

python

# step14_agent_with_memory.py
from dotenv import load_dotenv
import httpx
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()

# 创建自定义 httpx 客户端（禁用 SSL 验证以解决某些环境的连接问题）
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 创建 LLM
llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0, http_client=custom_client)

# 1. 定义工具
@tool
def search_web(query: str) -> str:
    """搜索互联网获取信息"""
    return f"搜索结果：关于 '{query}' 的最新信息"

@tool
def get_user_info(name: str) -> str:
    """获取用户信息"""
    users = {
        "张三": "张三，25岁，软件工程师",
        "李四": "李四，30岁，产品经理"
    }
    return users.get(name, f"未找到用户 {name} 的信息")

# 2. 创建记忆存储
memory = MemorySaver()

# 3. 创建带记忆的 Agent
agent = create_agent(
    model=llm,
    tools=[search_web, get_user_info],
    system_prompt="你是一个友好的助手，可以使用工具。请记住对话历史。",
    checkpointer=memory
)

# 4. 测试多轮对话
config = {"configurable": {"thread_id": "user_1"}}

print("=== 第一轮 ===")
result1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫张三"}]},
    config=config
)
print(result1["messages"][-1].content)

print("\n=== 第二轮 ===")
result2 = agent.invoke(
    {"messages": [{"role": "user", "content": "帮我搜索 Python 教程"}]},
    config=config
)
print(result2["messages"][-1].content)

print("\n=== 第三轮 ===")
result3 = agent.invoke(
    {"messages": [{"role": "user", "content": "你还记得我叫什么吗？"}]},
    config=config
)
print(result3["messages"][-1].content)

bash

py step14_agent_with_memory.py

第六步：完整项目实战

6.1 项目结构

code

langchain-project/
├── .env
├── requirements.txt
├── main.py
├── knowledge_base/
│   ├── docs/
│   │   ├── langchain_intro.txt
│   │   └── python_basics.txt
│   └── vector_store/
├── tools/
│   ├── __init__.py
│   ├── search.py
│   └── calculator.py
└── app/
    ├── __init__.py
    ├── chains.py
    ├── agents.py
    └── memory.py

6.2 创建知识库文档

python

# create_knowledge_base.py
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from langchain_huggingface import HuggingFaceEmbeddings

# 创建示例文档内容
docs_content = {
    "langchain_intro": """
LangChain 是一个用于开发由大语言模型驱动的应用程序的框架。

核心组件：
1. Models：支持 OpenAI、Anthropic 等多种 LLM
2. Prompts：管理和优化提示模板
3. Chains：串联多个组件的工作流
4. Memory：维护对话上下文
5. Agents：让 LLM 自主使用工具
6. Retrieval：从外部数据源获取信息

LangChain 的主要优势：
- 统一的接口设计
- 丰富的组件库
- 活跃的社区支持
- 完善的文档
""",
    "python_basics": """
Python 是一种高级编程语言，由 Guido van Rossum 于 1991 年创建。

Python 的特点：
1. 简洁易读的语法
2. 丰富的标准库
3. 跨平台支持
4. 动态类型系统
5. 自动内存管理

Python 常用数据类型：
- int：整数
- float：浮点数
- str：字符串
- list：列表
- dict：字典
- tuple：元组
- set：集合

Python 常用库：
- NumPy：数值计算
- Pandas：数据处理
- Matplotlib：数据可视化
- Requests：HTTP 请求
- Flask/Django：Web 框架
"""
}

# 创建文档对象
documents = []
for name, content in docs_content.items():
    documents.append(Document(page_content=content, metadata={"source": name}))
    print(f"创建文档: {name}")

# 分割文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=100
)
all_chunks = text_splitter.split_documents(documents)

print(f"总共分割成 {len(all_chunks)} 个块")

# 创建向量数据库（首次运行会自动下载模型）
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5"
)
vectorstore = Chroma.from_documents(
    documents=all_chunks,
    embedding=embeddings,
    persist_directory="knowledge_base/vector_store"
)

print("向量数据库创建完成")

bash

py create_knowledge_base.py

6.3 完整应用代码

python

# main.py
from dotenv import load_dotenv
import httpx
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_chroma import Chroma
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()

# 创建自定义 httpx 客户端（禁用 SSL 验证以解决某些环境的连接问题）
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

class LangChainApp:
    """LangChain 完整应用"""

    def __init__(self):
        self.embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5")
        self.vectorstore = self._init_vectorstore()
        self.tools = self._init_tools()
        self.memory = MemorySaver()
        self.llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0, http_client=custom_client)
        self.agent = self._init_agent()

    def _init_vectorstore(self):
        """初始化向量数据库"""
        try:
            return Chroma(
                persist_directory="knowledge_base/vector_store",
                embedding_function=self.embeddings
            )
        except:
            print("警告：向量数据库未找到，请先运行 create_knowledge_base.py")
            return None

    def _init_tools(self):
        """初始化工具"""
        @tool
        def search_knowledge_base(query: str) -> str:
            """搜索知识库获取信息"""
            if self.vectorstore is None:
                return "知识库未初始化"

            docs = self.vectorstore.similarity_search(query, k=3)
            return "\n\n".join(doc.page_content for doc in docs)

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

        return [search_knowledge_base, calculate]

    def _init_agent(self):
        """初始化 Agent"""
        return create_agent(
            model=self.llm,
            tools=self.tools,
            system_prompt="""你是一个智能助手。你可以：
            1. 搜索知识库回答问题
            2. 进行数学计算
            3. 记住对话历史

            请用中文回答问题。""",
            checkpointer=self.memory
        )

    def chat(self, message: str, session_id: str = "default") -> str:
        """对话接口"""
        config = {"configurable": {"thread_id": session_id}}
        result = self.agent.invoke(
            {"messages": [{"role": "user", "content": message}]},
            config=config
        )
        return result["messages"][-1].content

def main():
    """主函数"""
    print("=== LangChain 智能助手 ===")
    print("输入 'quit' 退出\n")

    app = LangChainApp()

    while True:
        user_input = input("你: ")
        if user_input.lower() in ['quit', 'exit', '退出']:
            break

        response = app.chat(user_input)
        print(f"助手: {response}\n")

if __name__ == "__main__":
    main()

bash

py main.py

第七步：进阶功能

7.1 使用 LangServe 部署 API

python

# server.py
from fastapi import FastAPI
from langserve import add_routes
from main import LangChainApp

app = FastAPI(
    title="LangChain API",
    version="1.0",
    description="LangChain 智能助手 API"
)

# 初始化应用
langchain_app = LangChainApp()

# 添加路由
add_routes(
    app,
    langchain_app.agent_executor,
    path="/chat"
)

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

bash

pip install langserve uvicorn
py server.py

7.2 使用 LangSmith 监控

python

# 配置 LangSmith
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-langchain-project"

7.3 性能优化

python

# 优化技巧
from langchain_core.runnables import RunnableParallel

# 1. 并行执行
parallel_chain = RunnableParallel(
    summary=summary_chain,
    keywords=keywords_chain
)

# 2. 缓存
from langchain_core.globals import set_llm_cache
from langchain_community.cache import SQLiteCache

set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# 3. 批量处理
results = chain.batch([input1, input2, input3])

常见问题解答

Q1: API Key 错误

code

错误：AuthenticationError
解决：检查 .env 文件中的 OPENAI_API_KEY 是否正确

Q2: 模型不存在

code

错误：Model not found
解决：确认模型名称正确，如 gpt-4o-mini、gpt-4o

Q3: 向量数据库为空

code

错误：No documents found
解决：先运行 create_knowledge_base.py 创建知识库

Q4: 内存不足

code

错误：Out of memory
解决：减少 chunk_size 或使用更小的模型

总结

本教程涵盖了 LangChain 的核心功能：

环境搭建 - 安装和配置
基础对话 - 提示模板和输出解析
对话记忆 - 维护上下文
RAG 系统 - 知识库问答
工具调用 - Agent 开发
完整项目 - 实际应用

下一步学习建议：

阅读 LangChain 官方文档
探索更多工具集成
学习 LangSmith 进行监控和调试
尝试部署到生产环境

延伸阅读：