1
# 安装 LangChain v1.0 核心包
2
pip install langchain langchain-core langchain-openai
3

4
# 安装其他常用包
5
pip install langchain-community langchain-chroma faiss-cpu tiktoken

1
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
2

3
# 基础提示模板
4
prompt = PromptTemplate.from_template("用一句话解释什么是{topic}？")
5
formatted = prompt.format(topic="量子计算")
6
print(formatted)
7

8
# 聊天提示模板（支持多角色）
9
chat_prompt = ChatPromptTemplate.from_messages([
10
    ("system", "你是一个{role}专家，回答要简洁专业。"),
11
    ("human", "{question}")
12
])
13

14
messages = chat_prompt.format_messages(
15
    role="Python编程",
16
    question="如何快速入门LangChain？"
17
)

1
from langchain_core.output_parsers import StrOutputParser
2

3
# 使用 LCEL 创建简单链
4
chain = chat_prompt | llm | StrOutputParser()
5

6
result = chain.invoke({
7
    "role": "AI技术",
8
    "question": "LangChain v1.0 有哪些主要变化？"
9
})
10
print(result)

1
from langchain.agents import create_agent
2
from langchain_openai import ChatOpenAI
3
from langchain.tools import tool
4

5
# 定义工具
6
@tool
7
def get_weather(city: str) -> str:
8
    """获取指定城市的天气（模拟）"""
9
    weather_data = {
10
        "北京": "晴朗，25°C",
11
        "上海": "多云，22°C",
12
        "广州": "阵雨，28°C"
13
    }
14
    return weather_data.get(city, f"{city}：天气数据暂无")
15

16
@tool
17
def calculate(expression: str) -> float:
18
    """执行数学计算"""
19
    return eval(expression)
20

21
# 创建 Agent
22
model = ChatOpenAI(model="gpt-4o-mini")
23
agent = create_agent(
24
    model=model,
25
    tools=[get_weather, calculate],
26
    system_prompt="你是一个有用的助手，可以使用工具回答问题。"
27
)
28

29
# 调用 Agent
30
result = agent.invoke({
31
    "messages": [{"role": "user", "content": "北京天气怎么样？25+15等于多少？"}]
32
})
33
print(result["content"])

1
from pydantic import BaseModel, Field
2
from langchain.agents import create_agent
3
from langchain_openai import ChatOpenAI
4

5
class WeatherReport(BaseModel):
6
    city: str = Field(description="城市名称")
7
    temperature: float = Field(description="温度（摄氏度）")
8
    condition: str = Field(description="天气状况")
9
    humidity: int = Field(description="湿度百分比")
10

11
agent = create_agent(
12
    model=ChatOpenAI(model="gpt-4o-mini"),
13
    tools=[get_weather],
14
    response_format=WeatherReport,
15
)
16

17
result = agent.invoke({
18
    "messages": [{"role": "user", "content": "北京的天气如何？"}]
19
})
20

21
print(result.structured_response)
22
# WeatherReport(city='北京', temperature=25.0, condition='晴朗', humidity=45)

1
用户问题 → 查询向量化 → 向量检索 → 上下文增强 → LLM 生成 → 答案
2
                                      ↑
3
                                向量数据库（存储文档块）

1
import os
2
from langchain.document_loaders import PyPDFLoader, TextLoader
3
from langchain.text_splitter import RecursiveCharacterTextSplitter
4
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
5
from langchain.vectorstores import Chroma
6
from langchain.chains import RetrievalQA
7
from langchain_core.prompts import ChatPromptTemplate
8

9
# 1. 加载文档
10
def load_documents(file_paths):
11
    documents = []
12
    for path in file_paths:
13
        if path.endswith('.pdf'):
14
            loader = PyPDFLoader(path)
15
        else:
16
            loader = TextLoader(path, encoding='utf-8')
17
        documents.extend(loader.load())
18
    return documents
19

20
# 2. 分割文本
21
text_splitter = RecursiveCharacterTextSplitter(
22
    chunk_size=1000,
23
    chunk_overlap=200,
24
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
25
)
26

27
docs = load_documents(["./data/technical_manual.pdf", "./data/faq.txt"])
28
split_docs = text_splitter.split_documents(docs)
29
print(f"文档已分割为 {len(split_docs)} 个文本块")
30

31
# 3. 创建向量存储
32
embeddings = OpenAIEmbeddings()
33
vectorstore = Chroma.from_documents(
34
    documents=split_docs,
35
    embedding=embeddings,
36
    persist_directory="./chroma_db"  # 持久化存储
37
)
38

39
# 4. 创建检索器
40
retriever = vectorstore.as_retriever(
41
    search_type="similarity",  # 相似度检索
42
    search_kwargs={"k": 4}      # 返回 top-4
43
)
44

45
# 5. 构建 RAG 链
46
prompt_template = ChatPromptTemplate.from_messages([
47
    ("system", """你是一个专业的技术助手。基于以下上下文信息回答问题。
48
如果无法从上下文中找到答案，请如实告知用户，不要编造信息。
49

50
上下文信息：
51
{context}"""),
52
    ("human", "{question}")
53
])
54

55
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
56

57
qa_chain = RetrievalQA.from_chain_type(
58
    llm=llm,
59
    chain_type="stuff",
60
    retriever=retriever,
61
    chain_type_kwargs={"prompt": prompt_template},
62
    return_source_documents=True  # 返回来源文档
63
)
64

65
# 6. 问答测试
66
def ask_question(question):
67
    result = qa_chain.invoke({"query": question})
68
    print(f"问题: {question}")
69
    print(f"答案: {result['result']}")
70
    print(f"来源: {[doc.metadata.get('source', '未知') for doc in result['source_documents']]}")
71
    print("-" * 50)
72
    return result
73

74
ask_question("系统的默认管理员密码是什么？")
75
ask_question("如何配置日志输出级别？")
76
ask_question("系统支持哪些数据库？")

1
# 混合检索（BM25 + 向量检索）
2
from langchain.retrievers import EnsembleRetriever
3
from langchain.retrievers import BM25Retriever
4

5
bm25_retriever = BM25Retriever.from_documents(split_docs)
6
bm25_retriever.k = 4
7

8
ensemble_retriever = EnsembleRetriever(
9
    retrievers=[bm25_retriever, vectorstore.as_retriever()],
10
    weights=[0.4, 0.6]  # BM25 权重 0.4，向量检索权重 0.6
11
)
12

13
# 查询重写
14
from langchain.retrievers import MultiQueryRetriever
15

16
multi_query_retriever = MultiQueryRetriever.from_llm(
17
    retriever=vectorstore.as_retriever(),
18
    llm=llm
19
)
20

21
# 上下文压缩
22
from langchain.retrievers import ContextualCompressionRetriever
23
from langchain.retrievers.document_compressors import LLMChainExtractor
24

25
compressor = LLMChainExtractor.from_llm(llm)
26
compression_retriever = ContextualCompressionRetriever(
27
    base_compressor=compressor,
28
    base_retriever=vectorstore.as_retriever()
29
)

1
from langchain_core.prompts import ChatPromptTemplate
2
from langchain_core.output_parsers import StrOutputParser
3
from langchain_openai import ChatOpenAI
4

5
prompt = ChatPromptTemplate.from_template("用一句话解释{topic}")
6
model = ChatOpenAI(model="gpt-4o-mini")
7
output_parser = StrOutputParser()
8

9
# 管道组合：prompt → model → parser
10
chain = prompt | model | output_parser
11

12
result = chain.invoke({"topic": "LangChain 表达式语言"})
13
print(result)

1
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
2

3
# 并行执行多个任务
4
parallel_chain = RunnableParallel(
5
    summary=prompt_template | model | output_parser,
6
    keywords=keyword_prompt | model | output_parser,
7
    sentiment=sentiment_prompt | model | output_parser
8
)
9

10
results = parallel_chain.invoke({"text": long_article})

1
from langchain_core.runnables import RunnableBranch
2

3
branch = RunnableBranch(
4
    (lambda x: "天气" in x["question"], weather_chain),
5
    (lambda x: "计算" in x["question"], calculator_chain),
6
    default_chain
7
)
8

9
result = branch.invoke({"question": "北京天气怎么样？"})

1
# 设置环境变量启用 LangSmith
2
export LANGCHAIN_TRACING_V2=true
3
export LANGCHAIN_API_KEY=your_langsmith_key
4
export LANGCHAIN_PROJECT=my-production-app

1
# 自动追踪所有调用
2
from langsmith import traceable
3

4
@traceable(run_type="tool")
5
def fetch_customer_data(customer_id: str):
6
    # 业务逻辑
7
    pass

1
from langgraph.checkpoint.sqlite import SqliteSaver
2

3
# 使用 SQLite 保存对话状态
4
checkpointer = SqliteSaver.from_conn_string("sqlite:///./conversations.db")
5

6
agent = create_agent(
7
    model=model,
8
    tools=tools,
9
    checkpointer=checkpointer  # 自动持久化
10
)
11

12
# 使用 thread_id 恢复对话
13
result = agent.invoke(
14
    {"messages": [{"role": "user", "content": "你好"}]},
15
    config={"configurable": {"thread_id": "user-123"}}
16
)
17

18
# 后续可以通过相同的 thread_id 继续对话

1
from langchain_core.runnables import RunnableRetry
2

3
# 自动重试（最多 3 次，指数退避）
4
retry_chain = chain.with_retry(
5
    stop_after_attempt=3,
6
    wait_exponential_jitter=True
7
)
8

9
# 自定义错误处理
10
try:
11
    result = agent.invoke(input_data)
12
except Exception as e:
13
    print(f"Agent 调用失败: {e}")
14
    fallback_result = fallback_chain.invoke(input_data)