1. 项目概述告别重复阅读让智能体“记住”实验过程最近在折腾AI智能体做自动化实验时碰到一个特别头疼的问题成本。不是硬件成本而是调用大模型API的“令牌”Token成本。每次智能体执行一个步骤比如“读取传感器A的数值”它都需要把之前所有的对话历史、实验步骤、观察结果一股脑儿地再喂给模型让它“回忆”上下文。实验步骤一多这个上下文窗口就像个无底洞疯狂吞噬令牌账单看着就肉疼。这就像你每走一步路都得把从家到现在的整条路线重新背一遍效率低得令人发指。“Remember, Don‘t Re-read”这个思路就是冲着解决这个痛点来的。它的核心思想是让智能体变得“有状态”Stateful。不再是每次行动都依赖完整的对话历史而是维护一个精简、结构化的“记忆体”或“状态”。智能体基于当前状态做决策执行动作然后将结果更新到状态中如此循环。这极大地减少了每次推理时需要处理的文本量从而实现了“令牌高效”Token-Efficient。结合ReActReasoning Acting框架智能体不仅能规划Think还能执行Act并通过观察Observe来更新自己的认知形成一个自主实验的闭环。而LangGraph这类工具则为构建这种有状态的、多步骤的工作流提供了绝佳的框架。简单说这项目就是教你如何打造一个“记性好、不啰嗦、会自己动手做实验”的AI智能体特别适合需要长期、多步骤交互的自动化场景比如材料模拟、代码调试、化学实验设计等目标就一个用更少的钱令牌干更多的活。2. 核心架构与设计哲学状态驱动的ReAct循环传统的ReAct智能体其工作流可以概括为“思考-行动-观察”的循环但这个循环是“无状态”或“弱状态”的。通常我们通过LangChain的AgentExecutor把整个对话历史作为上下文传递给大模型。模型基于全部历史生成下一个“思考”和“行动”。问题显而易见随着循环进行历史越来越长令牌消耗呈线性甚至指数增长。“Remember, Don‘t Re-read”架构的核心转变在于将智能体的核心从“完整的对话历史”转移到一个精心设计的“状态对象”上。2.1 状态State的设计记忆的骨架状态是整个系统的灵魂。它不是一个简单的字符串而是一个结构化的字典Pydantic模型是绝佳选择只保存对后续决策真正必要的信息。一个好的状态设计应该像实验记录本只记关键数据和结论不记流水账。一个典型的实验智能体状态可能包含以下字段from typing import TypedDict, List, Optional from datetime import datetime class ExperimentState(TypedDict): # 实验目标 objective: str # 当前步骤的假设或计划 current_hypothesis: Optional[str] # 已执行的操作序列精简描述 action_history: List[str] # 关键的观察结果数据、现象 key_observations: List[str] # 从观察中推导出的结论或学习 learnings: List[str] # 实验是否完成 is_complete: bool # 当前步骤编号 step_count: int # 任何错误或异常信息 error: Optional[str]为什么这么设计objective这是实验的“北极星”必须始终在场确保智能体不跑偏。current_hypothesis记录当前步骤的推理依据替代了冗长的“思考”过程文本。action_history和key_observations只记录“做了什么”和“看到了什么”的精华而不是完整的自然语言交互。例如记录“加热至150°C持续5分钟”和“观察到溶液由蓝变绿pH7.2”而不是包含所有修饰词的句子。learnings这是压缩后的“知识”。当观察积累到一定阶段智能体或一个专门的“总结节点”会提炼出规律如“温度超过160°C会导致产物分解”并存入此处。后续决策直接参考这些learnings无需重新分析原始观察记录。is_complete,step_count,error控制流和异常处理所必需的基础信息。通过这种方式每次调用大模型时我们只需要传递这个结构化的状态经过序列化而不是整个对话历史。状态的大小是可控的不会无限增长。2.2 LangGraph的角色工作流的调度中心LangGraph是实现这一架构的“神器”。它允许你将智能体的不同能力思考节点、执行节点、更新节点定义成图中的节点Node并通过边Edge来规定它们之间的流转逻辑。在这个状态化ReAct智能体中一个典型的工作流图可能包含以下节点Agent代理节点接收当前State调用大模型。提示词Prompt会基于State中的关键信息如objective,current_hypothesis,learnings来构建询问模型“基于当前状态和我们的目标下一步应该做什么思考以及具体执行什么动作行动”。模型的输出会被解析成结构化的指令。Tools工具执行节点接收来自Agent节点的结构化动作指令调用对应的工具函数如调用实验设备API、查询数据库、运行模拟软件。这个节点是“行动”的实体。Observer观察解析节点接收工具执行返回的原始结果可能是一堆数据、一段文本、一个错误码。这个节点的职责是将原始结果“翻译”或“提炼”成能够更新状态的信息。例如从一长串传感器日志中提取出“平均温度152.3°C”和“状态稳定”这两个关键观察。Updater状态更新节点这是“记忆”形成的关键。它接收来自Observer的提炼信息以及之前的State按照既定规则更新State。例如将新的观察追加到key_observations根据一系列观察推理出一个新的learning并存入增加step_count判断is_complete等。这个节点不一定需要大模型可以用确定性逻辑或轻量级模型实现进一步节省令牌。这些节点通过条件边Conditional Edge连接。例如从Agent节点出来后根据其输出的“动作类型”决定下一步是去Tools节点还是直接去Updater节点如果模型认为需要先更新某个学习结论。从Observer节点出来后总是进入Updater节点。更新完状态后根据State.is_complete和State.error的值决定是循环回Agent节点进行下一步还是结束工作流。提示在设计Updater节点时一个重要的技巧是“定期总结”。不要每步都总结而是在step_count达到5的倍数时或者当key_observations积累到一定数量时触发一个“总结子流程”调用一次大模型来生成新的learning。这平衡了令牌消耗和记忆质量。3. 实操构建从零搭建一个令牌高效的实验智能体理论说再多不如动手做一遍。下面我们以“自动化化学实验条件探索”为例分步拆解如何用LangGraph构建一个状态化ReAct智能体。我们将使用OpenAI的GPT-4作为推理模型但架构是模型无关的。3.1 环境准备与状态定义首先安装核心库并定义我们的状态结构。我们使用Pydantic来获得更好的类型提示和验证。pip install langgraph langchain-openai pydanticfrom typing import List, Optional, Annotated from typing_extensions import TypedDict from pydantic import BaseModel, Field from datetime import datetime import operator # 定义状态结构 class ExperimentState(TypedDict): 实验智能体的核心状态 objective: str # 实验目标固定不变 parameters: dict # 当前实验参数如温度、浓度、时间 action_history: List[str] # 执行过的动作摘要 observation_history: List[str] # 观察结果摘要 learnings: List[str] # 学到的经验规律 step: int # 当前步骤 max_steps: int # 最大步骤限制 is_complete: bool # 是否完成 error: Optional[str] # 错误信息 # 使用LangGraph的注解来定义聚合方式 hypothesis: Annotated[Optional[str], operator.add] # 当前步骤的假设可累积或覆盖 # 初始化状态 initial_state ExperimentState( objective探索反应温度50-200°C和催化剂浓度0.1%-2%对产物收率的影响找到收率高于85%的条件。, parameters{temperature: 100, catalyst_concentration: 1.0}, # 初始猜测 action_history[], observation_history[], learnings[], step0, max_steps20, is_completeFalse, errorNone, hypothesisNone )这里我们引入了Annotated和operator.add这是LangGraph的一个高级特性用于声明状态字段在多个节点并行修改时的合并策略如追加、替换、求和等。对于hypothesis我们使用add意味着后续节点可以对其追加内容。3.2 构建工具Tools与模拟环境在真实场景中工具可能是控制精密仪器的SDK。这里我们模拟一个实验执行函数。from langchain.tools import tool import random tool def run_experiment(temperature: float, catalyst_concentration: float) - dict: 在给定温度和催化剂浓度下运行一次模拟实验返回收率和观察现象。 这是一个模拟函数真实场景应替换为真实的设备调用。 # 模拟一个简单的响应曲面收率与温度、浓度有关 # 假设最佳条件在 150°C, 1.5% ideal_temp, ideal_conc 150.0, 1.5 temp_factor max(0, 1 - abs(temperature - ideal_temp) / 100) conc_factor max(0, 1 - abs(catalyst_concentration - ideal_conc) / 2) base_yield 70.0 yield_adjustment (temp_factor * 0.2 conc_factor * 0.1) * 100 simulated_yield min(99.0, base_yield yield_adjustment random.uniform(-5, 5)) # 加一点随机噪声 observations [] if temperature 180: observations.append(反应剧烈有少量副产物烟雾产生。) if catalyst_concentration 0.5: observations.append(反应启动缓慢。) if simulated_yield 85: observations.append(产物晶体析出良好色泽纯正。) return { yield: round(simulated_yield, 2), observations: ; .join(observations) if observations else 反应平稳无明显异常。, parameters_used: {temperature: temperature, catalyst_concentration: catalyst_concentration} } # 将工具包装成LangChain可用的工具列表 tools [run_experiment]3.3 创建智能体Agent节点这个节点负责“思考”。它接收状态生成下一步的“假设/计划”和“动作”。from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain_core.prompts import PromptTemplate # 初始化大模型 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0.1) # 构建一个高度定制化的Prompt它只引用状态中的关键信息 AGENT_PROMPT PromptTemplate.from_template( 你是一个自动化化学实验AI。你的目标是{objective} ## 当前状态概览 - 已进行步骤{step} - 关键学习避免重复错误{learnings} - 上一步参数{last_params} - 上一步结果{last_result} ## 历史摘要最近3步 {recent_history} ## 你的任务 1. **推理Reason**基于以上信息提出一个关于下一步实验条件的假设Hypothesis。例如“根据学习‘高温导致分解’我假设将温度降低10°C可能提高收率。” 2. **行动Act**根据你的假设决定下一个动作。你只能使用以下工具 - run_experiment: 运行实验。输入必须包含 temperature (50-200) 和 catalyst_concentration (0.1-2.0)。 请严格按照以下格式回应 Thought: [你的推理过程] Hypothesis: [你的假设一句话] Action: [要执行的动作必须是 run_experiment] Action Input: {{temperature: 数值, catalyst_concentration: 数值}} ) def agent_node(state: ExperimentState): 智能体节点基于状态生成思考和行动指令。 # 准备Prompt所需的动态信息 recent_actions state[action_history][-3:] if state[action_history] else [无] recent_obs state[observation_history][-3:] if state[observation_history] else [无] recent_history \n.join([f- 动作{a}\n 观察{o} for a, o in zip(recent_actions, recent_obs)]) last_params state[parameters] last_result f收率{state.get(last_yield, N/A)}% 现象{state.get(last_observation, N/A)} if state[step]0 else 无 # 构建Prompt prompt AGENT_PROMPT.format_prompt( objectivestate[objective], stepstate[step], learnings; .join(state[learnings]) if state[learnings] else 暂无, last_paramslast_params, last_resultlast_result, recent_historyrecent_history ) # 调用大模型 response llm.invoke(prompt.to_string()) # 这里简化了解析实际应用应使用更鲁棒的解析器如OutputFixingParser # 假设返回的文本可以直接按行分割解析 lines response.content.split(\n) thought next((l.split(Thought: )[1] for l in lines if l.startswith(Thought: )), ) hypothesis next((l.split(Hypothesis: )[1] for l in lines if l.startswith(Hypothesis: )), ) action_line next((l for l in lines if l.startswith(Action: )), ) action_input_line next((l for l in lines if l.startswith(Action Input: )), ) # 返回的结果会作为更新状态的一部分 return { thought: thought, hypothesis: hypothesis, action: action_line.replace(Action: , ), action_input: eval(action_input_line.replace(Action Input: , )) if action_input_line else {}, # 其他需要传递给后续节点的信息... }关键点注意Prompt的设计。我们没有传入完整的action_history和observation_history而是只传入了最近3步的摘要recent_history和提炼出的learnings。这就是“Don‘t Re-read”的精髓——用压缩的、高价值的信息替代冗长的原始记录。3.4 创建工具执行与观察节点这个节点执行动作并初步处理结果。def tools_node(state: ExperimentState, agent_output: dict): 工具执行节点运行实验并获取原始结果。 action agent_output.get(action) action_input agent_output.get(action_input, {}) if action run_experiment: result run_experiment.invoke(action_input) # 将本次使用的参数也记录下来方便后续更新状态 return { tool_result: result, used_parameters: action_input } else: # 处理其他可能的工具调用... return {tool_result: {error: f未知工具: {action}}, used_parameters: {}}3.5 创建状态更新Updater节点这是最关键的节点负责将新信息“写入”记忆。def updater_node(state: ExperimentState, tools_output: dict): 状态更新节点整合观察结果更新状态并决定下一步。 new_state state.copy() # 1. 更新步骤计数 new_state[step] 1 # 2. 记录动作和观察摘要形式非完整对话 action_desc f步骤{new_state[step]}: 运行实验 {tools_output.get(used_parameters, {})} obs_data tools_output.get(tool_result, {}) obs_desc f收率{obs_data.get(yield, N/A)}% | {obs_data.get(observations, )} new_state[action_history].append(action_desc) new_state[observation_history].append(obs_desc) # 3. 更新当前参数 new_state[parameters] tools_output.get(used_parameters, new_state[parameters]) # 4. 存储最后一次结果供下次Agent节点使用 new_state[last_yield] obs_data.get(yield) new_state[last_observation] obs_data.get(observations) # 5. 定期总结生成Learning例如每5步或当收率有显著变化时 if new_state[step] % 5 0 and len(new_state[observation_history]) 3: # 这里可以调用一个轻量级的总结模型或者使用规则 # 为简化我们用一个规则如果最近三次实验收率都低于80%且温度都160则生成一个学习 recent_yields [] for obs in new_state[observation_history][-3:]: try: y float(obs.split(收率)[1].split(%)[0]) recent_yields.append(y) except: pass if len(recent_yields) 3 and all(y 80 for y in recent_yields) and all(new_state[parameters][temperature] 160 for _ in range(3)): learning 高温160°C可能导致本反应收率持续低于80%。 if learning not in new_state[learnings]: new_state[learnings].append(learning) # 6. 检查终止条件 if new_state[step] new_state[max_steps]: new_state[is_complete] True new_state[error] 达到最大步骤限制。 elif obs_data.get(yield, 0) 85.0: # 达到目标收率 new_state[is_complete] True new_state[error] None elif obs_data.get(error): new_state[is_complete] True new_state[error] obs_data.get(error) return new_state3.6 使用LangGraph组装工作流现在我们用LangGraph将各个节点连接起来形成一个完整的有状态图。from langgraph.graph import StateGraph, END # 1. 创建图 workflow StateGraph(ExperimentState) # 2. 添加节点 workflow.add_node(agent, agent_node) workflow.add_node(tools, tools_node) workflow.add_node(updater, updater_node) # 3. 设置边工作流逻辑 workflow.set_entry_point(agent) # Agent节点之后总是去执行工具 workflow.add_edge(agent, tools) # 工具执行后总是去更新状态 workflow.add_edge(tools, updater) # 4. 从Updater节点出来的条件边根据更新后的状态决定是继续循环还是结束 def decide_next_step(state: ExperimentState): if state[is_complete]: return END else: return agent # 继续下一轮ReAct循环 workflow.add_conditional_edges( updater, decide_next_step, { agent: agent, END: END } ) # 5. 编译图 app workflow.compile()3.7 运行与可视化现在我们可以运行这个智能体了。# 运行工作流 final_state None for step, output in enumerate(app.stream(initial_state, stream_modevalues)): node_name list(output.keys())[0] print(f\n--- 步骤 {step 1}: 执行节点 [{node_name}] ---) if node_name agent: print(f思考: {output[node_name].get(thought)}) print(f假设: {output[node_name].get(hypothesis)}) print(f行动: {output[node_name].get(action)} {output[node_name].get(action_input)}) elif node_name tools: res output[node_name].get(tool_result, {}) print(f实验结果 - 收率: {res.get(yield)}%, 观察: {res.get(observations)}) elif node_name updater: s output[node_name] print(f状态更新 - 步骤: {s[step]}, 学习: {s[learnings]}, 完成: {s[is_complete]}) final_state output[node_name] print(\n 实验结束 ) print(f最终状态: {final_state})通过这种架构每次agent_node被调用时它接收到的state都是被updater_node精简和增强过的包含了摘要历史和核心学习而不是完整的对话历史。这通常能将每次调用模型的上下文长度减少50%以上显著降低令牌消耗。4. 高级技巧与避坑指南在实际部署中你会遇到比示例更复杂的情况。下面分享一些关键的实操心得和避坑点。4.1 状态设计的权衡记住什么忘记什么状态不是垃圾桶不能什么都往里装。设计时需要反复权衡必要性这个信息对后续决策是否绝对必要实验中的环境湿度可能影响结果但如果你的设备在恒湿环境中这个信息就不必存入状态。压缩性能否用更紧凑的形式表示与其存储“温度从150升到160度”不如存储“当前温度160”和一条学习“升温可能有益”。用learnings列表来存储推导出的规律是压缩信息的关键。时效性有些信息只在短期内有用。可以考虑为状态字段设计“衰减”或“滑动窗口”机制。例如只保留最近10条action_history的详细记录更早的可以归档或只保留统计摘要。注意避免在状态中存储原始的大模型响应文本如完整的Thought。这些内容非常冗长。只提取结构化信息如hypothesis、action、action_input。4.2 提示词Prompt工程引导模型关注状态模型的提示词需要精心设计以教会它如何使用状态。明确指令在Prompt中清晰指出“请基于以下当前状态进行推理无需回顾完整历史。”结构化呈现将state中的字段以清晰、结构化的方式呈现给模型比如使用Markdown的列表或表格帮助模型快速解析。示例Few-shot在Prompt中提供1-2个例子展示如何根据给定的状态摘要生成合理的Thought和Action。这对于让模型适应这种“摘要推理”模式至关重要。强调学习Learnings在Prompt中高亮learnings字段并说明“这些是从以往步骤中总结出的经验请在你的推理中优先考虑这些约束或启发。”4.3 错误处理与鲁棒性自主实验智能体必须能处理意外。工具调用失败在tools_node中要有完善的异常捕获。失败时不应直接崩溃而是将错误信息如{error: 设备未响应}返回并由updater_node将state.error字段置位在条件边中引导至错误处理节点或终止流程。模型输出解析失败使用LangChain的OutputFixingParser或RetryOutputParser来包裹你的解析逻辑让模型有机会自行纠正格式错误的输出。状态污染确保updater_node的逻辑是幂等的并且对异常输入有防御能力。例如在更新数组字段前检查是否为list类型。4.4 令牌效率的量化评估如何证明你的优化有效建立一个简单的评估基准传统方法记录一个无状态ReAct智能体完成相同实验任务所消耗的总令牌数包括所有历史消息。状态化方法记录你的状态化智能体消耗的总令牌数。对比指标单步平均令牌数状态化方法应显著更低。总令牌节省比例(传统令牌数 - 状态化令牌数) / 传统令牌数。状态大小监控监控state字典序列化后的字符串长度确保其增长是可控的、次线性的。在我的一个材料筛选模拟项目中采用状态化设计后单次Agent调用的提示词长度从平均2500令牌下降到了800令牌左右节省了近70%。对于长达50步的实验总成本降低了一半以上。4.5 与LangChain AgentExecutor的对比你可能会问LangChain自带的AgentExecutor不也能跑ReAct吗没错但它本质上是无状态的循环。它的“记忆”体现在将整个对话历史作为上下文。LangGraph 状态化设计的优势在于显式状态管理状态是完全可控、可定制、可持久化的数据结构。你可以轻松地将状态保存到数据库中断后恢复甚至在不同智能体间传递状态。更灵活的工作流你可以轻松地在图中插入额外的节点比如一个专门用于数据可视化的节点或一个在特定条件下触发的“人类审核”节点。更好的可观测性由于状态是结构化的你可以非常方便地监控实验的进展例如绘制参数与收率的关系图或者跟踪learnings的演变过程。5. 常见问题与排查实录在开发和调试状态化ReAct智能体时我遇到了不少坑这里记录下最典型的几个问题和解决思路。问题现象可能原因排查步骤与解决方案智能体陷入循环反复执行相同或无效动作。1.状态学习Learnings未生效Prompt中未强调或模型忽略了learnings。2.探索与利用失衡状态中缺乏对已尝试区域的记录导致重复探索。3.终止条件过严updater_node中is_complete的判断逻辑有误永远无法满足。1. 检查Prompt确保learnings部分被突出显示并加入few-shot示例展示如何使用learning。2. 在状态中增加tried_parameters列表记录所有尝试过的参数组合并在Prompt中提示模型避免重复。3. 在updater_node中增加调试打印检查is_complete的判断逻辑。临时放宽条件看循环是否能结束。令牌节省不明显甚至更多。1.状态设计不当状态中包含了过多冗余信息或者序列化后体积依然很大。2.Prompt过长尽管历史摘要短了但Prompt本身的指令和示例过于冗长。3.模型响应变长由于状态信息不完整模型可能需要生成更长的文本来“补全”推理反而增加了输出令牌。1. 审查状态结构移除所有可推导出的中间字段。使用更短的关键词作为字典键。2. 精简Prompt使用更直接的指令。考虑将复杂的few-shot示例移到微调数据中而不是放在每次调用的Prompt里。3. 在agent_node的解析步骤中强制限制模型输出的Thought部分长度例如在Prompt中要求“用一句话思考”。工具执行结果无法正确更新到状态。1.节点间数据流断裂tools_node的输出格式与updater_node的输入预期不匹配。2.状态更新逻辑错误updater_node中的代码有bug未能正确修改状态字典。1. 使用LangGraph的可视化功能app.get_graph().draw_mermaid()检查图结构确保边连接正确。在每个节点内打印输入/输出验证数据格式。2. 在updater_node中对state的修改要使用new_state state.copy()后再操作避免引用问题。仔细检查数组的append、字典的update等操作。智能体早期表现良好后期决策质量下降。状态信息过载或污染随着实验进行action_history或observation_history不断增长虽然只取最近几条但learnings列表可能积累了矛盾或过时的信息。1. 为learnings实现去重和冲突解决机制。当新增learning与旧learning矛盾时根据其来源例如基于高收率实验的learning权重更高进行取舍。2. 引入学习“有效期”或“置信度”。给每个learning附加一个置信度分数随着反例出现而降低低于阈值则移除。LangGraph图编译或运行时报错。1.状态Schema不匹配StateGraph定义的State类型与节点函数返回的字典结构不一致。2.条件边Conditional Edge返回值错误decide_next_step函数返回的字符串不是图中已定义的节点名。1. 严格使用TypedDict或Pydantic模型定义State并确保所有节点返回的字典都是其子集或兼容。2. 在add_conditional_edges时确保decide_next_step函数返回的所有可能值都出现在你提供的映射字典的键中。最简单的调试方法是在该函数内直接print其返回值。构建一个高效、鲁棒的状态化ReAct智能体是一个迭代过程。我的建议是从一个非常简单的状态和最小可行的工作流开始确保基础的数据流是通的。然后逐步增加复杂性先优化状态设计再精炼Prompt最后处理边界条件和错误。随时用量化的令牌消耗和任务成功率来评估你的每一次改动。记住目标是让智能体“记得牢、想得准、花得少”而这需要你在状态管理、提示工程和系统架构之间找到最佳的平衡点。