LangChain 概述与核心架构#

1
import micropip
2
await micropip.install("langchain_openai")
3
await micropip.install("langchain_core")
4

5
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
6
from langchain_core.messages import HumanMessage, SystemMessage
7

8
# --- Chat Model ---
9
chat_model = ChatOpenAI(model="gpt-4o", temperature=0.7)
10

11
messages = [
12
    SystemMessage(content="你是一位 Python 专家。"),
13
    HumanMessage(content="请解释什么是装饰器。"),
14
]
15
response = chat_model.invoke(messages)
16
print(response.content)
17

18
# --- Embedding Model ---
19
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
20
vectors = embeddings.embed_documents(["LangChain 是什么", "如何使用向量数据库"])
21
print(f"向量维度: {len(vectors[0])}")  # 1536

1
import subprocess
2
subprocess.check_call(["pip", "install", "langchain", "langchain-openai"])
3

4
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
5

6
# --- 基础 ChatPromptTemplate ---
7
prompt = ChatPromptTemplate.from_messages([
8
    ("system", "你是一位专业的{role}。"),
9
    ("human", "{question}"),
10
])
11

12
# 渲染结果
13
formatted = prompt.format_messages(role="数据分析师", question="什么是 A/B 测试？")
14
print(formatted)
15

16
# --- Few-Shot 示例模板 ---
17
examples = [
18
    {"input": "高兴", "output": "happy"},
19
    {"input": "悲伤", "output": "sad"},
20
]
21

22
example_prompt = ChatPromptTemplate.from_messages([
23
    ("human", "{input}"),
24
    ("ai", "{output}"),
25
])
26

27
few_shot_prompt = FewShotChatMessagePromptTemplate(
28
    example_prompt=example_prompt,
29
    examples=examples,
30
)
31

32
final_prompt = ChatPromptTemplate.from_messages([
33
    ("system", "你是一位中英翻译专家，请将中文词汇翻译为英文。"),
34
    few_shot_prompt,
35
    ("human", "{input}"),
36
])
37

38
print(final_prompt.format_messages(input="愤怒"))

1
import subprocess
2
subprocess.check_call(["pip", "install", "langchain", "langchain-openai"])
3

4
from langchain_openai import ChatOpenAI
5
from langchain_core.prompts import ChatPromptTemplate
6
from langchain_core.output_parsers import StrOutputParser
7

8
# --- LCEL 管道（推荐写法） ---
9
prompt = ChatPromptTemplate.from_messages([
10
    ("system", "你是一位技术博客作者。"),
11
    ("human", "请用 200 字概括：{topic}"),
12
])
13

14
model = ChatOpenAI(model="gpt-4o", temperature=0.5)
15
output_parser = StrOutputParser()
16

17
# 用 | 管道符组合
18
chain = prompt | model | output_parser
19

20
# 调用
21
result = chain.invoke({"topic": "Transformer 架构的核心原理"})
22
print(result)
23

24
# 流式输出
25
for chunk in chain.stream({"topic": "什么是注意力机制"}):
26
    print(chunk, end="", flush=True)

1
import subprocess
2
subprocess.check_call(["pip", "install", "langchain", "langchain-openai"])
3

4
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
5

6
# --- 并行链 ---
7
parallel_chain = RunnableParallel(
8
    summary=prompt | model | output_parser,
9
    original_input=RunnablePassthrough(),
10
)
11

12
result = parallel_chain.invoke({"topic": "微服务架构"})
13
print(result["summary"])
14
print(result["original_input"])

1
import subprocess
2
subprocess.check_call(["pip", "install", "langchain", "langchain-openai", "langchain-community"])
3

4
from langchain_openai import ChatOpenAI
5
from langchain_core.tools import tool
6
from langchain.agents import create_tool_calling_agent, AgentExecutor
7
from langchain_core.prompts import ChatPromptTemplate
8

9
# 定义工具
10
@tool
11
def search_weather(city: str) -> str:
12
    """查询指定城市的天气信息。"""
13
    # 模拟天气查询
14
    weather_data = {"北京": "晴，25°C", "上海": "多云，22°C", "深圳": "阵雨，28°C"}
15
    return weather_data.get(city, f"未找到 {city} 的天气信息")
16

17
@tool
18
def calculate(expression: str) -> str:
19
    """计算数学表达式的结果。"""
20
    try:
21
        return str(eval(expression))
22
    except Exception as e:
23
        return f"计算错误: {e}"
24

25
# 创建 Agent
26
llm = ChatOpenAI(model="gpt-4o", temperature=0)
27
tools = [search_weather, calculate]
28

29
prompt = ChatPromptTemplate.from_messages([
30
    ("system", "你是一个有用的助手，可以查询天气和进行数学计算。"),
31
    ("human", "{input}"),
32
    ("placeholder", "{agent_scratchpad}"),
33
])
34

35
agent = create_tool_calling_agent(llm, tools, prompt)
36
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
37

38
# 运行
39
result = agent_executor.invoke({"input": "北京今天天气怎么样？另外帮我算一下 1024 * 768"})
40
print(result["output"])

1
import subprocess
2
subprocess.check_call(["pip", "install", "langchain", "langchain-openai"])
3

4
from langchain_openai import ChatOpenAI
5
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
6
from langchain_core.chat_history import InMemoryChatMessageHistory
7
from langchain_core.runnables.history import RunnableWithMessageHistory
8

9
# 基于 RunnableWithMessageHistory 的推荐方式
10
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
11

12
prompt = ChatPromptTemplate.from_messages([
13
    ("system", "你是一个友好的助手。"),
14
    MessagesPlaceholder(variable_name="history"),
15
    ("human", "{input}"),
16
])
17

18
chain = prompt | llm
19

20
# 存储会话历史（按 session_id 隔离）
21
store = {}
22

23
def get_session_history(session_id: str):
24
    if session_id not in store:
25
        store[session_id] = InMemoryChatMessageHistory()
26
    return store[session_id]
27

28
chain_with_history = RunnableWithMessageHistory(
29
    chain,
30
    get_session_history,
31
    input_messages_key="input",
32
    history_messages_key="history",
33
)
34

35
# 多轮对话
36
config = {"configurable": {"session_id": "user_123"}}
37

38
response1 = chain_with_history.invoke({"input": "我叫小明"}, config=config)
39
print(response1.content)
40

41
response2 = chain_with_history.invoke({"input": "我叫什么名字？"}, config=config)
42
print(response2.content)  # 模型会记住"小明"

1
文档加载 → 文本分割 → 向量化 → 存入向量数据库 → 检索
2
(Loaders)   (Splitters)  (Embeddings) (Vector Stores) (Retrievers)

1
import subprocess
2
subprocess.check_call(["pip", "install", "langchain", "langchain-community"])
3

4
from langchain_community.document_loaders import (
5
    PyPDFLoader,
6
    WebBaseLoader,
7
    TextLoader,
8
)
9

10
# 加载 PDF
11
pdf_loader = PyPDFLoader("knowledge_base.pdf")
12
pdf_docs = pdf_loader.load()
13

14
# 加载网页
15
web_loader = WebBaseLoader("https://python.langchain.com/docs/introduction/")
16
web_docs = web_loader.load()
17

18
# 每个 Document 包含 page_content（文本）和 metadata（来源、页码等）
19
print(pdf_docs[0].page_content[:200])
20
print(pdf_docs[0].metadata)

1
import subprocess
2
subprocess.check_call(["pip", "install", "langchain"])
3

4
from langchain_text_splitters import RecursiveCharacterTextSplitter
5

6
splitter = RecursiveCharacterTextSplitter(
7
    chunk_size=500,        # 每块最大字符数
8
    chunk_overlap=50,      # 相邻块重叠字符数，防止语义被截断
9
    separators=["\n\n", "\n", "。", "，", " "],  # 中文友好的分隔符
10
)
11

12
chunks = splitter.split_documents(pdf_docs)
13
print(f"原始文档数: {len(pdf_docs)}, 分割后: {len(chunks)}")

1
import subprocess
2
subprocess.check_call(["pip", "install", "langchain-openai"])
3

4
from langchain_openai import OpenAIEmbeddings
5

6
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
7

8
# 单条文本向量化
9
vector = embeddings.embed_query("什么是 LangChain？")
10
print(f"维度: {len(vector)}")  # 1536

1
import subprocess
2
subprocess.check_call(["pip", "install", "langchain-openai", "langchain-community", "faiss-cpu"])
3

4
from langchain_community.vectorstores import FAISS
5
from langchain_openai import OpenAIEmbeddings
6

7
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
8

9
# 从文档块创建向量数据库
10
vectorstore = FAISS.from_documents(chunks, embeddings)
11

12
# 相似度搜索
13
results = vectorstore.similarity_search("LangChain 的核心模块有哪些？", k=3)
14
for doc in results:
15
    print(doc.page_content[:100])
16
    print("---")

1
# 从 vectorstore 获取 retriever
2
retriever = vectorstore.as_retriever(
3
    search_type="similarity",  # 也可用 "mmr"（最大边际相关性）
4
    search_kwargs={"k": 4},
5
)
6

7
# 检索相关文档
8
docs = retriever.invoke("LangChain 支持哪些向量数据库？")