# Langchain Architecture

> 掌握 LangChain 框架，构建包含智能体、链、记忆和工具集成的高级 LLM 应用。触发词：LangChain架构、LangChain开发、LLM应用开发、智能体链、对话记忆、工具集成、RAG管道、文档处理、LangChain模式

- Skill: `kscz0000/langchain-architecture` (Agent Skill)
- Install (CLI): `npx skillmds@latest add kscz0000/langchain-architecture`
- Raw SKILL.md: https://api.skillmd.com/api/skills/kscz0000/langchain-architecture/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: kscz0000 (https://skillmd.com/u/kscz0000)
- Updated: 2026-09-21
- Page: https://skillmd.com/skills/kscz0000/langchain-architecture

---


# LangChain 架构

掌握 LangChain 框架，构建包含智能体、链、记忆和工具集成的高级 LLM 应用。

## 不使用此技能的情况

- 任务与 LangChain 架构无关
- 需要此范围之外的其他领域或工具

## 使用说明

- 明确目标、约束和所需输入。
- 应用相关最佳实践并验证结果。
- 提供可操作的步骤和验证方法。
- 如需详细示例，请打开 ``resources/implementation-playbook.md``。

## 使用此技能的情况

- 构建具有工具访问能力的自主 AI 智能体
- 实现复杂的多步骤 LLM 工作流
- 管理对话记忆和状态
- 将 LLM 与外部数据源和 API 集成
- 创建模块化、可复用的 LLM 应用组件
- 实现文档处理管道
- 构建生产级 LLM 应用

## 核心概念

### 1. 智能体（Agents）
使用 LLM 决定采取哪些行动的自主系统。

**智能体类型：**
- **ReAct**：以交替方式进行推理和行动
- **OpenAI Functions**：利用函数调用 API
- **Structured Chat**：处理多输入工具
- **Conversational**：针对聊天界面优化
- **Self-Ask with Search**：分解复杂查询

### 2. 链（Chains）
对 LLM 或其他工具的调用序列。

**链类型：**
- **LLMChain**：基础提示词 + LLM 组合
- **SequentialChain**：多个链顺序执行
- **RouterChain**：将输入路由到专用链
- **TransformChain**：步骤间的数据转换
- **MapReduceChain**：并行处理与聚合

### 3. 记忆（Memory）
在交互之间维护上下文的系统。

**记忆类型：**
- **ConversationBufferMemory**：存储所有消息
- **ConversationSummaryMemory**：摘要旧消息
- **ConversationBufferWindowMemory**：保留最近 N 条消息
- **EntityMemory**：跟踪实体信息
- **VectorStoreMemory**：语义相似度检索

### 4. 文档处理
加载、转换和存储文档以供检索。

**组件：**
- **Document Loaders**：从多种来源加载
- **Text Splitters**：智能分块文档
- **Vector Stores**：存储和检索嵌入
- **Retrievers**：获取相关文档
- **Indexes**：组织文档以高效访问

### 5. 回调（Callbacks）
用于日志记录、监控和调试的钩子。

**使用场景：**
- 请求/响应日志记录
- Token 使用量追踪
- 延迟监控
- 错误处理
- 自定义指标收集

## 快速开始

```python
from langchain.agents import AgentType, initialize_agent, load_tools
from langchain.llms import OpenAI
from langchain.memory import ConversationBufferMemory

# Initialize LLM
llm = OpenAI(temperature=0)

# Load tools
tools = load_tools(["serpapi", "llm-math"], llm=llm)

# Add memory
memory = ConversationBufferMemory(memory_key="chat_history")

# Create agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    verbose=True
)

# Run agent
result = agent.run("What's the weather in SF? Then calculate 25 * 4")
```

## 架构模式

### 模式 1：使用 LangChain 实现 RAG
```python
from langchain.chains import RetrievalQA
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# Load and process documents
loader = TextLoader('documents.txt')
documents = loader.load()

text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
texts = text_splitter.split_documents(documents)

# Create vector store
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(texts, embeddings)

# Create retrieval chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(),
    return_source_documents=True
)

# Query
result = qa_chain({"query": "What is the main topic?"})
```

### 模式 2：带工具的自定义智能体
```python
from langchain.agents import Tool, AgentExecutor
from langchain.agents.react.base import ReActDocstoreAgent
from langchain.tools import tool

@tool
def search_database(query: str) -> str:
    """Search internal database for information."""
    # Your database search logic
    return f"Results for: {query}"

@tool
def send_email(recipient: str, content: str) -> str:
    """Send an email to specified recipient."""
    # Email sending logic
    return f"Email sent to {recipient}"

tools = [search_database, send_email]

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)
```

### 模式 3：多步骤链
```python
from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import PromptTemplate

# Step 1: Extract key information
extract_prompt = PromptTemplate(
    input_variables=["text"],
    template="Extract key entities from: {text}\n\nEntities:"
)
extract_chain = LLMChain(llm=llm, prompt=extract_prompt, output_key="entities")

# Step 2: Analyze entities
analyze_prompt = PromptTemplate(
    input_variables=["entities"],
    template="Analyze these entities: {entities}\n\nAnalysis:"
)
analyze_chain = LLMChain(llm=llm, prompt=analyze_prompt, output_key="analysis")

# Step 3: Generate summary
summary_prompt = PromptTemplate(
    input_variables=["entities", "analysis"],
    template="Summarize:\nEntities: {entities}\nAnalysis: {analysis}\n\nSummary:"
)
summary_chain = LLMChain(llm=llm, prompt=summary_prompt, output_key="summary")

# Combine into sequential chain
overall_chain = SequentialChain(
    chains=[extract_chain, analyze_chain, summary_chain],
    input_variables=["text"],
    output_variables=["entities", "analysis", "summary"],
    verbose=True
)
```

## 记忆管理最佳实践

### 选择合适的记忆类型
```python
# For short conversations (< 10 messages)
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()

# For long conversations (summarize old messages)
from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(llm=llm)

# For sliding window (last N messages)
from langchain.memory import ConversationBufferWindowMemory
memory = ConversationBufferWindowMemory(k=5)

# For entity tracking
from langchain.memory import ConversationEntityMemory
memory = ConversationEntityMemory(llm=llm)

# For semantic retrieval of relevant history
from langchain.memory import VectorStoreRetrieverMemory
memory = VectorStoreRetrieverMemory(retriever=retriever)
```

## 回调系统

### 自定义回调处理器
```python
from langchain.callbacks.base import BaseCallbackHandler

class CustomCallbackHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"LLM started with prompts: {prompts}")

    def on_llm_end(self, response, **kwargs):
        print(f"LLM ended with response: {response}")

    def on_llm_error(self, error, **kwargs):
        print(f"LLM error: {error}")

    def on_chain_start(self, serialized, inputs, **kwargs):
        print(f"Chain started with inputs: {inputs}")

    def on_agent_action(self, action, **kwargs):
        print(f"Agent taking action: {action}")

# Use callback
agent.run("query", callbacks=[CustomCallbackHandler()])
```

## 测试策略

```python
import pytest
from unittest.mock import Mock

def test_agent_tool_selection():
    # Mock LLM to return specific tool selection
    mock_llm = Mock()
    mock_llm.predict.return_value = "Action: search_database\nAction Input: test query"

    agent = initialize_agent(tools, mock_llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION)

    result = agent.run("test query")

    # Verify correct tool was selected
    assert "search_database" in str(mock_llm.predict.call_args)

def test_memory_persistence():
    memory = ConversationBufferMemory()

    memory.save_context({"input": "Hi"}, {"output": "Hello!"})

    assert "Hi" in memory.load_memory_variables({})['history']
    assert "Hello!" in memory.load_memory_variables({})['history']
```

## 性能优化

### 1. 缓存
```python
from langchain.cache import InMemoryCache
import langchain

langchain.llm_cache = InMemoryCache()
```

### 2. 批量处理
```python
# Process multiple documents in parallel
from langchain.document_loaders import DirectoryLoader
from concurrent.futures import ThreadPoolExecutor

loader = DirectoryLoader('./docs')
docs = loader.load()

def process_doc(doc):
    return text_splitter.split_documents([doc])

with ThreadPoolExecutor(max_workers=4) as executor:
    split_docs = list(executor.map(process_doc, docs))
```

### 3. 流式响应
```python
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

llm = OpenAI(streaming=True, callbacks=[StreamingStdOutCallbackHandler()])
```

## 资源

- **references/agents.md**：智能体架构深入解析
- **references/memory.md**：记忆系统模式
- **references/chains.md**：链组合策略
- **references/document-processing.md**：文档加载与索引
- **references/callbacks.md**：监控与可观测性
- **assets/agent-template.py**：生产级智能体模板
- **assets/memory-config.yaml**：记忆配置示例
- **assets/chain-example.py**：复杂链示例

## 常见陷阱

1. **内存溢出**：未管理对话历史长度
2. **工具选择错误**：糟糕的工具描述使智能体困惑
3. **超出上下文窗口**：超过 LLM token 限制
4. **无错误处理**：未捕获和处理智能体失败
5. **低效检索**：未优化向量存储查询

## 生产环境检查清单

- [ ] 实现适当的错误处理
- [ ] 添加请求/响应日志记录
- [ ] 监控 token 使用量和成本
- [ ] 设置智能体执行的超时限制
- [ ] 实现速率限制
- [ ] 添加输入验证
- [ ] 测试边界情况
- [ ] 设置可观测性（回调）
- [ ] 实现降级策略
- [ ] 版本控制提示词和配置

## 限制
- 仅当任务明确符合上述描述的范围时，才使用此技能。
- 不要将输出替代特定环境的验证、测试或专家审查。
- 如果缺少所需输入、权限、安全边界或成功标准，请停下来请求澄清。