LangChain 架构与核心抽象

LangChain 究竟是什么

组件模型概览

LLM 与 ChatModels 的区别

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

llm = ChatOpenAI(model='gpt-4o-mini')
response = llm.invoke([
    SystemMessage(content='You are a helpful assistant.'),
    HumanMessage(content='What is LangChain?')
])
print(response.content)

PromptTemplates：动态提示词

from langchain_core.prompts import ChatPromptTemplate

template = ChatPromptTemplate.from_messages([
    ('system', 'You are an expert in {domain}.'),
    ('human', 'Explain {concept} in simple terms.')
])

# Fill in variables at runtime
messages = template.invoke({'domain': 'machine learning', 'concept': 'embeddings'})
print(messages)

OutputParsers：塑造模型输出

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from pydantic import BaseModel

class MovieReview(BaseModel):
    title: str
    rating: int
    summary: str

parser = JsonOutputParser(pydantic_object=MovieReview)
# parser.get_format_instructions() returns JSON schema instructions
print(parser.get_format_instructions()[:200])

可运行对象接口

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model='gpt-4o-mini')

# Single call
result = llm.invoke('What is 2+2?')

# Streaming tokens
for chunk in llm.stream('Tell me a joke.'):
    print(chunk.content, end='', flush=True)

# Batch processing
results = llm.batch(['What is Paris?', 'What is Rome?'])

使用管道运算符进行组合

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template('Summarize this in one sentence: {text}')
llm = ChatOpenAI(model='gpt-4o-mini')
parser = StrOutputParser()

# Chain composed with pipe operator
chain = prompt | llm | parser

result = chain.invoke({'text': 'LangChain is a framework for LLM applications...'})
print(result)

文档加载器与文本分割器

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

loader = PyPDFLoader('report.pdf')
docs = loader.load()  # List of Document objects

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = splitter.split_documents(docs)
print(f'Split {len(docs)} pages into {len(chunks)} chunks')

作为可运行对象的向量存储

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# Convert to Retriever Runnable
retriever = vectorstore.as_retriever(
    search_type='similarity',
    search_kwargs={'k': 5}
)

results = retriever.invoke('How does LangChain work?')
for doc in results:
    print(doc.page_content[:100])

回调与可观测性

from langchain.callbacks import StdOutCallbackHandler
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model='gpt-4o-mini',
    callbacks=[StdOutCallbackHandler()]
)
# Now every call logs input/output automatically

# Or enable LangSmith tracing globally:
import os
os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_API_KEY'] = 'your-key'

链中的记忆与状态

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

store = {}  # session_id -> ChatMessageHistory

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

with_history = RunnableWithMessageHistory(chain, get_session_history)
response = with_history.invoke(
    {'input': 'Hi, my name is Alice'},
    config={'configurable': {'session_id': 'user-1'}}
)

快速检查

课程回顾