多步任务不能只靠下一次提示记住上一轮发生的事情。系统需要明确哪些数据属于当前任务，哪些数据可以跨会话保留，以及多个处理角色如何交换结果。以记忆方案咨询为例，一个角色解释保存方式，另一个角色检查数据丢失风险，最终形成统一答复。
目标、状态、执行边界和验收标准
| 项目 | 定义 |
|---|---|
| 目标 | 根据用户偏好整合知识与风险两类结果，同一会话保留历史，新会话保留用户级偏好。 |
| 状态 | messages保存会话消息；preference保存本次加载的偏好；research保存各角色成果；answer保存统一结果。 |
| 执行边界 | 用户身份由服务端提供；用户级数据使用独立namespace；每次查询固定两个独立角色；进程内存只支持教学环境。 |
| 验收标准 | 同线程消息累积；新线程历史独立；同用户新线程能取得偏好；其他用户不会取得该偏好；两类成果均出现在答案中。 |
方法与设计思想
状态管理：把事实放在明确字段里
StateGraph节点接收已有状态并返回更新字段。状态应保存任务事实和中间结果，避免把所有信息混在长段对话里。字段含义明确以后，分支才知道自己该读取和修改什么。
messages使用add_messages合并器处理消息，其余字段按当前节点结果更新。并发角色在RunnableParallel内部返回各自命名的成果，写作节点只在两项都完成后执行。
短期与长期记忆：区分线程和用户
checkpointer通过thread_id关联同一会话的状态；store通过用户namespace组织跨会话数据。同一个用户可以有多个线程，偏好相同但消息历史不同。
代码中的InMemorySaver和InMemoryStore都在进程内保存数据。它们演示两种作用范围，进程退出以后数据会消失。需要持久保存时，应替换为数据库实现，并验证重启后的恢复。
协作：先明确职责，再增加角色
knowledge角色负责保存方式，risk角色负责丢失风险，write节点负责整合。这里是两个职责独立的Runnable组成的协作工作流；固定实现不具备自主对话和动态委派能力。
如果真实任务需要模型专家，可以把各角色替换为各自的提示链或create_agent实例，仍保留统一结果契约。只有角色差异确实带来增益时，才值得增加调用和通信成本。
A2A：把远程协作当成独立边界
A2A适用于不同系统之间的能力发现与任务交换。当前流程在同一进程运行，因此无需加入远程协议。把Python函数调用改名为A2A并不会产生协议兼容性。需要跨服务时，应另外实现任务ID、状态查询、超时和产物传递，并做跨系统互操作验证。
为什么加入这些机制
| 机制 | 解决的问题 | 验证方式 |
|---|---|---|
| StateGraph | 明确各阶段的输入输出和推进顺序 | 检查节点更新与最终状态字段 |
| checkpointer | 关联同一线程的多次执行 | 第二次调用后消息数增加，新线程保持独立 |
| 用户级store | 跨会话取得偏好并隔离用户 | 同用户换线程仍有偏好，其他用户没有 |
| 角色协作 | 分别检查知识和风险 | 任一角色结果缺失时不得宣称完整交付 |
LangChain与LangGraph代码实现
主代码在Python 3.12.14、LangChain 1.4.3、langchain-core 1.6.6、LangGraph 1.2.12、Pydantic 2.13.5下运行验证。建议使用独立虚拟环境：
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install langchain==1.4.3 langchain-core==1.6.6 langgraph==1.2.12 pydantic==2.13.5
将下面完整代码保存为flow03.py，运行python flow03.py。默认模式不需要模型凭据，使用固定数据验证流程和异常分支。
import json
from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableLambda, RunnableParallel
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
class State(TypedDict, total=False):
user_id: str
question: str
messages: Annotated[list, add_messages]
preference: str
research: dict
answer: str
status: str
store = InMemoryStore()
# user_id由服务端登录身份提供，不能由模型替换。
store.put(("users", "u1"), "style", {"value": "先给结论，再解释"})
specialists = RunnableParallel({
"knowledge": RunnableLambda(lambda q: {
"id": "d1", "text": "长期记忆需要跨会话存储。", "question": q,
}),
"risk": RunnableLambda(lambda q: {
"id": "d2", "text": "内存存储在进程退出后消失。", "question": q,
}),
})
def load_memory(state: State) -> dict:
item = store.get(("users", state["user_id"]), "style")
return {"preference": item.value["value"] if item else "简洁",
"messages": [HumanMessage(content=state["question"])]}
def research(state: State) -> dict:
return {"research": specialists.invoke(state["question"])}
def write(state: State) -> dict:
parts = state["research"]
answer = f"风格：{state['preference']}\n"
answer += "\n".join(f"{p['id']}: {p['text']}" for p in parts.values())
return {"answer": answer, "status": "completed"}
def build():
graph = StateGraph(State)
for name, fn in [("memory", load_memory), ("research", research), ("write", write)]:
graph.add_node(name, fn)
graph.add_edge(START, "memory")
graph.add_edge("memory", "research")
graph.add_edge("research", "write")
graph.add_edge("write", END)
return graph.compile(checkpointer=InMemorySaver(), store=store)
if __name__ == "__main__":
app = build()
config = {"configurable": {"thread_id": "u1-session1"}}
first = app.invoke({"user_id": "u1", "question": "如何保存记忆？"}, config)
second = app.invoke({"user_id": "u1", "question": "哪些信息会丢失？"}, config)
third = app.invoke({"user_id": "u1", "question": "新会话"},
{"configurable": {"thread_id": "u1-session2"}})
other = app.invoke({"user_id": "u2", "question": "新用户"},
{"configurable": {"thread_id": "u2-session1"}})
assert len(first["messages"]) == 1
assert len(second["messages"]) == 2
assert len(third["messages"]) == 1
assert third["preference"] == first["preference"]
assert other["preference"] == "简洁"
assert set(first["research"]) == {"knowledge", "risk"}
print(json.dumps({"answer": first["answer"],
"same_session_messages": len(second["messages"]),
"new_session_messages": len(third["messages"]),
"other_user_style": other["preference"]}, ensure_ascii=False, indent=2))
load_memory先从用户namespace加载偏好，再追加当前问题。research执行两个独立角色；write合并成果。build同时配置checkpointer和store，但两者并不会自动拥有相同的数据语义。
验收与验证结果
脚本执行四次调用，覆盖同线程、同用户新线程和另一用户。断言检查消息条数和偏好隔离。服务端还应验证登录用户是否有权访问指定thread_id；示例中的固定身份并不代表已经实现认证。
Google ADK实现与LangChain对照
Google ADK版本使用SessionService管理会话与事件，user:style保存用户作用范围的偏好。两个角色分别写knowledge和risk，SequentialAgent在并行完成后汇总。用户级字段用于固定偏好，不等同于可检索的MemoryService。
| 比较项 | LangChain／LangGraph | Google ADK |
|---|---|---|
| 会话标识 | thread_id连接图状态 | app_name、user_id与session_id共同定位Session |
| 跨会话偏好 | store按用户namespace保存 | SessionService维护user:前缀字段 |
| 更新提交 | 节点返回状态更新 | 代理yield携带state_delta的Event，由Runner处理 |
| 长期检索 | 需要另外构造store查询或检索器 | 需要MemoryService；示例只验证用户偏好作用范围 |
Google ADK完整代码
对照代码已在google-adk==2.11.0与Python 3.12环境运行。安装python -m pip install google-adk==2.11.0，保存为adk03.py后运行。默认不调用真实模型。
import asyncio
import inspect
import json
from collections.abc import Callable
from typing import Any
from google.adk.agents import BaseAgent, SequentialAgent, ParallelAgent, LoopAgent
from google.adk.events import Event, EventActions
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
class Step(BaseAgent):
action: Callable[..., Any]
async def _run_async_impl(self, ctx):
delta = self.action(ctx.session.state)
if inspect.isawaitable(delta):
delta = await delta
yield Event(author=self.name, actions=EventActions(state_delta=delta))
async def run_state(agent, state, session_id="s1", user_id="u1", service=None):
service = service or InMemorySessionService()
await service.create_session(app_name="demo", user_id=user_id,
session_id=session_id, state=state)
runner = Runner(agent=agent, app_name="demo", session_service=service)
async for _ in runner.run_async(user_id=user_id, session_id=session_id,
new_message=types.Content(role="user", parts=[types.Part(text="执行任务")])):
pass
session = await service.get_session(app_name="demo", user_id=user_id,
session_id=session_id)
return dict(session.state)
def load(state):
return {"preference": state.get("user:style", "简洁")}
def merge(state):
return {"answer": state["knowledge"] + state["risk"], "status": "completed"}
def build():
return SequentialAgent(name="team", sub_agents=[
Step(name="load", action=load),
ParallelAgent(name="specialists", sub_agents=[
Step(name="knowledge", action=lambda _: {"knowledge": "跨会话信息需要外部存储。"}),
Step(name="risk", action=lambda _: {"risk": "内存服务在退出后丢失。"}),
]),
Step(name="merge", action=merge),
])
async def main():
service = InMemorySessionService()
first = await run_state(build(), {"user:style": "先给结论"}, "s1", "u1", service)
same_user = await run_state(build(), {}, "s2", "u1", service)
other = await run_state(build(), {}, "s3", "u2", service)
assert first["preference"] == same_user["preference"] == "先给结论"
assert other["preference"] == "简洁"
session1 = await service.get_session(app_name="demo", user_id="u1", session_id="s1")
session2 = await service.get_session(app_name="demo", user_id="u1", session_id="s2")
assert session1.id != session2.id and len(session1.events) > 0
print(json.dumps({"same_user_style": same_user["preference"],
"other_user_style": other["preference"],
"answer": first["answer"]}, ensure_ascii=False, indent=2))
if __name__ == "__main__":
asyncio.run(main())
ADK测试确认同用户新会话能取得偏好，其他用户不能取得。会话事件分别保存在各Session。两种内存服务都无法据此保证进程重启后的数据保留。
接口与版本边界
对照采用SequentialAgent、ParallelAgent、LoopAgent或BaseAgent模板接口，以便比较相同控制思想。ADK 2.x官方文档已引入图工作流和动态工作流，新项目应评估对应机制；这里的版本运行通过不代表模板API是所有新项目的首选。两种框架的运行结果不能直接等同于真实模型效果。
扩展边界
这是进程内的状态与协作示例，未验证重启恢复、数据库一致性、租户认证或A2A通信。长期作用范围与磁盘持久化是不同概念，生产系统需要两者同时满足。