langchain-architecture

支持材料可用性

此目录副本包含以下内联指南，但不包含以下支持路径。这些是外部前置条件，并非随附文件：

• <support-root>/references/agents.md
• <support-root>/references/memory.md
• <support-root>/references/chains.md
• <support-root>/references/document-processing.md
• <support-root>/references/callbacks.md
• <support-root>/assets/agent-template.py
• <support-root>/assets/memory-config.yaml
• <support-root>/assets/chain-example.py

在使用依赖这些路径的步骤之前，请索取原始支持材料或现有安装；此目录未提供经验证的下载来源。请根据该安装解析每条路径，检查其内容，并核实适用的许可证和执行前置条件。所需文件不存在时，请勿运行相关命令，也不要声称已执行所引用的流程。请勿伪造缺失的脚本，也不要将名称相同但无关的文件视为等效文件。此处完整记载的独立步骤可在用户授权范围内继续执行；请将任何依赖步骤报告为不可用。

LangChain 架构

掌握 LangChain 框架，使用 Agent、链、记忆和工具集成构建复杂的大模型应用。

以下情况请勿使用此技能

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

说明

• 明确目标、约束和所需输入。
• 应用相关最佳实践并验证结果。
• 提供可执行的步骤和验证方法。

以下情况请使用此技能

• 构建可调用工具的自主 AI Agent
• 实现复杂的多步骤大模型工作流
• 管理对话记忆和状态
• 将大模型与外部数据源和 API 集成
• 创建模块化、可复用的大模型应用组件
• 实现文档处理管道
• 构建可用于生产环境的大模型应用

核心概念

1. Agent

使用大模型决定采取哪些行动的自主系统。

Agent 类型：

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

2. 链

对大模型或其他工具的调用序列。

链类型：

• LLMChain：基本的提示词 + 大模型组合
• SequentialChain：按顺序执行多个链
• RouterChain：将输入路由到专用链
• TransformChain：在步骤之间进行数据转换
• MapReduceChain：并行处理并汇总结果

3. 记忆

用于在交互过程中保留上下文的系统。

记忆类型：

• ConversationBufferMemory：存储所有消息
• ConversationSummaryMemory：总结较早的消息
• ConversationBufferWindowMemory：保留最近的 N 条消息
• EntityMemory：跟踪实体相关信息
• VectorStoreMemory：检索语义相似内容

4. 文档处理

加载、转换并存储文档，以供检索。

组件：

• Document Loaders：从各种来源加载内容
• Text Splitters：智能拆分文档
• Vector Stores：存储并检索嵌入
• Retrievers：获取相关文档
• Indexes：整理文档，以便高效访问

5. 回调

用于日志记录、监控和调试的钩子。

用例：

• 请求/响应日志记录
• Token 用量跟踪
• 延迟监控
• 错误处理
• 自定义指标收集

版本和执行范围

内联代码使用的是旧版 LangChain API／导入方式。LangChain v1 将旧版链及相关功能移到了 langchain-classic；不要直接安装最新版本，并假设这些代码片段无需修改即可运行。编写新代码时，请使用官方 v1 迁移指南以及受支持的 Agent／状态 API：https://docs.langchain.com/oss/python/migrate/langchain-v1。固定兼容的提供方软件包版本，并验证其导入方式。下面的工具只是占位符，并非真实的数据库／电子邮件集成；未经验证且获授权的副作用，绝不要报告邮件已发送。限制迭代次数、令牌用量和付费工具调用；隔离不同租户的记忆，并将检索到的文本视为不可信内容。日志必须对提示词、结果和凭据进行脱敏。

快速开始

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

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：使用工具的自定义 Agent

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
    raise NotImplementedError("Configure an authorized email adapter and verify delivery before reporting success")

tools = [search_database, send_email]

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

模式 3：多步链

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
)

记忆管理最佳实践

选择合适的记忆类型

# 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)

回调系统

自定义回调处理程序

from langchain.callbacks.base import BaseCallbackHandler

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

    def on_llm_end(self, response, **kwargs):
        print("LLM ended")

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

    def on_chain_start(self, serialized, inputs, **kwargs):
        print("Chain started")

    def on_agent_action(self, action, **kwargs):
        print("Agent action selected")

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

测试策略

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 the actual selected tool invocation with a compatible fake model/tool spy;
    # call arguments to the model only show the prompt, not tool execution.

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. 缓存

from langchain.cache import InMemoryCache
import langchain

langchain.llm_cache = InMemoryCache()

2. 批处理

# 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. 流式响应

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

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

资源

• references/agents.md：深入了解 Agent 架构
• references/memory.md：记忆系统模式
• references/chains.md：链组合策略
• references/document-processing.md：文档加载和索引
• references/callbacks.md：监控和可观测性
• assets/agent-template.py：可用于生产环境的 Agent 模板
• assets/memory-config.yaml：记忆配置示例
• assets/chain-example.py：复杂链示例

常见问题

1. 记忆溢出：未管理对话历史长度
2. 工具选择错误：工具描述不佳，导致 Agent 混淆
3. 超出上下文窗口：超出大模型的令牌限制
4. 未处理错误：未捕获和处理 Agent 故障
5. 检索效率低：未优化向量存储查询

生产环境检查清单

• 实现适当的错误处理
• 添加请求／响应日志记录
• 监控令牌用量和成本
• 为 Agent 执行设置超时限制
• 实现速率限制
• 添加输入验证
• 使用边界情况进行测试
• 设置可观测性（回调）
• 实现回退策略
• 对提示词和配置进行版本控制

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 请勿将输出视为针对特定环境的验证、测试或专家审查的替代品。
• 如果缺少必要的输入、权限、安全边界或成功标准，请停止并要求澄清。