⚡

LangChain架构Skill langchain-architecture

LangChain框架用于开发和设计基于大语言模型的应用程序，集成代理、链、记忆和工具，适用于AI智能体构建、多步工作流管理、文档处理、RAG应用等场景，提升LLM应用开发效率，关键词包括LangChain、AI代理、大模型应用、RAG、LLM工作流、记忆管理。

AI应用 0 次安装 2 次浏览 更新于 3/16/2026

name: langchain-architecture description: 使用LangChain框架设计LLM应用程序，包括代理、记忆和工具集成模式。适用于构建LangChain应用、实现AI代理或创建复杂的LLM工作流。

LangChain架构

掌握LangChain框架，用于构建具有代理、链、记忆和工具集成的复杂LLM应用程序。

何时使用此技能

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

核心概念

1. 代理

使用LLM决定采取哪些行动的自系统。

代理类型:

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

2. 链

对LLM或其他工具的一系列调用序列。

链类型:

  • LLMChain: 基本提示 + LLM组合
  • SequentialChain: 多个链按顺序执行
  • RouterChain: 将输入路由到专门化链
  • TransformChain: 步骤之间的数据转换
  • MapReduceChain: 并行处理与聚合

3. 记忆

跨交互维护上下文的系统。

记忆类型:

  • ConversationBufferMemory: 存储所有消息
  • ConversationSummaryMemory: 总结较旧消息
  • ConversationBufferWindowMemory: 保持最后N条消息
  • EntityMemory: 跟踪实体的信息
  • VectorStoreMemory: 语义相似性检索

4. 文档处理

加载、转换和存储文档以便检索。

组件:

  • Document Loaders: 从各种来源加载
  • Text Splitters: 智能切分文档
  • Vector Stores: 存储和检索嵌入
  • Retrievers: 获取相关文档
  • Indexes: 组织文档以提高访问效率

5. 回调

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

用例:

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

快速开始

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

# 初始化LLM
llm = OpenAI(temperature=0)

# 加载工具
tools = load_tools(["serpapi", "llm-math"], llm=llm)

# 添加记忆
memory = ConversationBufferMemory(memory_key="chat_history")

# 创建代理
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    verbose=True
)

# 运行代理
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

# 加载和处理文档
loader = TextLoader('documents.txt')
documents = loader.load()

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

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(texts, embeddings)

# 创建检索链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(),
    return_source_documents=True
)

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

模式2: 带有工具的自定义代理

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:
    """搜索内部数据库获取信息。"""
    # 您的数据库搜索逻辑
    return f"Results for: {query}"

@tool
def send_email(recipient: str, content: str) -> str:
    """向指定收件人发送电子邮件。"""
    # 电子邮件发送逻辑
    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: 多步链

from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import PromptTemplate

# 步骤1: 提取关键信息
extract_prompt = PromptTemplate(
    input_variables=["text"],
    template="从以下文本中提取关键实体: {text}

实体:"
)
extract_chain = LLMChain(llm=llm, prompt=extract_prompt, output_key="entities")

# 步骤2: 分析实体
analyze_prompt = PromptTemplate(
    input_variables=["entities"],
    template="分析这些实体: {entities}

分析:"
)
analyze_chain = LLMChain(llm=llm, prompt=analyze_prompt, output_key="analysis")

# 步骤3: 生成摘要
summary_prompt = PromptTemplate(
    input_variables=["entities", "analysis"],
    template="摘要:
实体: {entities}
分析: {analysis}

摘要:"
)
summary_chain = LLMChain(llm=llm, prompt=summary_prompt, output_key="summary")

# 组合成顺序链
overall_chain = SequentialChain(
    chains=[extract_chain, analyze_chain, summary_chain],
    input_variables=["text"],
    output_variables=["entities", "analysis", "summary"],
    verbose=True
)

记忆管理最佳实践

选择合适的记忆类型

# 对于短对话 (< 10条消息)
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()

# 对于长对话 (总结旧消息)
from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(llm=llm)

# 对于滑动窗口 (最后N条消息)
from langchain.memory import ConversationBufferWindowMemory
memory = ConversationBufferWindowMemory(k=5)

# 对于实体跟踪
from langchain.memory import ConversationEntityMemory
memory = ConversationEntityMemory(llm=llm)

# 对于相关历史的语义检索
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(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}")

# 使用回调
agent.run("query", callbacks=[CustomCallbackHandler()])

测试策略

import pytest
from unittest.mock import Mock

def test_agent_tool_selection():
    # 模拟LLM以返回特定工具选择
    mock_llm = Mock()
    mock_llm.predict.return_value = "Action: search_database
Action Input: test query"

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

    result = agent.run("test query")

    # 验证选择了正确工具
    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. 缓存

from langchain.cache import InMemoryCache
import langchain

langchain.llm_cache = InMemoryCache()

2. 批量处理

# 并行处理多个文档
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: 代理架构深度解析
  • 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令牌限制
  4. 无错误处理: 未捕获和处理代理失败
  5. 低效检索: 未优化向量存储查询

生产检查清单

  • [ ] 实现适当的错误处理
  • [ ] 添加请求/响应日志记录
  • [ ] 监控令牌使用和成本
  • [ ] 设置代理执行超时限制
  • [ ] 实现速率限制
  • [ ] 添加输入验证
  • [ ] 测试边缘案例
  • [ ] 设置可观测性 (回调)
  • [ ] 实现回退策略
  • [ ] 版本控制提示和配置