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 的典型依赖如下包名版本职责langchain1.2.x高层编排chains、agentslangchain-core1.2.x核心抽象messages、prompts、toolslangchain-community—第三方集成langgraph—Agent 编排与状态管理langchain-openai—OpenAI 集成langchain-anthropic—Anthropic / Claude 集成langchain-voyageai—Voyage AI embeddingslangchain-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 检索增强生成RAGRetrieval-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(modelclaude-sonnet-5) embeddings VoyageAIEmbeddings(modelvoyage-3-large) vectorstore PineconeVectorStore(index_namedocs, embeddingembeddings) 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(contextcontext_text, questionstate[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_typehybrid, 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」。三、架构模式二带结构化工具的自定义 AgentAgent 的核心能力是自主调用工具。为让 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(descriptionSearch query) filters: dict Field(default{}, descriptionOptional filters) class EmailInput(BaseModel): Input for sending email. recipient: str Field(descriptionEmail recipient) subject: str Field(descriptionEmail subject) content: str Field(descriptionEmail body) async def search_database(query: str, filters: dict {}) - str: Search internal database for information. # Your database search logic return fResults 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 fEmail sent to {recipient} tools [ StructuredTool.from_function( coroutinesearch_database, namesearch_database, descriptionSearch internal database, args_schemaSearchInput ), StructuredTool.from_function( coroutinesend_email, namesend_email, descriptionSend an email, args_schemaEmailInput ) ] agent create_react_agent(llm, tools)要点说明create_react_agent来自langgraph.prebuilt见下文的模式四以及 SKILL.md 的 Quick Start 中对from langgraph.prebuilt import create_react_agent的导入示例它封装了 ReActReasoning Acting循环思考 → 选工具 → 执行 → 观察 → 再思考。Pydantic schema 的价值args_schemaSearchInput让 LLM 能按query/filters字段生成合法的 JSON 入参Field(description...)中的描述会被注入模型提示词直接影响工具选择的准确性。除了StructuredTool.from_functionSKILL.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 fExtract 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 fAnalyze 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 fSummarize: 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 fBased 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长对话摘要压缩、ConversationTokenBufferMemoryToken 窗口、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, checkpointercheckpointer) # 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, Whats my name?)]}, config) # Agent remembers: Your name is Alice关键概念——thread_idconfig {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:passlocalhost/langgraph ) agent create_react_agent(llm, tools, checkpointercheckpointer)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(modelvoyage-3-large) memory_store Chroma( collection_nameconversation_memory, embedding_functionembeddings, 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, kk) 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_searchk控制返回条数职责实际项目中可将召回结果与当前问题一起拼入 prompt实现「VectorStoreRetrieverMemory」式的语义记忆。综合来看记忆选型建议为短会话用 Token 窗口、长会话用摘要压缩、跨会话用 Checkpointer 向量记忆的组合见 langchain-agent.md 的 Memory Systems 一节。七、可观测性LangSmith 追踪与自定义 CallbackLLM 应用调试困难可观测性必须从第一天就纳入架构。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(modelclaude-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(fLLM started with {len(prompts)} prompts) def on_llm_end(self, response, **kwargs) - None: print(fLLM completed: {len(response.generations)} generations) def on_llm_error(self, error: Exception, **kwargs) - None: print(fLLM error: {error}) def on_tool_start( self, serialized: Dict[str, Any], input_str: str, **kwargs ) - None: print(fTool started: {serialized.get(name)}) def on_tool_end(self, output: str, **kwargs) - None: print(fTool 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(modelclaude-sonnet-5, streamingTrue) # Stream tokens async for chunk in llm.astream(Tell me a story): print(chunk.content, end, flushTrue) # Stream agent events async for event in agent.astream_events( {messages: [(user, Search and summarize)]}, versionv2 ): 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_eventsversionv2事件流包含on_chat_model_stream模型逐 token、on_tool_start工具开始执行等事件前端可借此实现「正在调用工具」的状态提示极大提升 Agent 交互的可理解性。在服务端集成时可按 langchain-agent.md 的 FastAPI 示例用StreamingResponse(..., media_typetext/event-stream)把上述事件流暴露为 SSE 接口非流式场景则直接await agent.ainvoke(...)。九、测试策略与性能优化9.1 测试用 Mock 隔离 LLM、验证工具选择与记忆持久化SKILL.md 给出了一套轻量级测试思路用pytestunittest.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(contentUsing 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的evaluateRunEvalConfig(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_keyos.environ[PINECONE_API_KEY]) index pc.Index(my-index) # Create vector store with existing index vectorstore PineconeVectorStore(indexindex, embeddingembeddings)set_llm_cache(RedisCache(...))开启全局 LLM 响应缓存可显著降低重复提问的成本与延迟。asyncio.gather并行处理文档分块与向量化适合离线索引构建。复用 Pinecone 客户端连接避免每次重建是「Connection Pooling: Reuse vector DB connections」的落地做法。此外langchain-agent.md 还建议用tenacity的retry(stopstop_after_attempt(3), waitwait_exponential(...))实现指数退避重试、为所有异步操作设置超时、以多 worker 轮询实现负载均衡。十、生产落地清单综合langchain-architecture技能文档与 langchain-agent.md 的实现清单生产级 LangChain 应用需要覆盖初始化 LLM如claude-sonnet-5与 Voyage AI embeddingsvoyage-3-large为所有工具提供异步支持与错误处理try/except fallback并用 Pydantic schema 声明入参按场景选择记忆系统Token 窗口 / 摘要 / 向量记忆 / Checkpointer 组合用 LangGraph StateGraph 构建工作流必要时compile(checkpointer...)开启 LangSmith 追踪配置结构化日志与健康检查实现流式响应SSE与 Redis 缓存层配置重试逻辑与超时编写单元测试、集成测试与离线评测集文档化 API 端点与架构用 Checkpointer 保证状态可复现。十一、源码导航本文主体详细模式文档四大架构模式、记忆管理、Callback、流式响应技能总纲SKILL.md包结构、核心概念、ReAct 快速开始、测试与性能优化插件总览与版本演进README.mdLangChain 0.x → 1.x 迁移记录、环境要求命令级实践指南langchain-agent.mdAgent 类型选型、生产部署、评测与实现清单相邻技能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 应用。【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考