企业官网建设流程全解析

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 将能力拆分为多个独立包，职责边界清晰：

这一拆分意味着：核心抽象（消息、提示词、工具）依赖langchain-core，Agent 编排依赖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 编排模式：

1. ReAct（推理 + 行动）：通过create_react_agent快速创建，Agent 循环执行"思考 → 调用工具 → 观察结果 → 再思考"；
2. Plan-and-Execute：将"规划"与"执行"分离为独立节点，适合复杂任务的前置规划；
3. Multi-Agent（多 Agent 协作）：由 Supervisor（监督者）在多个专精 Agent 之间路由；
4. 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 给出的记忆选型如下：

开发环境：内存 Checkpointer

from langgraph.checkpoint.memory import MemorySaver # 开发环境使用内存检查点 checkpointer = MemorySaver() agent = create_react_agent(llm, tools, checkpointer=checkpointer) # 每个 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", "What's my name?")]}, config) # Agent 能回忆起："Your name is Alice"

生产环境：PostgreSQL Checkpointer

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

长期记忆：向量库语义检索

对于需要跨会话长期保留的业务事实，details.md 展示了以 Chroma + Voyage AI 实现语义记忆的方案：写入时aadd_texts存入文本与元数据，读取时asimilarity_search按语义召回相关历史片段。这与插件中vector-database-engineerAgent 的职责（agents/vector-database-engineer.md）相互呼应。

五、快速上手：基于 LangGraph 的现代 ReAct Agent

SKILL.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(model="claude-sonnet-5") # 定义工具（@tool 装饰器 + 类型注解即生成 Pydantic Schema） @tool def search_database(query: str) -> str: """Search internal database for information.""" # 你的数据库检索逻辑 return f"Results 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(f"Unsupported operation: {type(node)}") try: tree = ast.parse(expression, mode='eval') return str(_eval(tree.body)) except Exception as e: return f"Error: {e}" tools = [search_database, calculate] # 创建 Checkpointer 实现记忆持久化 checkpointer = MemorySaver() # 创建 ReAct Agent agent = create_react_agent( llm, tools, checkpointer=checkpointer ) # 通过 thread_id 关联会话记忆 config = {"configurable": {"thread_id": "user-123"}} result = await agent.ainvoke( {"messages": [("user", "Search for Python tutorials and calculate 25 * 4")]}, config=config )

值得注意的工程细节：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(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}) # 定义节点 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(context=context_text, question=state["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定义参数 Schema：

from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field class SearchInput(BaseModel): """数据库检索入参。""" query: str = Field(description="Search query") filters: dict = Field(default={}, description="Optional filters") class EmailInput(BaseModel): """邮件发送入参。""" 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.""" 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.""" 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)

要点：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 = 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: """分析抽取的实体。""" 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: """生成最终摘要。""" 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"]: """依据当前状态路由到下一步。""" 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 = 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"]: """基于 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(model="claude-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(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]}...") # 通过 config 传入回调 result = await agent.ainvoke( {"messages": [("user", "query")]}, config={"callbacks": [CustomCallbackHandler()]} )

八、流式输出：Token 级与事件级

实时体验依赖流式响应。SKILL.md 的 details 层给出两种粒度：

from langchain_anthropic import ChatAnthropic llm = ChatAnthropic(model="claude-sonnet-5", streaming=True) # 1. Token 级流式输出 async for chunk in llm.astream("Tell me a story"): print(chunk.content, end="", flush=True) # 2. Agent 事件流（v2 协议） 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']}]")

第二种方式对前端很有价值：既能看到生成文本的逐步输出，又能在on_tool_start事件时向用户展示"当前正在调用哪个工具"。命令 langchain-agent.md 进一步给出了 FastAPI 中以StreamingResponse+text/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(content="Using 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_id，Agent 的输出必须包含第一条消息中的事实。命令 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_key=os.environ["PINECONE_API_KEY"]) index = pc.Index("my-index") # 用已有 index 创建向量库 vectorstore = PineconeVectorStore(index=index, embedding=embeddings)

十一、生产化清单

综合 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