LangChain 1.x + LangGraph 架构实战：RAG、结构化工具、多智能体编排与记忆系统的完整模式库-平芜编程栈

LangChain 1.x + LangGraph 架构实战：RAG、结构化工具、多智能体编排与记忆系统的完整模式库

【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents

本篇技术指南基于 agents24/agents 仓库中llm-application-dev插件内langchain-architecture技能（详细模式文档与 SKILL.md）编写，系统梳理在 LangChain 1.x 与 LangGraph 体系下构建 LLM 应用的四大核心架构模式（RAG、结构化工具 Agent、多步工作流、多智能体编排），并深入讲解记忆管理、LangSmith 可观测性、流式响应与生产级优化方案。读完本文，你将掌握一套可直接复制运行的代码模板，以及从本地开发（内存检查点）到生产部署（PostgreSQL 检查点、Redis 缓存、LangSmith 追踪）的完整技术路径。

一、背景：LangChain 1.x 与 LangGraph 的定位

langchain-architecture技能面向 LangChain 1.x 生态设计。从插件 README.md 的版本记录（2.0.0, January 2026）可以看到一次重要的技术迁移：项目从 LangChain 0.x 全面迁移到 LangChain 1.x / LangGraph，废弃了旧的initialize_agent()式 API，改用LangGraph StateGraph工作流，并引入 Voyage AI 作为 Claude 应用的推荐 embedding 方案、以 Pydantic 实现结构化输出、以 checkpointer 实现异步持久化执行。

根据 SKILL.md 中的包结构说明，LangChain 1.x 的典型依赖如下：

包名	版本	职责
`langchain`	1.2.x	高层编排（chains、agents）
`langchain-core`	1.2.x	核心抽象（messages、prompts、tools）
`langchain-community`	—	第三方集成
`langgraph`	—	Agent 编排与状态管理
`langchain-openai`	—	OpenAI 集成
`langchain-anthropic`	—	Anthropic / Claude 集成
`langchain-voyageai`	—	Voyage AI embeddings
`langchain-pinecone`	—	Pinecone 向量库

对应插件环境要求为：LangChain >= 1.2.0、LangGraph >= 0.3.0、Python 3.11+（见 README.md 的 Requirements 一节）。文档中统一以claude-sonnet-5作为 LLM 示例、voyage-3-large作为 embedding 示例，这是仓库内各文档（SKILL.md、ai-engineer.md）反复出现的推荐组合，实际使用时请替换为你所在环境可用的模型与密钥。

二、架构模式一：基于 LangGraph 的 RAG 检索增强生成

RAG（Retrieval-Augmented Generation）是生产环境中最常用的 LLM 应用形态。以下模式将「检索」与「生成」两个步骤建模为 LangGraph 图中的两个节点，利用显式状态（TypedDict）在节点间传递数据。

from langgraph.graph import StateGraph, START, END from langchain_anthropic import ChatAnthropic from langchain_voyageai import VoyageAIEmbeddings from langchain_pinecone import PineconeVectorStore from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from typing import TypedDict, Annotated class RAGState(TypedDict): question: str context: Annotated[list[Document], "retrieved documents"] answer: str # Initialize components llm = ChatAnthropic(model="claude-sonnet-5") embeddings = VoyageAIEmbeddings(model="voyage-3-large") vectorstore = PineconeVectorStore(index_name="docs", embedding=embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 4}) # Define nodes async def retrieve(state: RAGState) -> RAGState: """Retrieve relevant documents.""" docs = await retriever.ainvoke(state["question"]) return {"context": docs} async def generate(state: RAGState) -> RAGState: """Generate answer from context.""" prompt = ChatPromptTemplate.from_template( """Answer based on the context below. If you cannot answer, say so. Context: {context} Question: {question} Answer:""" ) context_text = "\n\n".join(doc.page_content for doc in state["context"]) response = await llm.ainvoke( prompt.format(context=context_text, question=state["question"]) ) return {"answer": response.content} # Build graph builder = StateGraph(RAGState) builder.add_node("retrieve", retrieve) builder.add_node("generate", generate) builder.add_edge(START, "retrieve") builder.add_edge("retrieve", "generate") builder.add_edge("generate", END) rag_chain = builder.compile() # Use the chain result = await rag_chain.ainvoke({"question": "What is the main topic?"})

关键设计点解析：

显式状态模型：RAGState用TypedDict声明question、context、answer三个字段，其中context通过Annotated附加说明性元数据。这正是 LangGraph 与传统 Chain 的本质区别——状态是类型化、显式、可审查的（对应 SKILL.md 中「StateGraph: Explicit state management with typed state」的特性）。
节点即纯函数：每个节点是一个async函数，输入完整状态、输出增量字段，LangGraph 负责合并状态并按图结构调度。
检索器配置：vectorstore.as_retriever(search_kwargs={"k": 4})设置召回 top-4 文档；如果使用混合检索，可按 langchain-agent.md 中的示例配置search_type="hybrid", search_kwargs={"k": 20, "alpha": 0.5}进行向量 + 关键词融合召回，再用 Cohere Rerank 等模型重排。
异步贯穿：ainvoke、astream、aget_relevant_documents等异步 API 是 LangChain 1.x 的标准用法，也是生产吞吐的基础（见 langchain-agent.md 的 Best Practices：「Always use async」）。

三、架构模式二：带结构化工具的自定义 Agent

Agent 的核心能力是自主调用工具。为让 LLM 准确生成工具入参，LangChain 1.x 推荐用 Pydantic 模型声明工具的参数 schema，再通过StructuredTool.from_function包装：

from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field class SearchInput(BaseModel): """Input for database search.""" query: str = Field(description="Search query") filters: dict = Field(default={}, description="Optional filters") class EmailInput(BaseModel): """Input for sending email.""" recipient: str = Field(description="Email recipient") subject: str = Field(description="Email subject") content: str = Field(description="Email body") async def search_database(query: str, filters: dict = {}) -> str: """Search internal database for information.""" # Your database search logic return f"Results for '{query}' with filters {filters}" async def send_email(recipient: str, subject: str, content: str) -> str: """Send an email to specified recipient.""" # Email sending logic return f"Email sent to {recipient}" tools = [ StructuredTool.from_function( coroutine=search_database, name="search_database", description="Search internal database", args_schema=SearchInput ), StructuredTool.from_function( coroutine=send_email, name="send_email", description="Send an email", args_schema=EmailInput ) ] agent = create_react_agent(llm, tools)

要点说明：

create_react_agent来自langgraph.prebuilt（见下文的模式四，以及 SKILL.md 的 Quick Start 中对from langgraph.prebuilt import create_react_agent的导入示例），它封装了 ReAct（Reasoning + Acting）循环：思考 → 选工具 → 执行 → 观察 → 再思考。
Pydantic schema 的价值：args_schema=SearchInput让 LLM 能按query/filters字段生成合法的 JSON 入参，Field(description=...)中的描述会被注入模型提示词，直接影响工具选择的准确性。
除了StructuredTool.from_function，SKILL.md 还展示了更简洁的@tool装饰器写法，并给出了一个值得借鉴的安全实现——用 Pythonast模块做安全的数学表达式求值（白名单操作符映射ast.Add → operator.add等），替代不安全的eval。这对应插件 README 中「Fixed security issue: replaced unsafe code execution with AST-based safe math evaluation」的修复记录。

四、架构模式三：基于 StateGraph 的多步工作流

当任务需要「抽取实体 → 分析实体 → 生成总结」这类固定流水线时，可以用add_conditional_edges按状态字段动态路由。该模式展示了 LangGraph 的条件分支能力：

from langgraph.graph import StateGraph, START, END from typing import TypedDict, Literal class WorkflowState(TypedDict): text: str entities: list analysis: str summary: str current_step: str async def extract_entities(state: WorkflowState) -> WorkflowState: """Extract key entities from text.""" prompt = f"Extract key entities from: {state['text']}\n\nReturn as JSON list." response = await llm.ainvoke(prompt) return {"entities": response.content, "current_step": "analyze"} async def analyze_entities(state: WorkflowState) -> WorkflowState: """Analyze extracted entities.""" prompt = f"Analyze these entities: {state['entities']}\n\nProvide insights." response = await llm.ainvoke(prompt) return {"analysis": response.content, "current_step": "summarize"} async def generate_summary(state: WorkflowState) -> WorkflowState: """Generate final summary.""" prompt = f"""Summarize: Entities: {state['entities']} Analysis: {state['analysis']} Provide a concise summary.""" response = await llm.ainvoke(prompt) return {"summary": response.content, "current_step": "complete"} def route_step(state: WorkflowState) -> Literal["analyze", "summarize", "end"]: """Route to next step based on current state.""" step = state.get("current_step", "extract") if step == "analyze": return "analyze" elif step == "summarize": return "summarize" return "end" # Build workflow builder = StateGraph(WorkflowState) builder.add_node("extract", extract_entities) builder.add_node("analyze", analyze_entities) builder.add_node("summarize", generate_summary) builder.add_edge(START, "extract") builder.add_conditional_edges("extract", route_step, { "analyze": "analyze", "summarize": "summarize", "end": END }) builder.add_conditional_edges("analyze", route_step, { "summarize": "summarize", "end": END }) builder.add_edge("summarize", END) workflow = builder.compile()

设计要点：

路由函数：route_step读取状态中的current_step字段，返回目标节点名；返回END（通过映射"end": END）即可终止流程。
条件边映射表：add_conditional_edges("extract", route_step, {...})的第三参数是把路由返回值映射到具体节点（或END）的字典。该语法与 langchain-agent.md 中的通用模式builder.add_conditional_edges("node1", router, {"a": "node2", "b": END})完全一致。
可观测的状态机：整个流水线的每一步都显式记录在current_step中，便于调试、断点续跑与人工介入（对应 SKILL 中「Human-in-the-Loop: Inspect and modify state at any point」的能力）。

五、架构模式四：多智能体编排（Supervisor 路由）

将多个职责单一的 Agent（研究员 / 写作者 / 评审员）交给一个 Supervisor 节点统一调度，是生产级多智能体系统的经典形态。Supervisor 每次根据对话内容决定下一个执行者，所有 Agent 执行完毕后回到 Supervisor，直到它判定任务完成：

from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import create_react_agent from langchain_core.messages import HumanMessage from typing import Literal class MultiAgentState(TypedDict): messages: list next_agent: str # Create specialized agents researcher = create_react_agent(llm, research_tools) writer = create_react_agent(llm, writing_tools) reviewer = create_react_agent(llm, review_tools) async def supervisor(state: MultiAgentState) -> MultiAgentState: """Route to appropriate agent based on task.""" prompt = f"""Based on the conversation, which agent should handle this? Options: - researcher: For finding information - writer: For creating content - reviewer: For reviewing and editing - FINISH: Task is complete Messages: {state['messages']} Respond with just the agent name.""" response = await llm.ainvoke(prompt) return {"next_agent": response.content.strip().lower()} def route_to_agent(state: MultiAgentState) -> Literal["researcher", "writer", "reviewer", "end"]: """Route based on supervisor decision.""" next_agent = state.get("next_agent", "").lower() if next_agent == "finish": return "end" return next_agent if next_agent in ["researcher", "writer", "reviewer"] else "end" # Build multi-agent graph builder = StateGraph(MultiAgentState) builder.add_node("supervisor", supervisor) builder.add_node("researcher", researcher) builder.add_node("writer", writer) builder.add_node("reviewer", reviewer) builder.add_edge(START, "supervisor") builder.add_conditional_edges("supervisor", route_to_agent, { "researcher": "researcher", "writer": "writer", "reviewer": "reviewer", "end": END }) # Each agent returns to supervisor for agent in ["researcher", "writer", "reviewer"]: builder.add_edge(agent, "supervisor") multi_agent = builder.compile()

核心机制：

预构建 Agent 作为节点：create_react_agent(llm, tools)创建的 ReAct Agent 可以直接作为 StateGraph 的节点，无需再包一层函数。
循环路由：for agent in [...]: builder.add_edge(agent, "supervisor")让每个专职 Agent 完成后都回到 Supervisor，从而形成「调度 → 执行 → 再调度」的循环；只有当 Supervisor 输出finish时，route_to_agent才返回"end"终止图。这正是「Multi-Agent: Supervisor routing between specialized agents」模式的落地实现（见 SKILL.md 的 Agent Patterns 一节）。
注意：本模式中supervisor使用llm.ainvoke(prompt)完成纯文本路由决策；在更复杂的场景中，也可以像 langchain-agent.md 提到的那样使用Command[Literal["agent1", "agent2", END]]机制进行强类型路由。

六、记忆管理：从内存检查点到生产级持久化

LangGraph 的记忆体系是它区别于普通 Chain 的核心能力之一。SKILL.md 归纳了多级记忆方案：ConversationBufferMemory（短对话全量消息）、ConversationSummaryMemory（长对话摘要压缩）、ConversationTokenBufferMemory（Token 窗口）、VectorStoreRetrieverMemory（语义相似检索）、以及 LangGraph Checkpointers（跨会话持久状态）。细节文档则重点给出了 Checkpointer 与向量记忆的三种实操写法。

6.1 基于 MemorySaver 的 Token 级会话记忆（开发环境）

from langgraph.checkpoint.memory import MemorySaver from langgraph.prebuilt import create_react_agent # In-memory checkpointer (development) checkpointer = MemorySaver() # Create agent with persistent memory agent = create_react_agent(llm, tools, checkpointer=checkpointer) # Each thread_id maintains separate conversation config = {"configurable": {"thread_id": "session-abc123"}} # Messages persist across invocations with same thread_id result1 = await agent.ainvoke({"messages": [("user", "My name is Alice")]}, config) result2 = await agent.ainvoke({"messages": [("user", "What's my name?")]}, config) # Agent remembers: "Your name is Alice"

关键概念——thread_id：config = {"configurable": {"thread_id": "session-abc123"}}是对话隔离的钥匙。同一个thread_id下的多次调用共享同一份检查点状态（消息会持久累积），不同thread_id则完全隔离。开发阶段使用进程内MemorySaver即可验证记忆行为。

6.2 基于 PostgreSQL 的生产级检查点

from langgraph.checkpoint.postgres import PostgresSaver # Production checkpointer checkpointer = PostgresSaver.from_conn_string( "postgresql://user:pass@localhost/langgraph" ) agent = create_react_agent(llm, tools, checkpointer=checkpointer)

PostgresSaver将图状态落盘到 PostgreSQL，进程重启、多副本部署都不丢失会话状态。这与「Durable Execution: Agents persist through failures」和「Checkpointing: Save and resume agent state」两大特性直接对应，是生产环境的标准选择。

6.3 基于向量库的长期记忆

对话历史超过上下文窗口后，可以只检索「最相关的历史片段」注入提示词。以下写法用 Chroma 存储历史消息的 embedding，按语义相似度召回：

from langchain_community.vectorstores import Chroma from langchain_voyageai import VoyageAIEmbeddings embeddings = VoyageAIEmbeddings(model="voyage-3-large") memory_store = Chroma( collection_name="conversation_memory", embedding_function=embeddings, persist_directory="./memory_db" ) async def retrieve_relevant_memory(query: str, k: int = 5) -> list: """Retrieve relevant past conversations.""" docs = await memory_store.asimilarity_search(query, k=k) return [doc.page_content for doc in docs] async def store_memory(content: str, metadata: dict = {}): """Store conversation in long-term memory.""" await memory_store.aadd_texts([content], metadatas=[metadata])

两个辅助函数分别承担「写入」（aadd_texts，可携带metadata）与「召回」（asimilarity_search，k控制返回条数）职责，实际项目中可将召回结果与当前问题一起拼入 prompt，实现「VectorStoreRetrieverMemory」式的语义记忆。综合来看，记忆选型建议为：短会话用 Token 窗口、长会话用摘要压缩、跨会话用 Checkpointer + 向量记忆的组合（见 langchain-agent.md 的 Memory Systems 一节）。

七、可观测性：LangSmith 追踪与自定义 Callback

LLM 应用调试困难，可观测性必须从第一天就纳入架构。LangSmith 是 LangChain 官方生态的观测方案（SKILL.md 总结其能力：请求/响应日志、Token 用量统计、延迟监控、错误追踪、Trace 可视化）。

7.1 开启 LangSmith 全链路追踪

import os from langchain_anthropic import ChatAnthropic # Enable LangSmith tracing os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "your-api-key" os.environ["LANGCHAIN_PROJECT"] = "my-project" # All LangChain/LangGraph operations are automatically traced llm = ChatAnthropic(model="claude-sonnet-5")

设置这三个环境变量后，后续所有 LangChain / LangGraph 操作（LLM 调用、工具执行、图节点调度）都会被自动追踪上报，无需侵入式改造业务代码。

7.2 自定义 Callback Handler

当内置追踪不足以覆盖自定义逻辑（如旁路日志、指标上报）时，继承BaseCallbackHandler即可挂钩 LLM 与工具的生命周期事件：

from langchain_core.callbacks import BaseCallbackHandler from typing import Any, Dict, List class CustomCallbackHandler(BaseCallbackHandler): def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], **kwargs ) -> None: print(f"LLM started with {len(prompts)} prompts") def on_llm_end(self, response, **kwargs) -> None: print(f"LLM completed: {len(response.generations)} generations") def on_llm_error(self, error: Exception, **kwargs) -> None: print(f"LLM error: {error}") def on_tool_start( self, serialized: Dict[str, Any], input_str: str, **kwargs ) -> None: print(f"Tool started: {serialized.get('name')}") def on_tool_end(self, output: str, **kwargs) -> None: print(f"Tool completed: {output[:100]}...") # Use callbacks result = await agent.ainvoke( {"messages": [("user", "query")]}, config={"callbacks": [CustomCallbackHandler()]} )

该 Handler 覆盖了 LLM 的start / end / error与工具的start / end五类事件，通过config={"callbacks": [...]}注入到单次调用中。除 Callback 外，langchain-agent.md 还建议生产环境补充 Prometheus 指标（请求量、延迟、错误数）、structlog结构化日志以及面向 LLM / 工具 / 记忆 / 外部服务的健康检查。

八、流式响应：Token 级与事件级

对话式应用的用户体验高度依赖流式输出。LangChain 1.x 提供两层流式能力：

from langchain_anthropic import ChatAnthropic llm = ChatAnthropic(model="claude-sonnet-5", streaming=True) # Stream tokens async for chunk in llm.astream("Tell me a story"): print(chunk.content, end="", flush=True) # Stream agent events async for event in agent.astream_events( {"messages": [("user", "Search and summarize")]}, version="v2" ): if event["event"] == "on_chat_model_stream": print(event["data"]["chunk"].content, end="") elif event["event"] == "on_tool_start": print(f"\n[Using tool: {event['name']}]")

模型级astream：直接逐 token 消费 LLM 输出，适合纯生成场景。
图级astream_events（version="v2"）：事件流包含on_chat_model_stream（模型逐 token）、on_tool_start（工具开始执行）等事件，前端可借此实现「正在调用工具」的状态提示，极大提升 Agent 交互的可理解性。

在服务端集成时，可按 langchain-agent.md 的 FastAPI 示例，用StreamingResponse(..., media_type="text/event-stream")把上述事件流暴露为 SSE 接口；非流式场景则直接await agent.ainvoke(...)。

九、测试策略与性能优化

9.1 测试：用 Mock 隔离 LLM、验证工具选择与记忆持久化

SKILL.md 给出了一套轻量级测试思路：用pytest+unittest.mock替换 LLM 调用，验证 Agent 的工具选择；利用 checkpointer 的thread_id机制验证跨调用记忆：

import pytest from unittest.mock import AsyncMock, patch @pytest.mark.asyncio async def test_agent_tool_selection(): """Test agent selects correct tool.""" with patch.object(llm, 'ainvoke') as mock_llm: mock_llm.return_value = AsyncMock(content="Using search_database") result = await agent.ainvoke({ "messages": [("user", "search for documents")] }) # Verify tool was called assert "search_database" in str(result) @pytest.mark.asyncio async def test_memory_persistence(): """Test memory persists across invocations.""" config = {"configurable": {"thread_id": "test-thread"}} # First message await agent.ainvoke( {"messages": [("user", "Remember: the code is 12345")]}, config ) # Second message should remember result = await agent.ainvoke( {"messages": [("user", "What was the code?")]}, config ) assert "12345" in result["messages"][-1].content

更进一步，可用langsmith.evaluation的evaluate+RunEvalConfig(evaluators=["qa", "context_qa", "cot_qa"])构建离线评测集，持续回归 Agent 质量（见 langchain-agent.md 的 Testing & Evaluation 一节）。

9.2 性能：Redis 缓存、异步批量与连接复用

# 1. Caching with Redis from langchain_community.cache import RedisCache from langchain_core.globals import set_llm_cache import redis redis_client = redis.Redis.from_url("redis://localhost:6379") set_llm_cache(RedisCache(redis_client))

# 2. Async Batch Processing import asyncio from langchain_core.documents import Document async def process_documents(documents: list[Document]) -> list: """Process documents in parallel.""" tasks = [process_single(doc) for doc in documents] return await asyncio.gather(*tasks) async def process_single(doc: Document) -> dict: """Process a single document.""" chunks = text_splitter.split_documents([doc]) embeddings = await embeddings_model.aembed_documents( [c.page_content for c in chunks] ) return {"doc_id": doc.metadata.get("id"), "embeddings": embeddings}

# 3. Connection Pooling from langchain_pinecone import PineconeVectorStore from pinecone import Pinecone # Reuse Pinecone client pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) index = pc.Index("my-index") # Create vector store with existing index vectorstore = PineconeVectorStore(index=index, embedding=embeddings)

set_llm_cache(RedisCache(...))开启全局 LLM 响应缓存，可显著降低重复提问的成本与延迟。
asyncio.gather并行处理文档分块与向量化，适合离线索引构建。
复用 Pinecone 客户端连接避免每次重建，是「Connection Pooling: Reuse vector DB connections」的落地做法。

此外，langchain-agent.md 还建议：用tenacity的@retry(stop=stop_after_attempt(3), wait=wait_exponential(...))实现指数退避重试、为所有异步操作设置超时、以多 worker 轮询实现负载均衡。

十、生产落地清单

综合langchain-architecture技能文档与 langchain-agent.md 的实现清单，生产级 LangChain 应用需要覆盖：

初始化 LLM（如claude-sonnet-5）与 Voyage AI embeddings（voyage-3-large）；
为所有工具提供异步支持与错误处理（try/except + fallback），并用 Pydantic schema 声明入参；
按场景选择记忆系统（Token 窗口 / 摘要 / 向量记忆 / Checkpointer 组合）；
用 LangGraph StateGraph 构建工作流，必要时compile(checkpointer=...)；
开启 LangSmith 追踪，配置结构化日志与健康检查；
实现流式响应（SSE）与 Redis 缓存层；
配置重试逻辑与超时；
编写单元测试、集成测试与离线评测集；
文档化 API 端点与架构，用 Checkpointer 保证状态可复现。

十一、源码导航

本文主体：详细模式文档（四大架构模式、记忆管理、Callback、流式响应）
技能总纲：SKILL.md（包结构、核心概念、ReAct 快速开始、测试与性能优化）
插件总览与版本演进：README.md（LangChain 0.x → 1.x 迁移记录、环境要求）
命令级实践指南：langchain-agent.md（Agent 类型选型、生产部署、评测与实现清单）
相邻技能：RAG 的混合检索与重排在 rag-implementation 与 hybrid-search-implementation 中有更深入的展开。

需要说明的是，本文示例中的模型名（claude-sonnet-5、voyage-3-large）与版本约束（LangChain 1.2.x、LangGraph 0.3.x）均以仓库当前文档为准；实际开发时请以你所使用环境支持的模型与已安装的包版本为准，并注意details.md中部分模式（如模式二）引用了create_react_agent，其导入方式统一为from langgraph.prebuilt import create_react_agent（见模式四示例）。把上面的代码模板当作起点，结合状态路由、记忆与可观测性三层能力，即可搭建出结构清晰、可维护、可监控的生产级 LLM 应用。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考