LangChain 1.x 与 LangGraph 架构实战构建生产级 Agent、记忆与工具编排系统【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents导读本文以agents仓库中llm-application-dev插件的langchain-architecture技能为核心系统讲解基于 LangChain 1.x 与 LangGraph 设计 LLM 应用的整体架构覆盖 Agent 编排、状态管理、记忆系统、文档处理链路、可观测性以及测试与性能优化。读者完成后将掌握从零搭建带工具调用的 ReAct Agent、用 StateGraph 构建多步骤工作流与多 Agent 协作系统并为记忆持久化、流式输出与生产部署找到可直接落地的方案。本文所有代码与架构依据来自 SKILL.md 及其深度模式文档 details.md并辅以同插件 README.md 与相邻技能佐证。一、技术选型LangChain 1.x 包结构在进入架构之前先明确现代 LangChain 生态的包划分。LangChain 1.x 将能力拆分为多个独立包职责边界清晰包版本参考职责langchain1.2.x高层编排组合链、Agent、文档处理流水线langchain-core1.2.x核心抽象消息、提示词、工具、文档、回调langchain-community—第三方集成各类向量库、缓存等langgraph—Agent 编排与状态管理图执行引擎langchain-openai—OpenAI 模型接入langchain-anthropic—Anthropic / Claude 模型接入langchain-voyageai—Voyage AI 嵌入模型接入langchain-pinecone—Pinecone 向量库接入这一拆分意味着核心抽象消息、提示词、工具依赖langchain-coreAgent 编排依赖langgraph模型与向量库通过独立集成包接入。该插件的 README.md 明确要求环境为 LangChain 1.2.0、LangGraph 0.3.0、Python 3.11并提示这是从 LangChain 0.x 迁移而来的 2.0.0 版本已全面废弃initialize_agent()等旧 API。二、核心概念为什么用 LangGraph 构建 Agent插件将 LangGraph 定位为 2026 年构建 Agent 的标准方案其关键能力围绕以下五点StateGraph显式状态管理状态通过TypedDict定义类型安全Durable Execution持久执行Agent 在故障后可从检查点恢复而非从头重跑Human-in-the-Loop可在任意节点暂停人工检查并修改状态后再继续Memory跨会话的短期与长期记忆Checkpointing保存并恢复 Agent 运行状态。在此基础上SKILL.md 归纳了四种 Agent 编排模式ReAct推理 行动通过create_react_agent快速创建Agent 循环执行思考 → 调用工具 → 观察结果 → 再思考Plan-and-Execute将规划与执行分离为独立节点适合复杂任务的前置规划Multi-Agent多 Agent 协作由 Supervisor监督者在多个专精 Agent 之间路由Tool-Calling使用 Pydantic Schema 描述工具入参实现结构化工具调用。三、状态管理TypedDict 驱动的显式状态LangGraph 的核心设计是状态即数据流。所有节点函数接收当前状态、返回状态增量图引擎负责合并。SKILL.md 给出两种状态定义方式from typing import Annotated, TypedDict from langgraph.graph import MessagesState # 基于消息的状态对话型 Agent 的默认选择 class AgentState(MessagesState): Extends MessagesState with custom fields. context: Annotated[list, retrieved documents] # 复杂 Agent 的自定义状态 class CustomState(TypedDict): messages: Annotated[list, conversation history] context: Annotated[dict, retrieved context] current_step: str results: list其中Annotated的第二个字符串参数是状态字段的注释/归约提示用于描述该字段的语义对于累加型字段如消息列表LangGraph 会依据 reducer 决定是覆盖还是追加。在 details.md 的 RAG 示例中context: Annotated[list[Document], retrieved documents]直接承载检索结果并传递给生成节点体现了状态在节点间流转的典型用法。四、记忆系统从会话缓冲到生产级 Checkpointer记忆是 Agent 具备上下文连续性的基础。SKILL.md 给出的记忆选型如下记忆类型适用场景ConversationBufferMemory短对话保留全部消息ConversationSummaryMemory长对话对较早消息做摘要压缩ConversationTokenBufferMemory基于 Token 数的窗口截断VectorStoreRetrieverMemory基于语义相似度检索历史记忆LangGraph Checkpointers跨会话的持久化状态推荐生产使用开发环境内存 Checkpointerfrom langgraph.checkpoint.memory import MemorySaver # 开发环境使用内存检查点 checkpointer MemorySaver() agent create_react_agent(llm, tools, checkpointercheckpointer) # 每个 thread_id 维护独立的会话上下文 config {configurable: {thread_id: session-abc123}} # 相同 thread_id 的多次调用之间消息会自动持久化 result1 await agent.ainvoke({messages: [(user, My name is Alice)]}, config) result2 await agent.ainvoke({messages: [(user, Whats my name?)]}, config) # Agent 能回忆起Your name is Alice生产环境PostgreSQL Checkpointerfrom langgraph.checkpoint.postgres import PostgresSaver checkpointer PostgresSaver.from_conn_string( postgresql://user:passlocalhost/langgraph ) agent create_react_agent(llm, tools, checkpointercheckpointer)长期记忆向量库语义检索对于需要跨会话长期保留的业务事实details.md 展示了以 Chroma Voyage AI 实现语义记忆的方案写入时aadd_texts存入文本与元数据读取时asimilarity_search按语义召回相关历史片段。这与插件中vector-database-engineerAgent 的职责agents/vector-database-engineer.md相互呼应。五、快速上手基于 LangGraph 的现代 ReAct AgentSKILL.md 提供了完整的可运行示例核心流程为初始化模型 → 定义工具 → 创建 Checkpointer → 创建 Agent → 带thread_id调用。from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.memory import MemorySaver from langchain_anthropic import ChatAnthropic from langchain_core.tools import tool import ast import operator # 初始化 LLM推荐 Claude Sonnet 5 llm ChatAnthropic(modelclaude-sonnet-5) # 定义工具tool 装饰器 类型注解即生成 Pydantic Schema tool def search_database(query: str) - str: Search internal database for information. # 你的数据库检索逻辑 return fResults for: {query} tool def calculate(expression: str) - str: Safely evaluate a mathematical expression. Supports: , -, *, /, **, %, parentheses Example: (2 3) * 4 returns 20 # 基于 ast 的安全数学求值避免 eval 注入风险 allowed_operators { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.Mod: operator.mod, ast.USub: operator.neg, } def _eval(node): if isinstance(node, ast.Constant): return node.value elif isinstance(node, ast.BinOp): left _eval(node.left) right _eval(node.right) return allowed_operatorstype(node.op) elif isinstance(node, ast.UnaryOp): operand _eval(node.operand) return allowed_operatorstype(node.op) else: raise ValueError(fUnsupported operation: {type(node)}) try: tree ast.parse(expression, modeeval) return str(_eval(tree.body)) except Exception as e: return fError: {e} tools [search_database, calculate] # 创建 Checkpointer 实现记忆持久化 checkpointer MemorySaver() # 创建 ReAct Agent agent create_react_agent( llm, tools, checkpointercheckpointer ) # 通过 thread_id 关联会话记忆 config {configurable: {thread_id: user-123}} result await agent.ainvoke( {messages: [(user, Search for Python tutorials and calculate 25 * 4)]}, configconfig )值得注意的工程细节calculate工具使用AST抽象语法树解析 白名单运算符映射而不是eval()。这正是该插件 2.0.0 版本修复的安全问题——将不安全的代码执行替换为基于 AST 的安全数学求值见 README.md Changelog。这个模式值得在需要 LLM 驱动计算的场景中直接复用。六、深度模式详解四个可复用的 StateGraph 工作流details.md 提供了四个带完整代码的工作模式是 SKILL.md 导航层之上的第二层知识库。模式一RAG with LangGraph将检索与生成构建为两个节点串联成一条确定性链路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 # 初始化组件 llm ChatAnthropic(modelclaude-sonnet-5) embeddings VoyageAIEmbeddings(modelvoyage-3-large) vectorstore PineconeVectorStore(index_namedocs, embeddingembeddings) retriever vectorstore.as_retriever(search_kwargs{k: 4}) # 定义节点 async def retrieve(state: RAGState) - RAGState: 检索相关文档。 docs await retriever.ainvoke(state[question]) return {context: docs} async def generate(state: RAGState) - RAGState: 基于上下文生成回答。 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} # 构建图 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() # 使用链路 result await rag_chain.ainvoke({question: What is the main topic?})该模式与插件中独立的rag-implementation技能rag-implementation/SKILL.md结构一致后者还补充了混合检索dense sparse、HyDE、RAG-Fusion、重排reranking等进阶策略可配合阅读。模式二自定义 Agent 与结构化工具当工具入参较多时使用StructuredTool.from_function PydanticBaseModel定义参数 Schemafrom langchain_core.tools import StructuredTool from pydantic import BaseModel, Field class SearchInput(BaseModel): 数据库检索入参。 query: str Field(descriptionSearch query) filters: dict Field(default{}, descriptionOptional filters) class EmailInput(BaseModel): 邮件发送入参。 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. return fResults for {query} with filters {filters} async def send_email(recipient: str, subject: str, content: str) - str: Send an email to specified recipient. 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)要点Field(description...)的文本会作为工具参数说明注入 LLM 提示词直接影响模型生成参数 JSON 的准确性因此描述应当具体、无歧义。命令 langchain-agent.md 中同样强调工具函数要内建 try/except 错误处理与 fallback。模式三带条件路由的多步骤工作流通过add_conditional_edges与路由函数实现分支跳转让图结构随状态演化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: 从文本抽取关键实体。 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: 分析抽取的实体。 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: 生成最终摘要。 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]: 依据当前状态路由到下一步。 step state.get(current_step, extract) if step analyze: return analyze elif step summarize: return summarize return end 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返回的字符串必须与add_conditional_edges第三参数字典中的键完全一致未命中的路由落入end从而保证图始终存在合法出口。模式四Supervisor 路由的多 Agent 编排将多个create_react_agent创建的专精 Agent 挂到同一个图上由 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 # 创建专精 Agent 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: 根据任务路由到合适的 Agent。 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]: 基于 supervisor 决策路由。 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 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 }) # 每个 Agent 完成后回到 supervisor 重新路由 for agent in [researcher, writer, reviewer]: builder.add_edge(agent, supervisor) multi_agent builder.compile()命令 langchain-agent.md 建议在生产环境使用Command[Literal[agent1, agent2, END]]这类显式类型标注的路由写法让返回类型与路由键在静态检查阶段即可对齐。七、可观测性LangSmith 追踪与自定义回调开启 LangSmith 追踪import os from langchain_anthropic import ChatAnthropic # 启用 LangSmith 追踪 os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_API_KEY] your-api-key os.environ[LANGCHAIN_PROJECT] my-project # 之后所有 LangChain/LangGraph 操作都会自动被追踪 llm ChatAnthropic(modelclaude-sonnet-5)LangSmith 提供的观测维度包括请求/响应日志、Token 用量追踪、延迟监控、错误追踪、Trace 可视化。这与 ai-engineer.md 中可观测性LangSmith、Phoenix、Weights Biases的能力清单一致。自定义回调 Handler当需要将事件接入自有监控体系时继承BaseCallbackHandler覆写事件钩子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]}...) # 通过 config 传入回调 result await agent.ainvoke( {messages: [(user, query)]}, config{callbacks: [CustomCallbackHandler()]} )八、流式输出Token 级与事件级实时体验依赖流式响应。SKILL.md 的 details 层给出两种粒度from langchain_anthropic import ChatAnthropic llm ChatAnthropic(modelclaude-sonnet-5, streamingTrue) # 1. Token 级流式输出 async for chunk in llm.astream(Tell me a story): print(chunk.content, end, flushTrue) # 2. Agent 事件流v2 协议 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]}])第二种方式对前端很有价值既能看到生成文本的逐步输出又能在on_tool_start事件时向用户展示当前正在调用哪个工具。命令 langchain-agent.md 进一步给出了 FastAPI 中以StreamingResponsetext/event-stream封装该能力的上层做法。九、测试策略单元测试 Agent 行为SKILL.md 给出了两个典型的测试维度——工具选择与记忆持久化import pytest from unittest.mock import AsyncMock, patch pytest.mark.asyncio async def test_agent_tool_selection(): 验证 Agent 选择了正确的工具。 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)] }) # 校验工具被调用 assert search_database in str(result) pytest.mark.asyncio async def test_memory_persistence(): 验证记忆在多次调用间持久化。 config {configurable: {thread_id: test-thread}} # 第一条消息 await agent.ainvoke( {messages: [(user, Remember: the code is 12345)]}, config ) # 第二条消息应能回忆起 result await agent.ainvoke( {messages: [(user, What was the code?)]}, config ) assert 12345 in result[messages][-1].content记忆持久化测试是 Checkpointer 正确性的直接验证只要两次调用使用相同thread_idAgent 的输出必须包含第一条消息中的事实。命令 langchain-agent.md 还建议引入langsmith.evaluation的自动化评测如qa、context_qa、cot_qaevaluator将评估纳入 CI。十、性能优化缓存、异步批处理与连接池1. Redis 语义缓存对重复度高的请求用 Redis 缓存 LLM 响应可显著降低延迟与成本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))set_llm_cache设置的是全局缓存接入后相同输入的调用会直接命中缓存返回。命令 langchain-agent.md 建议为其设置 TTL避免缓存过期数据。2. 异步批处理文档处理是典型的 IO 密集任务用asyncio.gather并行化import asyncio from langchain_core.documents import Document async def process_documents(documents: list[Document]) - list: 并行处理文档。 tasks [process_single(doc) for doc in documents] return await asyncio.gather(*tasks) async def process_single(doc: Document) - dict: 处理单个文档。 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. 向量库连接池避免每次请求都重新创建 Pinecone 客户端复用连接显著降低握手开销from langchain_pinecone import PineconeVectorStore from pinecone import Pinecone # 复用 Pinecone 客户端 pc Pinecone(api_keyos.environ[PINECONE_API_KEY]) index pc.Index(my-index) # 用已有 index 创建向量库 vectorstore PineconeVectorStore(indexindex, embeddingembeddings)十一、生产化清单综合 SKILL.md 与命令文档 langchain-agent.md落地生产前建议逐项核对异步优先全程使用ainvoke/astream/aembed_documents等异步 API避免阻塞事件循环错误处理工具函数与 LLM 调用内建 try/except配合tenacity实现指数退避重试如stop_after_attempt(3)wait_exponential可观测LangSmith 全链路追踪 结构化日志 健康检查校验 LLM、工具、记忆、外部服务连通性成本控制Redis 缓存、Token 上限、记忆压缩、批量嵌入密钥安全全部使用环境变量禁止硬编码如PINECONE_API_KEY状态版本化Checkpointer 保证状态可复现、可回滚流式体验FastAPIStreamingResponse暴露 SSE 流。结语langchain-architecture技能SKILL.md为 LangChain 1.x / LangGraph 应用提供了从 Agent 骨架、状态定义、记忆选型到深度工作流模式的完整架构模板。配合 details.md 中的 RAG 链路、结构化工具、条件路由、多 Agent 协作四个可复用模式以及本插件相邻的rag-implementation、embedding-strategies、llm-evaluation等技能即可在统一方法论下搭建生产级 LLM 应用。安装该插件可执行/plugin install llm-application-dev随后通过/llm-application-dev:langchain-agent命令直接生成 LangGraph Agent 骨架。【免费下载链接】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），仅供参考