Back to Details

langchain-framework

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

LangChain Framework

LangChain 框架

progressive_disclosure: entry_point: summary: "LLM application framework with chains, agents, RAG, and memory" when_to_use: - "When building LLM-powered applications" - "When implementing RAG (Retrieval Augmented Generation)" - "When creating AI agents with tools" - "When chaining multiple LLM calls" quick_start: - "pip install langchain langchain-anthropic" - "Set up LLM (ChatAnthropic or ChatOpenAI)" - "Create chain with prompts and LLM" - "Invoke chain with input" token_estimate: entry: 85 full: 5200

progressive_disclosure: entry_point: summary: "LLM应用框架，具备chains、agents、RAG和memory功能" when_to_use: - "构建LLM驱动的应用程序时" - "实现RAG（检索增强生成）时" - "创建带工具的AI Agent时" - "串联多个LLM调用时" quick_start: - "pip install langchain langchain-anthropic" - "设置LLM（ChatAnthropic或ChatOpenAI）" - "使用提示词和LLM创建chain" - "传入输入调用chain" token_estimate: entry: 85 full: 5200

Core Concepts

核心概念

LangChain Expression Language (LCEL)

LangChain Expression Language (LCEL)

Modern composable syntax for building chains with 
|
operator.
Basic Chain:
python
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
用于构建chain的现代化可组合语法，使用
|
运算符。
基础Chain:
python
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

Components

Components

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022") prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}") output_parser = StrOutputParser()
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022") prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}") output_parser = StrOutputParser()

Compose with LCEL

Compose with LCEL

chain = prompt | llm | output_parser
chain = prompt | llm | output_parser

Invoke

Invoke

result = chain.invoke({"topic": "programming"})

**Why LCEL**:
- Type safety and auto-completion
- Streaming support built-in
- Async by default
- Observability with LangSmith
- Easier debugging
result = chain.invoke({"topic": "programming"})

**为什么选择LCEL**:
- 类型安全与自动补全
- 内置流式支持
- 默认异步
- 借助LangSmith实现可观测性
- 更易于调试

Chain Components

Chain 组件

Prompts:
python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
提示词（Prompts）:
python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

Simple template

Simple template

prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant."), ("user", "{input}") ])
prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant."), ("user", "{input}") ])

With message history

With message history

prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant."), MessagesPlaceholder(variable_name="history"), ("user", "{input}") ])
prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant."), MessagesPlaceholder(variable_name="history"), ("user", "{input}") ])

Few-shot examples

Few-shot examples

from langchain_core.prompts import FewShotChatMessagePromptTemplate
examples = [ {"input": "2+2", "output": "4"}, {"input": "3*5", "output": "15"} ]
example_prompt = ChatPromptTemplate.from_messages([ ("human", "{input}"), ("ai", "{output}") ])
few_shot_prompt = FewShotChatMessagePromptTemplate( example_prompt=example_prompt, examples=examples )

**LLMs**:
```python
from langchain_core.prompts import FewShotChatMessagePromptTemplate
examples = [ {"input": "2+2", "output": "4"}, {"input": "3*5", "output": "15"} ]
example_prompt = ChatPromptTemplate.from_messages([ ("human", "{input}"), ("ai", "{output}") ])
few_shot_prompt = FewShotChatMessagePromptTemplate( example_prompt=example_prompt, examples=examples )

**LLMs**:
```python

Anthropic Claude

Anthropic Claude

from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic( model="claude-3-5-sonnet-20241022", temperature=0.7, max_tokens=1024, timeout=60.0 )
from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic( model="claude-3-5-sonnet-20241022", temperature=0.7, max_tokens=1024, timeout=60.0 )

OpenAI

OpenAI

from langchain_openai import ChatOpenAI
llm = ChatOpenAI( model="gpt-4-turbo-preview", temperature=0.7 )
from langchain_openai import ChatOpenAI
llm = ChatOpenAI( model="gpt-4-turbo-preview", temperature=0.7 )

Streaming

Streaming

for chunk in llm.stream("Tell me a story"): print(chunk.content, end="", flush=True)

**Output Parsers**:
```python
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
for chunk in llm.stream("Tell me a story"): print(chunk.content, end="", flush=True)

**输出解析器（Output Parsers）**:
```python
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

String parser

String parser

str_parser = StrOutputParser()
str_parser = StrOutputParser()

JSON parser

JSON parser

json_parser = JsonOutputParser()
json_parser = JsonOutputParser()

Structured output

Structured output

class Person(BaseModel): name: str = Field(description="Person's name") age: int = Field(description="Person's age")
parser = PydanticOutputParser(pydantic_object=Person) prompt = ChatPromptTemplate.from_template( "Extract person info.\n{format_instructions}\n{query}" ) chain = prompt | llm | parser
undefined
class Person(BaseModel): name: str = Field(description="Person's name") age: int = Field(description="Person's age")
parser = PydanticOutputParser(pydantic_object=Person) prompt = ChatPromptTemplate.from_template( "Extract person info.\n{format_instructions}\n{query}" ) chain = prompt | llm | parser
undefined

RAG (Retrieval Augmented Generation)

RAG（检索增强生成）

Document Loading

文档加载

python
from langchain_community.document_loaders import (
    TextLoader,
    PyPDFLoader,
    DirectoryLoader,
    WebBaseLoader
)
python
from langchain_community.document_loaders import (
    TextLoader,
    PyPDFLoader,
    DirectoryLoader,
    WebBaseLoader
)

Text files

Text files

loader = TextLoader("document.txt") docs = loader.load()
loader = TextLoader("document.txt") docs = loader.load()

PDFs

PDFs

loader = PyPDFLoader("document.pdf") docs = loader.load()
loader = PyPDFLoader("document.pdf") docs = loader.load()

Directory of files

Directory of files

loader = DirectoryLoader( "./docs", glob="**/*.md", show_progress=True ) docs = loader.load()
loader = DirectoryLoader( "./docs", glob="**/*.md", show_progress=True ) docs = loader.load()

Web pages

Web pages

loader = WebBaseLoader("https://example.com") docs = loader.load()
undefined
loader = WebBaseLoader("https://example.com") docs = loader.load()
undefined

Text Splitting

文本分割

python
from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    TokenTextSplitter
)
python
from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    TokenTextSplitter
)

Recursive splitter (recommended)

Recursive splitter (recommended)

text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, separators=["\n\n", "\n", " ", ""] )
chunks = text_splitter.split_documents(docs)
text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, separators=["\n\n", "\n", " ", ""] )
chunks = text_splitter.split_documents(docs)

Token-aware splitting

Token-aware splitting

from langchain.text_splitter import TokenTextSplitter
splitter = TokenTextSplitter( chunk_size=512, chunk_overlap=50 )
undefined
from langchain.text_splitter import TokenTextSplitter
splitter = TokenTextSplitter( chunk_size=512, chunk_overlap=50 )
undefined

Vector Stores

向量存储（Vector Stores）

python
from langchain_community.vectorstores import Chroma, FAISS, Pinecone
from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings
python
from langchain_community.vectorstores import Chroma, FAISS, Pinecone
from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings

Embeddings

Embeddings

embeddings = OpenAIEmbeddings()
embeddings = OpenAIEmbeddings()

Chroma (local, persistent)

Chroma (local, persistent)

vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db" )
vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db" )

FAISS (local, in-memory)

FAISS (local, in-memory)

vectorstore = FAISS.from_documents( documents=chunks, embedding=embeddings ) vectorstore.save_local("./faiss_index")
vectorstore = FAISS.from_documents( documents=chunks, embedding=embeddings ) vectorstore.save_local("./faiss_index")

Pinecone (cloud)

Pinecone (cloud)

from langchain_community.vectorstores import Pinecone import pinecone
pinecone.init(api_key="your-key", environment="us-west1-gcp") vectorstore = Pinecone.from_documents( documents=chunks, embedding=embeddings, index_name="langchain-index" )
undefined
from langchain_community.vectorstores import Pinecone import pinecone
pinecone.init(api_key="your-key", environment="us-west1-gcp") vectorstore = Pinecone.from_documents( documents=chunks, embedding=embeddings, index_name="langchain-index" )
undefined

RAG Chain

RAG Chain

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

Create retriever

Create retriever

retriever = vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 4} )
retriever = vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 4} )

RAG prompt

RAG prompt

template = """Answer based on context:
Context: {context}
Question: {question}
Answer:"""
prompt = ChatPromptTemplate.from_template(template)
template = """基于上下文回答问题:
上下文: {context}
问题: {question}
答案:"""
prompt = ChatPromptTemplate.from_template(template)

Format documents

Format documents

def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs)
def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs)

RAG chain

RAG chain

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

Query

Query

answer = rag_chain.invoke("What is LangChain?")
undefined
answer = rag_chain.invoke("What is LangChain?")
undefined

Advanced RAG Patterns

高级RAG模式

python
undefined
python
undefined

Multi-query retrieval

Multi-query retrieval

from langchain.retrievers.multi_query import MultiQueryRetriever
retriever = MultiQueryRetriever.from_llm( retriever=vectorstore.as_retriever(), llm=llm )
from langchain.retrievers.multi_query import MultiQueryRetriever
retriever = MultiQueryRetriever.from_llm( retriever=vectorstore.as_retriever(), llm=llm )

Contextual compression

Contextual compression

from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor
compressor = LLMChainExtractor.from_llm(llm) compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=vectorstore.as_retriever() )
from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor
compressor = LLMChainExtractor.from_llm(llm) compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=vectorstore.as_retriever() )

Parent document retriever

Parent document retriever

from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore
store = InMemoryStore() retriever = ParentDocumentRetriever( vectorstore=vectorstore, docstore=store, child_splitter=text_splitter )
undefined
from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore
store = InMemoryStore() retriever = ParentDocumentRetriever( vectorstore=vectorstore, docstore=store, child_splitter=text_splitter )
undefined

Agents and Tools

Agents与工具

Tool Creation

工具创建

python
from langchain.tools import tool
from langchain_core.tools import Tool
python
from langchain.tools import tool
from langchain_core.tools import Tool

Decorator approach

Decorator approach

@tool def search_wikipedia(query: str) -> str: """Search Wikipedia for information.""" # Implementation return f"Results for: {query}"
@tool def search_wikipedia(query: str) -> str: """Search Wikipedia for information.""" # Implementation return f"Results for: {query}"

Class approach

Class approach

from langchain.tools import BaseTool from pydantic import BaseModel, Field
class CalculatorInput(BaseModel): expression: str = Field(description="Mathematical expression")
class CalculatorTool(BaseTool): name = "calculator" description = "Useful for math calculations" args_schema = CalculatorInput
def _run(self, expression: str) -> str:
    return str(eval(expression))
from langchain.tools import BaseTool from pydantic import BaseModel, Field
class CalculatorInput(BaseModel): expression: str = Field(description="Mathematical expression")
class CalculatorTool(BaseTool): name = "calculator" description = "Useful for math calculations" args_schema = CalculatorInput
def _run(self, expression: str) -> str:
    return str(eval(expression))

Pre-built tools

Pre-built tools

from langchain_community.tools import ( DuckDuckGoSearchRun, WikipediaQueryRun, PythonREPLTool )
search = DuckDuckGoSearchRun() wikipedia = WikipediaQueryRun() python_repl = PythonREPLTool()
undefined
from langchain_community.tools import ( DuckDuckGoSearchRun, WikipediaQueryRun, PythonREPLTool )
search = DuckDuckGoSearchRun() wikipedia = WikipediaQueryRun() python_repl = PythonREPLTool()
undefined

Agent Types

Agent类型

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

ReAct agent (recommended)

ReAct agent (recommended)

prompt = hub.pull("hwchase17/react") tools = [search_wikipedia, CalculatorTool()]
agent = create_react_agent(llm, tools, prompt) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=3, handle_parsing_errors=True )
result = agent_executor.invoke({"input": "What is 2+2 and who invented addition?"})
prompt = hub.pull("hwchase17/react") tools = [search_wikipedia, CalculatorTool()]
agent = create_react_agent(llm, tools, prompt) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=3, handle_parsing_errors=True )
result = agent_executor.invoke({"input": "What is 2+2 and who invented addition?"})

Structured chat agent (function calling)

Structured chat agent (function calling)

from langchain.agents import create_structured_chat_agent
agent = create_structured_chat_agent(llm, tools, prompt)
from langchain.agents import create_structured_chat_agent
agent = create_structured_chat_agent(llm, tools, prompt)

OpenAI functions agent

OpenAI functions agent

from langchain.agents import create_openai_functions_agent
agent = create_openai_functions_agent(llm, tools, prompt)
undefined
from langchain.agents import create_openai_functions_agent
agent = create_openai_functions_agent(llm, tools, prompt)
undefined

Agent with Memory

带记忆的Agent

python
from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)
python
from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

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

Conversational loop

Conversational loop

while True: user_input = input("You: ") if user_input.lower() == "exit": break response = agent_executor.invoke({"input": user_input}) print(f"Agent: {response['output']}")
undefined
while True: user_input = input("You: ") if user_input.lower() == "exit": break response = agent_executor.invoke({"input": user_input}) print(f"Agent: {response['output']}")
undefined

Memory Systems

记忆系统

Memory Types

记忆类型

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

Full conversation history

Full conversation history

memory = ConversationBufferMemory(return_messages=True)
memory = ConversationBufferMemory(return_messages=True)

Last K messages

Last K messages

memory = ConversationBufferWindowMemory(k=5, return_messages=True)
memory = ConversationBufferWindowMemory(k=5, return_messages=True)

Summarized history

Summarized history

memory = ConversationSummaryMemory(llm=llm, return_messages=True)
memory = ConversationSummaryMemory(llm=llm, return_messages=True)

Summary + recent buffer

Summary + recent buffer

memory = ConversationSummaryBufferMemory( llm=llm, max_token_limit=100, return_messages=True )
undefined
memory = ConversationSummaryBufferMemory( llm=llm, max_token_limit=100, return_messages=True )
undefined

Conversation Chain with Memory

带记忆的对话Chain

python
from langchain.chains import ConversationChain

conversation = ConversationChain(
    llm=llm,
    memory=ConversationBufferMemory()
)

conversation.predict(input="Hi, I'm Alice")
conversation.predict(input="What's my name?")  # "Alice"
python
from langchain.chains import ConversationChain

conversation = ConversationChain(
    llm=llm,
    memory=ConversationBufferMemory()
)

conversation.predict(input="Hi, I'm Alice")
conversation.predict(input="What's my name?")  # "Alice"

Custom prompt with memory

Custom prompt with memory

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant."), MessagesPlaceholder(variable_name="history"), ("human", "{input}") ])
chain = prompt | llm | StrOutputParser()
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant."), MessagesPlaceholder(variable_name="history"), ("human", "{input}") ])
chain = prompt | llm | StrOutputParser()

Manual memory management

Manual memory management

from langchain_core.messages import HumanMessage, AIMessage
history = []
def chat(user_input): response = chain.invoke({"input": user_input, "history": history}) history.append(HumanMessage(content=user_input)) history.append(AIMessage(content=response)) return response
undefined
from langchain_core.messages import HumanMessage, AIMessage
history = []
def chat(user_input): response = chain.invoke({"input": user_input, "history": history}) history.append(HumanMessage(content=user_input)) history.append(AIMessage(content=response)) return response
undefined

Advanced Chain Patterns

高级Chain模式

Sequential Chains

链式Chain（Sequential Chains）

python
from langchain.chains import SequentialChain, LLMChain
python
from langchain.chains import SequentialChain, LLMChain

Step 1: Generate synopsis

Step 1: Generate synopsis

synopsis_chain = LLMChain( llm=llm, prompt=ChatPromptTemplate.from_template("Write synopsis for: {title}"), output_key="synopsis" )
synopsis_chain = LLMChain( llm=llm, prompt=ChatPromptTemplate.from_template("Write synopsis for: {title}"), output_key="synopsis" )

Step 2: Generate review

Step 2: Generate review

review_chain = LLMChain( llm=llm, prompt=ChatPromptTemplate.from_template("Review this synopsis: {synopsis}"), output_key="review" )
review_chain = LLMChain( llm=llm, prompt=ChatPromptTemplate.from_template("Review this synopsis: {synopsis}"), output_key="review" )

Combine

Combine

overall_chain = SequentialChain( chains=[synopsis_chain, review_chain], input_variables=["title"], output_variables=["synopsis", "review"], verbose=True )
result = overall_chain({"title": "AI Revolution"})
undefined
overall_chain = SequentialChain( chains=[synopsis_chain, review_chain], input_variables=["title"], output_variables=["synopsis", "review"], verbose=True )
result = overall_chain({"title": "AI Revolution"})
undefined

Router Chains

路由Chain（Router Chains）

python
from langchain.chains.router import MultiPromptChain
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
python
from langchain.chains.router import MultiPromptChain
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser

Define specialized prompts

Define specialized prompts

physics_template = """You are a physics expert. Answer: {input}""" math_template = """You are a math expert. Answer: {input}"""
prompt_infos = [ { "name": "physics", "description": "Good for physics questions", "prompt_template": physics_template }, { "name": "math", "description": "Good for math questions", "prompt_template": math_template } ]
physics_template = """You are a physics expert. Answer: {input}""" math_template = """You are a math expert. Answer: {input}"""
prompt_infos = [ { "name": "physics", "description": "Good for physics questions", "prompt_template": physics_template }, { "name": "math", "description": "Good for math questions", "prompt_template": math_template } ]

Create router

Create router

from langchain.chains.router.multi_prompt_prompt import MULTI_PROMPT_ROUTER_TEMPLATE
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(destinations="\n".join( [f"{p['name']}: {p['description']}" for p in prompt_infos] ))
router_prompt = ChatPromptTemplate.from_template(router_template) router_chain = LLMRouterChain.from_llm(llm, router_prompt)
from langchain.chains.router.multi_prompt_prompt import MULTI_PROMPT_ROUTER_TEMPLATE
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(destinations="\n".join( [f"{p['name']}: {p['description']}" for p in prompt_infos] ))
router_prompt = ChatPromptTemplate.from_template(router_template) router_chain = LLMRouterChain.from_llm(llm, router_prompt)

Build multi-prompt chain

Build multi-prompt chain

chain = MultiPromptChain( router_chain=router_chain, destination_chains={ "physics": LLMChain(llm=llm, prompt=ChatPromptTemplate.from_template(physics_template)), "math": LLMChain(llm=llm, prompt=ChatPromptTemplate.from_template(math_template)) }, default_chain=LLMChain(llm=llm, prompt=ChatPromptTemplate.from_template("{input}")), verbose=True )
undefined
chain = MultiPromptChain( router_chain=router_chain, destination_chains={ "physics": LLMChain(llm=llm, prompt=ChatPromptTemplate.from_template(physics_template)), "math": LLMChain(llm=llm, prompt=ChatPromptTemplate.from_template(math_template)) }, default_chain=LLMChain(llm=llm, prompt=ChatPromptTemplate.from_template("{input}")), verbose=True )
undefined

Parallel Execution

并行执行

python
from langchain_core.runnables import RunnableParallel
python
from langchain_core.runnables import RunnableParallel

Execute multiple chains in parallel

Execute multiple chains in parallel

parallel_chain = RunnableParallel( summary=summary_chain, translation=translation_chain, sentiment=sentiment_chain )
result = parallel_chain.invoke({"text": "Long article text..."})
parallel_chain = RunnableParallel( summary=summary_chain, translation=translation_chain, sentiment=sentiment_chain )
result = parallel_chain.invoke({"text": "Long article text..."})

Returns: {"summary": "...", "translation": "...", "sentiment": "..."}

Returns: {"summary": "...", "translation": "...", "sentiment": "..."}

undefined
undefined

Async Patterns

异步模式

Async Chains

异步Chain

python
import asyncio
python
import asyncio

Async invoke

Async invoke

async def process(): result = await chain.ainvoke({"input": "Hello"}) return result
async def process(): result = await chain.ainvoke({"input": "Hello"}) return result

Async streaming

Async streaming

async def stream(): async for chunk in chain.astream({"input": "Tell me a story"}): print(chunk, end="", flush=True)
async def stream(): async for chunk in chain.astream({"input": "Tell me a story"}): print(chunk.content, end="", flush=True)

Async batch

Async batch

async def batch(): results = await chain.abatch([ {"input": "Question 1"}, {"input": "Question 2"} ]) return results
async def batch(): results = await chain.abatch([ {"input": "Question 1"}, {"input": "Question 2"} ]) return results

Run

Run

asyncio.run(process())
undefined
asyncio.run(process())
undefined

Concurrent Processing

并发处理

python
from langchain_core.runnables import RunnablePassthrough

async def process_documents(docs):
    # Process multiple documents concurrently
    tasks = [chain.ainvoke({"doc": doc}) for doc in docs]
    results = await asyncio.gather(*tasks)
    return results
python
from langchain_core.runnables import RunnablePassthrough

async def process_documents(docs):
    # Process multiple documents concurrently
    tasks = [chain.ainvoke({"doc": doc}) for doc in docs]
    results = await asyncio.gather(*tasks)
    return results

With rate limiting

With rate limiting

from langchain.callbacks import get_openai_callback
async def process_with_limits(docs, max_concurrent=5): semaphore = asyncio.Semaphore(max_concurrent)
async def process_one(doc):
    async with semaphore:
        return await chain.ainvoke({"doc": doc})

tasks = [process_one(doc) for doc in docs]
return await asyncio.gather(*tasks)
undefined
from langchain.callbacks import get_openai_callback
async def process_with_limits(docs, max_concurrent=5): semaphore = asyncio.Semaphore(max_concurrent)
async def process_one(doc):
    async with semaphore:
        return await chain.ainvoke({"doc": doc})

tasks = [process_one(doc) for doc in docs]
return await asyncio.gather(*tasks)
undefined

LangSmith Tracing

LangSmith 追踪

Setup and Configuration

设置与配置

python
import os
python
import os

Enable LangSmith

Enable LangSmith

os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key" os.environ["LANGCHAIN_PROJECT"] = "my-project"
os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key" os.environ["LANGCHAIN_PROJECT"] = "my-project"

Trace automatically captures all LangChain operations

Trace automatically captures all LangChain operations

result = chain.invoke({"input": "Hello"})
result = chain.invoke({"input": "Hello"})

View trace at: https://smith.langchain.com

View trace at: https://smith.langchain.com

undefined
undefined

Custom Tracing

自定义追踪

python
from langsmith import trace

@trace
def my_function(input_text):
    # Custom function tracing
    result = chain.invoke({"input": input_text})
    return result
python
from langsmith import trace

@trace
def my_function(input_text):
    # Custom function tracing
    result = chain.invoke({"input": input_text})
    return result

Add metadata

Add metadata

from langchain.callbacks import LangChainTracer
tracer = LangChainTracer( project_name="my-project", metadata={"environment": "production", "version": "1.0"} )
chain.invoke({"input": "Hello"}, config={"callbacks": [tracer]})
undefined
from langchain.callbacks import LangChainTracer
tracer = LangChainTracer( project_name="my-project", metadata={"environment": "production", "version": "1.0"} )
chain.invoke({"input": "Hello"}, config={"callbacks": [tracer]})
undefined

Evaluation

评估

python
from langsmith import Client
from langchain.evaluation import load_evaluator

client = Client()
python
from langsmith import Client
from langchain.evaluation import load_evaluator

client = Client()

Create dataset

Create dataset

dataset = client.create_dataset("my-dataset") client.create_examples( inputs=[{"input": "What is AI?"}], outputs=[{"output": "Artificial Intelligence..."}], dataset_id=dataset.id )
dataset = client.create_dataset("my-dataset") client.create_examples( inputs=[{"input": "What is AI?"}], outputs=[{"output": "Artificial Intelligence..."}], dataset_id=dataset.id )

Evaluate

Evaluate

def predict(input_dict): return chain.invoke(input_dict)
def predict(input_dict): return chain.invoke(input_dict)

Run evaluation

Run evaluation

results = client.run_on_dataset( dataset_name="my-dataset", llm_or_chain_factory=lambda: chain, evaluation=load_evaluator("qa"), project_name="my-evaluation" )
undefined
results = client.run_on_dataset( dataset_name="my-dataset", llm_or_chain_factory=lambda: chain, evaluation=load_evaluator("qa"), project_name="my-evaluation" )
undefined

Production Deployment

生产部署

Error Handling

错误处理

python
from langchain_core.runnables import RunnableWithFallbacks
python
from langchain_core.runnables import RunnableWithFallbacks

Fallback chain

Fallback chain

primary_llm = ChatAnthropic(model="claude-3-5-sonnet-20241022") fallback_llm = ChatOpenAI(model="gpt-4-turbo-preview")
chain = (prompt | primary_llm).with_fallbacks([prompt | fallback_llm])
primary_llm = ChatAnthropic(model="claude-3-5-sonnet-20241022") fallback_llm = ChatOpenAI(model="gpt-4-turbo-preview")
chain = (prompt | primary_llm).with_fallbacks([prompt | fallback_llm])

Retry logic

Retry logic

from langchain_core.runnables import RunnableRetry
chain_with_retry = chain.with_retry( retry_if_exception_type=(RateLimitError,), wait_exponential_jitter=True, stop_after_attempt=3 )
from langchain_core.runnables import RunnableRetry
chain_with_retry = chain.with_retry( retry_if_exception_type=(RateLimitError,), wait_exponential_jitter=True, stop_after_attempt=3 )

Error handling

Error handling

try: result = chain.invoke({"input": "Hello"}) except Exception as e: logger.error(f"Chain failed: {e}") # Handle gracefully
undefined
try: result = chain.invoke({"input": "Hello"}) except Exception as e: logger.error(f"Chain failed: {e}") # Handle gracefully
undefined

Caching

缓存

python
from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache
python
from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache

In-memory cache

In-memory cache

set_llm_cache(InMemoryCache())
set_llm_cache(InMemoryCache())

Persistent cache

Persistent cache

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

Redis cache

Redis cache

from langchain.cache import RedisCache import redis
set_llm_cache(RedisCache(redis_=redis.Redis()))
from langchain.cache import RedisCache import redis
set_llm_cache(RedisCache(redis_=redis.Redis()))

Semantic cache

Semantic cache

from langchain.cache import RedisSemanticCache from langchain_openai import OpenAIEmbeddings
set_llm_cache(RedisSemanticCache( redis_url="redis://localhost:6379", embedding=OpenAIEmbeddings(), score_threshold=0.8 ))
undefined
from langchain.cache import RedisSemanticCache from langchain_openai import OpenAIEmbeddings
set_llm_cache(RedisSemanticCache( redis_url="redis://localhost:6379", embedding=OpenAIEmbeddings(), score_threshold=0.8 ))
undefined

Rate Limiting

速率限制

python
from langchain.llms.base import BaseLLM
from ratelimit import limits, sleep_and_retry

class RateLimitedLLM(BaseLLM):
    @sleep_and_retry
    @limits(calls=50, period=60)  # 50 calls per minute
    def _call(self, prompt, stop=None, **kwargs):
        return self.llm._call(prompt, stop, **kwargs)
python
from langchain.llms.base import BaseLLM
from ratelimit import limits, sleep_and_retry

class RateLimitedLLM(BaseLLM):
    @sleep_and_retry
    @limits(calls=50, period=60)  # 50 calls per minute
    def _call(self, prompt, stop=None, **kwargs):
        return self.llm._call(prompt, stop, **kwargs)

Token budget tracking

Token budget tracking

from langchain.callbacks import get_openai_callback
with get_openai_callback() as cb: result = chain.invoke({"input": "Hello"}) print(f"Tokens used: {cb.total_tokens}") print(f"Cost: ${cb.total_cost}")
undefined
from langchain.callbacks import get_openai_callback
with get_openai_callback() as cb: result = chain.invoke({"input": "Hello"}) print(f"Tokens used: {cb.total_tokens}") print(f"Cost: ${cb.total_cost}")
undefined

Monitoring

监控

python
from langchain.callbacks.base import BaseCallbackHandler

class MetricsCallback(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        # Log LLM start
        logger.info(f"LLM started: {prompts}")

    def on_llm_end(self, response, **kwargs):
        # Log LLM completion
        logger.info(f"LLM completed: {response}")

    def on_llm_error(self, error, **kwargs):
        # Log errors
        logger.error(f"LLM error: {error}")
python
from langchain.callbacks.base import BaseCallbackHandler

class MetricsCallback(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        # Log LLM start
        logger.info(f"LLM started: {prompts}")

    def on_llm_end(self, response, **kwargs):
        # Log LLM completion
        logger.info(f"LLM completed: {response}")

    def on_llm_error(self, error, **kwargs):
        # Log errors
        logger.error(f"LLM error: {error}")

Use callback

Use callback

chain.invoke({"input": "Hello"}, config={"callbacks": [MetricsCallback()]})
undefined
chain.invoke({"input": "Hello"}, config={"callbacks": [MetricsCallback()]})
undefined

Best Practices

最佳实践

Code Organization

代码组织

python
undefined
python
undefined

chains.py

chains.py

from langchain_anthropic import ChatAnthropic from langchain_core.prompts import ChatPromptTemplate
def create_summarization_chain(): llm = ChatAnthropic(model="claude-3-5-sonnet-20241022") prompt = ChatPromptTemplate.from_template("Summarize: {text}") return prompt | llm
from langchain_anthropic import ChatAnthropic from langchain_core.prompts import ChatPromptTemplate
def create_summarization_chain(): llm = ChatAnthropic(model="claude-3-5-sonnet-20241022") prompt = ChatPromptTemplate.from_template("Summarize: {text}") return prompt | llm

main.py

main.py

from chains import create_summarization_chain
chain = create_summarization_chain()
undefined
from chains import create_summarization_chain
chain = create_summarization_chain()
undefined

Environment Configuration

环境配置

python
undefined
python
undefined

config.py

config.py

from pydantic_settings import BaseSettings
class Settings(BaseSettings): anthropic_api_key: str langsmith_api_key: str | None = None langsmith_project: str = "default" model_name: str = "claude-3-5-sonnet-20241022" temperature: float = 0.7
class Config:
    env_file = ".env"
settings = Settings()
from pydantic_settings import BaseSettings
class Settings(BaseSettings): anthropic_api_key: str langsmith_api_key: str | None = None langsmith_project: str = "default" model_name: str = "claude-3-5-sonnet-20241022" temperature: float = 0.7
class Config:
    env_file = ".env"
settings = Settings()

Use in chains

Use in chains

llm = ChatAnthropic( model=settings.model_name, temperature=settings.temperature, anthropic_api_key=settings.anthropic_api_key )
undefined
llm = ChatAnthropic( model=settings.model_name, temperature=settings.temperature, anthropic_api_key=settings.anthropic_api_key )
undefined

Testing

测试

python
import pytest
from langchain_core.prompts import ChatPromptTemplate

def test_chain_output():
    # Use mock LLM for testing
    from langchain.llms.fake import FakeListLLM

    llm = FakeListLLM(responses=["Mocked response"])
    prompt = ChatPromptTemplate.from_template("Test: {input}")
    chain = prompt | llm

    result = chain.invoke({"input": "test"})
    assert result == "Mocked response"
python
import pytest
from langchain_core.prompts import ChatPromptTemplate

def test_chain_output():
    # Use mock LLM for testing
    from langchain.llms.fake import FakeListLLM

    llm = FakeListLLM(responses=["Mocked response"])
    prompt = ChatPromptTemplate.from_template("Test: {input}")
    chain = prompt | llm

    result = chain.invoke({"input": "test"})
    assert result == "Mocked response"

Integration test

Integration test

@pytest.mark.integration def test_real_chain(): chain = create_summarization_chain() result = chain.invoke({"text": "Long text..."}) assert len(result) > 0
undefined
@pytest.mark.integration def test_real_chain(): chain = create_summarization_chain() result = chain.invoke({"text": "Long text..."}) assert len(result) > 0
undefined

Common Pitfalls

常见陷阱

Memory Leaks

内存泄漏

python
undefined
python
undefined

❌ WRONG: Memory grows unbounded

❌ WRONG: Memory grows unbounded

memory = ConversationBufferMemory() while True: chain.invoke({"input": user_input}) # History never cleared
memory = ConversationBufferMemory() while True: chain.invoke({"input": user_input}) # History never cleared

✅ CORRECT: Use windowed memory

✅ CORRECT: Use windowed memory

memory = ConversationBufferWindowMemory(k=10)
memory = ConversationBufferWindowMemory(k=10)

Or clear periodically

Or clear periodically

if len(memory.chat_memory.messages) > 100: memory.clear()
undefined
if len(memory.chat_memory.messages) > 100: memory.clear()
undefined

Inefficient Retrieval

低效检索

python
undefined
python
undefined

❌ WRONG: Retrieving too many documents

❌ WRONG: Retrieving too many documents

retriever = vectorstore.as_retriever(search_kwargs={"k": 100})
retriever = vectorstore.as_retriever(search_kwargs={"k": 100})

✅ CORRECT: Optimize k value

✅ CORRECT: Optimize k value

retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

✅ BETTER: Use MMR for diversity

✅ BETTER: Use MMR for diversity

retriever = vectorstore.as_retriever( search_type="mmr", search_kwargs={"k": 4, "fetch_k": 20} )
undefined
retriever = vectorstore.as_retriever( search_type="mmr", search_kwargs={"k": 4, "fetch_k": 20} )
undefined

Blocking Async

阻塞异步

python
undefined
python
undefined

❌ WRONG: Blocking in async context

❌ WRONG: Blocking in async context

async def process(): result = chain.invoke({"input": "test"}) # Blocks event loop
async def process(): result = chain.invoke({"input": "test"}) # Blocks event loop

✅ CORRECT: Use async methods

✅ CORRECT: Use async methods

async def process(): result = await chain.ainvoke({"input": "test"})
undefined
async def process(): result = await chain.ainvoke({"input": "test"})
undefined

Unstructured Outputs

非结构化输出

python
undefined
python
undefined

❌ WRONG: Parsing string outputs

❌ WRONG: Parsing string outputs

result = chain.invoke({"input": "Extract name and age"})
result = chain.invoke({"input": "Extract name and age"})

Then: parse result string manually

Then: parse result string manually

✅ CORRECT: Use structured output

✅ CORRECT: Use structured output

from langchain.output_parsers import PydanticOutputParser
parser = PydanticOutputParser(pydantic_object=Person) chain = prompt | llm | parser result = chain.invoke({"input": "Extract name and age"})
from langchain.output_parsers import PydanticOutputParser
parser = PydanticOutputParser(pydantic_object=Person) chain = prompt | llm | parser result = chain.invoke({"input": "Extract name and age"})

Returns: Person(name="John", age=30)

Returns: Person(name="John", age=30)

undefined
undefined

Token Waste

Token浪费

python
undefined
python
undefined

❌ WRONG: Sending full context every time

❌ WRONG: Sending full context every time

for question in questions: chain.invoke({"context": long_document, "question": question})
for question in questions: chain.invoke({"context": long_document, "question": question})

✅ CORRECT: Use RAG retrieval

✅ CORRECT: Use RAG retrieval

for question in questions: relevant_docs = retriever.get_relevant_documents(question) chain.invoke({"context": relevant_docs, "question": question})
undefined
for question in questions: relevant_docs = retriever.get_relevant_documents(question) chain.invoke({"context": relevant_docs, "question": question})
undefined

Quick Reference

快速参考

Installation:
bash
pip install langchain langchain-anthropic langchain-openai
pip install chromadb faiss-cpu  # Vector stores
pip install langsmith  # Observability
Essential Imports:
python
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
Basic Chain: 
prompt | llm | parser
RAG Chain: 
retriever | format_docs | prompt | llm
With Memory: Use 
MessagesPlaceholder
in prompt + memory object
Async: Replace 
invoke
with 
ainvoke
, 
stream
with 
astream
Debugging: Set 
verbose=True
or enable LangSmith tracing
Production: Add error handling, caching, rate limiting, monitoring
安装:
bash
pip install langchain langchain-anthropic langchain-openai
pip install chromadb faiss-cpu  # Vector stores
pip install langsmith  # Observability
核心导入:
python
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
基础Chain: 
prompt | llm | parser
RAG Chain: 
retriever | format_docs | prompt | llm
带记忆: 在提示词中使用
MessagesPlaceholder
+ 记忆对象
异步: 将
invoke
替换为
ainvoke
，
stream
替换为
astream
调试: 设置
verbose=True
或启用LangSmith追踪
生产环境: 添加错误处理、缓存、速率限制、监控