《实用笔记：LangGraph思维模型——标准化架构指南》
《实用笔记操作指南：LangGraph思维模型——标准化架构指南：适用于采用该模式的团队的契约、校验规则及可直接插入的代码模块》。
以下笔记为《LangGraph心智模型：您构建的每一个智能体都适用的标准化架构指南》提供了一条实用路径。重点在于契约、校验以及可直接插入的代码占位符，而非激励性表述。 在阅读概览部分时，首先写下契约内容：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持清晰。 优先选择小型、可测试的单元，而非庞大的脚本。当某一步骤失败时，故障应指向单一责任点，而非复杂的流程链。
引言：为何即便概念不难，LangGraph代码却显得复杂
简介：为何即使概念本身没有问题，LangGraph代码依然难以理解？将其视为可测量的表面来处理效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，杜绝默许的半完成状态。 为每轮及每次会话设定token预算。智能工具会大量扩展上下文；设置上限可避免演示变成意外的费用账单。
整体架构：四个模块，一种文件顺序
整体概览：将“四个模块、一个文件顺序”的结构视为可度量的框架最为有效。在扩大范围之前，先记录一份优秀的测试用例、一个故障案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。为每轮及每次会话设定令牌预算——智能工具往往会大量消耗上下文资源，设置上限能防止演示过程变成意外收费的源头。
langgraph_agent.py
│
├── MODULE 1: IMPORTS & CONFIGURATION
│ └── All your libraries, API keys, model setup
│
├── MODULE 2: STATE
│ └── The TypedDict that defines your agent's memory
│
├── MODULE 3: TOOLS (optional, but common)
│ └── Functions decorated with @tool that the LLM can call
│
├── MODULE 4: NODES
│ └── Functions that do the actual work at each graph step
│
├── MODULE 5: EDGES & ROUTING
│ └── Functions that decide what happens next
│
├── MODULE 6: GRAPH ASSEMBLY
│ └── Where you build, wire, and compile the graph
│
└── MODULE 7: ENTRYPOINT
└── The __main__ block or invoke() call that runs everything
模块1：导入与配置
模块1：导入与配置功能若被视为可度量的对象，其效果会更好。在扩大范围之前，先记录一份理想的运行案例、一个故障实例以及回滚说明。 应将配置信息置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 为每轮对话和每次会话设定预算令牌限制。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程突然产生额外费用。
概念
将该概念视为可测量的界面来处理时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品的一部分，而非后续需要补充的内容。为每轮对话和每次会话设定token预算。智能工具会大量消耗上下文，设置上限可避免演示过程变成意外的费用账单。
你需要了解的关键术语
《你需要了解的关键词》这一方法在被视为可度量的指标时效果最佳。在扩大范围之前，先记录一份优秀的案例、一个失败实例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体的责任主体，而非复杂的流程链。 需为每轮对话和每次会话设定token预算。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。
标准模板
将标准模板视为可度量的对象使用效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默完成部分任务。 为每轮及每次会话设定token预算。智能工具会大量消耗上下文资源；设置上限可避免演示过程变成意外的费用账单。
# --- Standard Library ---
import os
from typing import TypedDict, Annotated, Literal
# --- LangChain Core ---
from langchain_openai import ChatOpenAI # or ChatAnthropic, ChatGroq, etc.
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
from langchain_core.tools import tool
# --- LangGraph Core ---
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages # The message reducer
from langgraph.prebuilt import ToolNode # Pre-built node for tool execution
from langgraph.checkpoint.memory import MemorySaver
# --- Configuration ---
# Always name your model variable 'llm' - it's the standard in every node
llm = ChatOpenAI(
model="gpt-4o", # or "claude-3-5-sonnet-20241022", etc.
temperature=0, # 0 = deterministic; raise for creativity
api_key=os.environ.get("OPENAI_API_KEY")
)
为何要采用这种结构
为何将这种特定结构视为可测量的界面时效果最佳？在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外账单。为每轮及每次会话设定令牌预算——智能工具往往会大量消耗上下文，设置上限可防止演示过程变成意外收费的源头。
模块2：状态
模块2：将状态视为可测量的对象来处理效果最佳。在扩大范围之前，先记录一份优秀的案例、一个失败案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放，以便操作人员无需查看整个系统结构即可进行审计。 为每轮对话和每次会话设定预算令牌限制。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。
概念
将该概念视为可测量的界面来处理时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品的一部分，而非后续需要补充的内容。为每轮对话和每次会话设定token预算。智能工具会大量消耗上下文，设置上限可避免演示过程变成意外的费用账单。
你需要了解的关键术语
《你需要了解的关键词》这一方法在被视为可度量的指标时效果最佳。在扩大范围之前，先记录一份优秀的案例、一个失败实例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体责任方，而非复杂的流程问题。 为每轮对话和每次会话设定token预算。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。
标准模板
将标准模板视为可度量的对象使用效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默地只完成部分工作。 为每轮及每次会话设定token预算。智能工具往往会过度扩展上下文；设置上限可避免演示变成意外的费用账单。
# ============================================================
# MODULE 2: STATE
# ============================================================
class AgentState(TypedDict):
# 'messages' is the heartbeat of almost every LangGraph agent.
# Annotated[list, add_messages] means: "this is a list, and when
# a node writes to it, append - don't replace."
messages: Annotated[list[BaseMessage], add_messages]
# Add custom fields below for your specific agent's needs.
# Fields without a reducer are REPLACED each time a node writes to them.
# Example: a simple string field (gets replaced each write)
current_task: str
# Example: a list you want to accumulate (use operator.add as reducer)
# results: Annotated[list[str], operator.add]
# Example: a counter
# iteration_count: int
简化思维模型
将“简化器思维模型”视为可测量的界面使用效果最佳。在扩大范围之前，先记录一份优秀的处理案例、一个失败案例以及回滚说明。在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外账单。为每轮及每次会话设定令牌预算。智能工具往往会大量消耗上下文资源，设置上限可防止演示环境变成令人意外的收费项目。
模块3：工具
模块3：将工具视为可度量的对象使用效果最佳。在扩大范围之前，先记录一份优秀的处理案例、一个故障实例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放，以便操作人员无需查看全部内容即可进行审计。 为每轮对话和每次会话设定预算令牌限制。智能工具会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。
概念
将该概念视为可测量的界面来处理时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品的一部分，而非后续需要补充的内容。为每轮对话和每次会话设定token预算。智能工具会大量消耗上下文，设置上限可避免演示过程变成意外的费用账单。
你需要了解的关键术语
《你需要了解的关键词》这一方法在被视为可度量的指标时效果最佳。在扩大范围之前，先记录一份优秀的案例、一个失败实例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体的责任主体，而非复杂的流程链。 为每轮对话和每次会话设定token预算。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。
标准模板
将标准模板视为可测量的对象使用效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默地只完成部分工作。 需为每轮及每次会话设定token预算。智能工具往往会过度扩展上下文；设置上限可避免演示过程变成意外的费用账单。
# ============================================================
# MODULE 3: TOOLS
# ============================================================
@tool
def search_web(query: str) -> str:
"""Search the web for current information about a topic.
Use this when you need real-time information that is not in
your training data, such as recent news or live prices.
Args:
query: The search query string.
Returns:
A string containing search results.
"""
# Your actual implementation here (e.g., Tavily, SerpAPI, etc.)
# Placeholder for illustration:
return f"Search results for: {query}"
@tool
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression and return the result.
Use this for any arithmetic, algebra, or numerical computation.
Args:
expression: A valid Python math expression as a string, e.g. '2 + 2 * 10'
Returns:
The computed result as a string.
"""
try:
return str(eval(expression))
except Exception as e:
return f"Error: {e}"
# Collect all tools into a list - this is the pattern you always follow
tools = [search_web, calculate]
# Bind tools to the LLM so it knows they exist and can choose to call them
llm_with_tools = llm.bind_tools(tools)
# Create the pre-built ToolNode that will execute tool calls automatically
tool_node = ToolNode(tools)
工具文档字符串至关重要
“工具文档是至关重要的”，将其视为可度量的指标时效果最佳。在扩大范围之前，先记录一份理想的测试用例、一个故障案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外账单。为每轮对话和每次会话设定令牌预算。智能工具会大量消耗上下文资源，设置上限可防止演示环境变成令人意外的收费来源。
第4模块：节点
第4模块：将节点视为可测量的表面来处理时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放，这样操作人员无需查看整个结构即可进行审计。 为每轮及每次会话设定预算令牌限制。智能工具会大量消耗上下文资源；设置上限可避免演示过程变成意外的费用账单。
概念
将该概念视为可测量的界面来处理时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品的一部分，而非后续需要补充的内容。为每轮对话和每次会话设定token预算。智能工具会大量消耗上下文，设置上限可避免演示过程变成意外的费用账单。
你需要了解的关键术语
《你需要了解的关键词》这一方法在被视为可度量的指标时效果最佳。在扩大范围之前，先记录一份优秀的案例、一个失败实例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤出错时，故障应能指向单一责任主体，而非复杂的流程链。 为每轮对话和每次会话设定token预算。智能工具会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。
标准模板
将标准模板视为可测量的对象使用效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默完成不完整的工作。 为每轮及每次会话设定token预算。智能工具往往会过度扩展上下文；设置上限可避免演示变成意外的费用账单。
# ============================================================
# MODULE 4: NODES
# ============================================================
# ── Node: Agent (the reasoning brain) ───────────────────────
def agent_node(state: AgentState) -> dict:
"""The central reasoning node. Calls the LLM and decides
whether to respond or call a tool."""
# Build the message list to send to the LLM.
# Always include a system message to set behavior.
system_prompt = SystemMessage(content=(
"You are a helpful assistant. Use the available tools "
"when you need real-time information or computation. "
"Respond clearly and concisely."
))
# The LLM receives the system prompt + all previous messages in state
messages_to_send = [system_prompt] + state["messages"]
# Call the LLM. Use llm_with_tools if you have tools; plain llm if not.
response = llm_with_tools.invoke(messages_to_send)
# Return the LLM's response as a state update.
# add_messages will APPEND this AIMessage to state["messages"].
return {"messages": [response]}
# ── Node: Summarizer (example of a non-LLM processing node) ─
def summarize_node(state: AgentState) -> dict:
"""Summarizes the conversation so far to keep context short.
This shows that nodes don't have to call an LLM - they can
do any Python processing."""
all_messages = state["messages"]
# Summarize with the LLM (a different prompt, same LLM)
summary_prompt = [
SystemMessage(content="Summarize the following conversation in 2-3 sentences."),
HumanMessage(content=str(all_messages))
]
summary_response = llm.invoke(summary_prompt)
# Replace messages with a fresh start containing just the summary
return {
"messages": [AIMessage(content=f"[Summary] {summary_response.content}")]
}
节点心智模型
将节点思维模型视为可测量的对象时，其效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。在功能结果旁同时记录执行时间以及代币或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。为每轮及每次会话设定代币预算。智能工具往往会大量消耗上下文资源，设置上限能防止演示过程变成意外收费的源头。
模块5：边缘节点与路由
模块5：将边缘计算与路由处理视为可测量的表面，效果最佳。在扩大范围之前，先记录一份优秀的操作日志、一个故障案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放，以便操作人员无需查看整个系统结构即可进行审计。 为每轮对话和每次会话设定预算令牌限制。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程突然产生额外费用。
概念
将该概念视为可测量的界面来处理时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品的一部分，而非后续需要补充的内容。为每轮对话和每次会话设定token预算。智能工具会大量消耗上下文，设置上限可避免演示过程变成意外的费用账单。
你需要了解的关键术语
《你需要了解的关键词》这一方法在被视为可度量的指标时效果最佳。在扩大范围之前，先记录一份优秀的案例、一个失败实例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体的责任主体，而非复杂的流程链。 为每轮对话和每次会话设定token预算。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。
标准模板
将标准模板视为可测量的对象使用效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默完成不完整的工作。 为每轮及每次会话设定token预算。智能工具会大量消耗上下文资源；设置上限可避免演示过程变成意外的费用账单。
# ============================================================
# MODULE 5: EDGES & ROUTING
# ============================================================
# Import the pre-built tool routing function
from langgraph.prebuilt import tools_condition
# ── Custom Routing Function Example ─────────────────────────
def should_continue(state: AgentState) -> Literal["tools", "summarize", "__end__"]:
"""Custom router for the agent node.
Routing functions always:
1. Receive the current state as input
2. Return a string that maps to the next node (or END)
The return values must match the keys in add_conditional_edges' mapping.
"""
last_message = state["messages"][-1] # Look at what the LLM just said
# Case 1: The LLM decided to call a tool
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
# Case 2: The conversation is getting long - summarize before continuing
if len(state["messages"]) > 20:
return "summarize"
# Case 3: The LLM gave a direct answer - we're done
return "__end__" # LangGraph's internal name for END
路由思维模型
将路由思维模型视为可测量的对象时，其效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。为每轮及每次会话设定令牌预算。智能工具往往会大量消耗上下文资源，设置上限能防止演示环境变成令人意外的收费来源。
第6模块：图结构构建
第6模块：将图结构视为可测量的表面来处理时，其组装效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将配置信息置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个图结构即可进行审计。 为每轮及每次会话设定令牌预算。智能工具会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。
概念
将该概念视为可测量的界面来处理时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品的一部分，而非后续需要补充的内容。为每轮对话和每次会话设定token预算。智能工具会大量消耗上下文，设置上限可避免演示过程变成意外的费用账单。
你需要了解的关键术语
《你需要了解的关键词》这一方法在被视为可度量的指标时效果最佳。在扩大范围之前，先记录一份优秀的案例、一个失败实例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤出错时，故障应能指向单一责任主体，而非复杂的流程链。 为每轮对话和每次会话设定token预算。智能工具会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。
标准模板
将标准模板视为可测量的对象使用效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默完成不完整的工作。 为每轮及每次会话设定token预算。智能工具会大量消耗上下文资源；设置上限可避免演示过程变成意外的费用账单。
# ============================================================
# MODULE 6: GRAPH ASSEMBLY
# ============================================================
# ── Step 1: Initialize ──────────────────────────────────────
# Always pass your State class to StateGraph
graph_builder = StateGraph(AgentState)
# ── Step 2: Register All Nodes ──────────────────────────────
# Format: add_node("node_name_as_string", node_function)
# The string name is what you use in ALL edge definitions
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", tool_node) # The pre-built ToolNode from Module 3
graph_builder.add_node("summarize", summarize_node)
# ── Step 3: Set Entry Point ─────────────────────────────────
# Which node runs first when we invoke the graph?
graph_builder.set_entry_point("agent")
# ── Step 4: Wire the Edges ──────────────────────────────────
# Conditional edge from agent: check if we need tools, a summary, or we're done
graph_builder.add_conditional_edges(
"agent", # Source node
should_continue, # Routing function from Module 5
{
"tools": "tools", # If router returns "tools" → go to tools node
"summarize": "summarize", # If router returns "summarize" → go to summarize node
"__end__": END, # If router returns "__end__" → stop the graph
}
)
# Static edge: after tools run, always go back to agent (the ReAct loop)
graph_builder.add_edge("tools", "agent")
# Static edge: after summarization, always return to agent
graph_builder.add_edge("summarize", "agent")
# ── Step 5: Compile ─────────────────────────────────────────
# Without checkpointer: no persistent memory (stateless per invocation)
# With checkpointer: memory persists across turns (stateful conversations)
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
组装心智模型
将“装配心智模型”视为可测量的对象时，其效果最佳。在扩大应用范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外账单。为每轮及每次会话设定令牌预算。智能工具往往会大量消耗上下文资源，设置上限可防止演示过程变成意外收费的源头。
第7模块：入口点与调用
第7模块：将入口点与调用过程视为可度量的对象，这样才能实现最佳效果。在扩大范围之前，先记录一份理想的操作日志、一个故障案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放，这样操作人员无需查看整个系统结构即可进行审计。 为每轮对话和每次会话设定令牌预算。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。
概念
将该概念视为可测量的界面来处理时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品的一部分，而非后续需要补充的内容。为每轮对话和每次会话设定token预算。智能工具会大量消耗上下文，设置上限可避免演示过程变成意外的费用账单。
你需要了解的关键术语
《你需要了解的关键词》这一方法在被视为可度量的指标时效果最佳。在扩大范围之前，先记录一份优秀的案例、一个失败实例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体的责任主体，而非复杂的流程链。 为每轮对话和每次会话设定token预算。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程变成意外的费用账单。
标准模板
将标准模板视为可测量的对象使用效果最佳。在扩大范围之前，先记录一份理想的成果、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默完成部分任务。 为每轮及每次会话设定token预算。智能工具会大量消耗上下文资源；设置上限可避免演示过程变成意外的费用账单。
# ============================================================
# MODULE 7: ENTRYPOINT & INVOCATION
# ============================================================
if __name__ == "__main__":
# ── Config: defines this conversation's memory session ──
# Change thread_id to start a fresh conversation.
# Keep the same thread_id to continue an existing one.
config = {"configurable": {"thread_id": "user-session-001"}}
# ── Single Invocation (synchronous) ─────────────────────
user_input = "What is the current price of Bitcoin?"
result = graph.invoke(
input={"messages": [HumanMessage(content=user_input)]},
config=config
)
# The result is the final state dictionary.
# Access the last message to get the agent's final answer.
final_answer = result["messages"][-1].content
print(f"Agent: {final_answer}")
# ── Streaming Invocation (for real-time output) ──────────
for chunk in graph.stream(
input={"messages": [HumanMessage(content=user_input)]},
config=config,
stream_mode="values" # Yields the full state after each node runs
):
# Each chunk is a state snapshot. The last message shows progress.
latest = chunk["messages"][-1]
if hasattr(latest, "content") and latest.content:
print(f"[Streaming] {latest.content}")
# ── Multi-turn Conversation Loop ─────────────────────────
print("\n--- Starting Interactive Session ---")
while True:
user_text = input("You: ").strip()
if user_text.lower() in ("exit", "quit", "bye"):
break
response = graph.invoke(
input={"messages": [HumanMessage(content=user_text)]},
config=config # Same config = same memory thread
)
print(f"Agent: {response['messages'][-1].content}\n")
完整规范模板：全部文件
完整规范模板：将整个文件视为可度量的对象使用效果最佳。在扩大范围之前，先记录一个理想案例、一个失败案例以及回滚说明。在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外账单。为每轮及每次会话设定令牌预算。智能代理工具会大量消耗上下文资源，设置上限能防止演示过程变成意外收费的源头。
# ============================================================
# LANGGRAPH CANONICAL AGENT TEMPLATE
# Modules: Imports → State → Tools → Nodes → Edges → Assembly → Entrypoint
# ============================================================
# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
import os
from typing import TypedDict, Annotated, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# ── MODULE 2: STATE ─────────────────────────────────────────
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# Add your custom fields here
# ── MODULE 3: TOOLS ─────────────────────────────────────────
@tool
def my_tool(input: str) -> str:
"""Describe clearly what this tool does and when the LLM should use it."""
return f"Result for: {input}"
tools = [my_tool]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)
# ── MODULE 4: NODES ─────────────────────────────────────────
def agent_node(state: AgentState) -> dict:
"""The reasoning node. Calls the LLM, optionally triggers tool calls."""
messages = [SystemMessage(content="You are a helpful assistant.")] + state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
# ── MODULE 5: EDGES & ROUTING ───────────────────────────────
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
"""Decide: did the LLM call a tool, or did it give a final answer?"""
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "__end__"
# ── MODULE 6: GRAPH ASSEMBLY ────────────────────────────────
graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", tool_node)
graph_builder.set_entry_point("agent")
graph_builder.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "__end__": END}
)
graph_builder.add_edge("tools", "agent")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
# ── MODULE 7: ENTRYPOINT ────────────────────────────────────
if __name__ == "__main__":
config = {"configurable": {"thread_id": "session-001"}}
while True:
user_text = input("You: ").strip()
if not user_text or user_text.lower() in ("exit", "quit"):
break
response = graph.invoke(
{"messages": [HumanMessage(content=user_text)]},
config=config
)
print(f"Agent: {response['messages'][-1].content}\n")
高级模块：多智能体系统
高级模块：将多智能体系统视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份理想的操作日志、一个故障案例以及回滚说明。 将配置信息置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放，以便操作人员无需查看整个系统结构即可进行审计。 为每轮对话和每次会话设定token预算。智能体工具往往会大量消耗上下文资源，设置上限可避免演示过程突然产生额外费用。 高级模块：将多智能体系统视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份理想的操作日志、一个故障案例以及回滚说明。 相比庞大的脚本，应优先使用小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个特定责任模块，而非复杂的流程链。
你需要了解的关键术语（多智能体）
对于需要了解的关键词（多智能体），在修改代码之前应明确输入内容、该步骤的负责人以及终止标准。操作人员应当能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 当下一步操作是编写代码或调用工具时，应优先选择具有架构验证的结构化输出，而非自由形式的文字描述。
多智能体结构模板
对于多智能体结构模板，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前显示成本信息，可避免在从演示环境切换到共享环境时出现意外费用。当下一步操作为编写代码或调用工具时，优先选择经过模式验证的结构化输出，而非自由形式的文字描述。
# ── MULTI-AGENT PATTERN ─────────────────────────────────────
# Each specialist is a compiled graph (a subgraph)
# Sub-agent 1: A researcher
researcher_graph = StateGraph(AgentState)
# ... (built with its own nodes, edges, and tools)
researcher = researcher_graph.compile()
# Sub-agent 2: A writer
writer_graph = StateGraph(AgentState)
# ... (built with its own nodes, edges, and tools)
writer = writer_graph.compile()
# ── SUPERVISOR NODE ─────────────────────────────────────────
def supervisor_node(state: AgentState) -> dict:
"""Decides which sub-agent should handle the current task."""
# The supervisor LLM decides: "researcher" or "writer" or "FINISH"
response = supervisor_llm.invoke(state["messages"])
return {"messages": [response], "next_agent": response.content}
def route_to_agent(state: AgentState) -> Literal["researcher", "writer", "__end__"]:
"""Routes to the appropriate sub-agent based on supervisor's decision."""
return state.get("next_agent", "__end__")
# ── SUPERVISOR GRAPH ────────────────────────────────────────
supervisor_builder = StateGraph(AgentState)
supervisor_builder.add_node("supervisor", supervisor_node)
supervisor_builder.add_node("researcher", researcher) # Subgraph as a node!
supervisor_builder.add_node("writer", writer) # Subgraph as a node!
supervisor_builder.set_entry_point("supervisor")
supervisor_builder.add_conditional_edges(
"supervisor",
route_to_agent,
{"researcher": "researcher", "writer": "writer", "__end__": END}
)
supervisor_builder.add_edge("researcher", "supervisor")
supervisor_builder.add_edge("writer", "supervisor")
supervisor_graph = supervisor_builder.compile(checkpointer=MemorySaver())
关键词参考卡
对于关键词参考卡，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审计。 当下一步操作为代码编写或工具调用时，应优先使用具有结构化格式且经过模式验证的输出，而非自由形式的文字描述。 对于关键词参考卡，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相比冗长的脚本，应优先选择小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向某个特定功能模块，而非整个复杂的流程。
结论：LangGraph的肌肉记忆
在撰写“结论：LangGraph的肌肉记忆”时，首先列出相关契约：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 缓存稳定的系统指令和工具架构。重复发送相同的开头信息是导致效率低下的常见原因。
操作检查清单
针对操作检查清单，在修改代码之前需明确输入内容、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。
请同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续才添加的功能。
当下一步操作是编写代码或调用工具时，应优先使用带有模式验证的结构化输出，而非自由形式的文字描述。
在成本较高的操作之后设置检查点。当操作员重新尝试后续节点时，恢复功能不应再次计费相同的大型语言模型调用费用。
锁定依赖项的版本，并记录用于演示的镜像摘要。可重复性远比团队内部的知识更重要。
在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境过渡到共享环境时出现意外账单。
在推广该技术栈之前，应先冻结版本，为关键流程记录标准化的文本转录结果，并明确回滚步骤。共享环境需要设置速率限制、租户验证机制，以及负责密钥轮换的明确责任人。与其展示花哨的一次性演示，不如注重扎实的可靠性。
d02265f3bebf的批量处理说明：不要将提供商密钥放入代码仓库，为每个会话设置令牌使用上限，并将文本转录结果存储在评估用示例文件旁，以便后续模型更换时保持数据可比性。