企业官网建设流程全解析

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习

导读

MemoryWriteComponent是 openJiuwen agent-core 工作流（Workflow）体系中的一个预置组件，用于将对话消息写入长期记忆（Long-Term Memory）。本文以该组件的官方 API 文档为主体，结合openjiuwen/core/workflow/components/resource/memory_write_comp.py源码实现、openjiuwen/core/memory/long_term_memory.py记忆引擎与tests/unit_tests/core/workflow/test_workflow_with_memory.py单元测试，完整讲解组件的配置参数、输入输出协议、底层调用链与典型接入方式。读完本文，你将能够在自己的工作流图中挂载记忆写入节点，并理解"多用户/多场景数据隔离、自动记忆片段生成、历史上下文引用"等机制是如何落地的。

一、组件定位：工作流中的记忆写入节点

在 openJiuwen 的工作流体系中，openjiuwen.core.workflow.components.resource.memory_write_comp模块提供了将对话消息持久化到长期记忆的可组合组件。它封装了MemoryWriteExecutable，供工作流图（Graph）直接使用，核心职责是把一组BaseMessage消息写入长期记忆引擎。

组件位于工作流预置组件目录之下。根据 components.README.md 的说明，openjiuwen.core.workflow.components提供了一批与工作流基类配套使用的预置组件，包括memory_retrieval_comp（记忆检索）、memory_write_comp（记忆写入）、knowledge_retrieval_comp（知识库检索）、llm_comp（模型调用）等。组件类统一通过openjiuwen.core.workflow导出，官方推荐使用from openjiuwen.core.workflow import ...方式导入。

从源码结构看，模块内共包含四个核心类型（见 memory_write_comp.py）：

其中MemoryWriteComponent继承自ComponentComposable，MemoryWriteExecutable继承自ComponentExecutable，二者分别对应"图上的可组合单元"与"实际执行单元"两层抽象（基类定义见 component.py）。

二、配置参数详解：MemoryWriteCompConfig

MemoryWriteCompConfig是用于配置长期记忆写入参数的数据类，继承ComponentConfig，定义如下：

@dataclass(kw_only=True, config=ConfigDict(arbitrary_types_allowed=True)) class MemoryWriteCompConfig(ComponentConfig): memory: LongTermMemory scope_id: str = LongTermMemory.DEFAULT_VALUE user_id: str = LongTermMemory.DEFAULT_VALUE session_id: str = LongTermMemory.DEFAULT_VALUE agent_config: AgentMemoryConfig = field(default_factory=AgentMemoryConfig) gen_mem: bool = field(default=True) gen_mem_with_history_msg_num: int = field(default=2)

各参数说明如下：

2.1 数据隔离三要素：scope_id / user_id / session_id

LongTermMemory.DEFAULT_VALUE在记忆引擎源码中定义为常量"__default__"（见 long_term_memory.py）。这三个 ID 构成了记忆数据的三层隔离维度：

• scope_id：按应用场景隔离。例如同一套系统为"客服助手"与"编程助手"两个场景分别维护不同的记忆库，互不串扰；
• user_id：按最终用户隔离，保证不同用户之间的记忆互不可见；
• session_id：关联到具体会话，用于溯源当前写入消息所属的对话上下文。

写入时，这三个 ID 会原样透传给记忆引擎的add_messages方法，配合用户级分布式锁（DistributedLock(self.kv_store, f"user/{user_id}")）确保同一用户并发写入时数据一致（见 long_term_memory.py）。

2.2 agent_config：控制记忆生成行为

agent_config的类型是AgentMemoryConfig，定义于 config.py，其关键字段如下：

这些开关直接决定add_messages内部通过Generator.gen_all_memory生成哪些类型的记忆单元。结合记忆引擎定义（long_term_memory.py），写入一次对话后可能产出的记忆类型包括：用户变量（variables）、用户画像（user_profile）、语义记忆（semantic_memory）、事件记忆（episodic_memory）以及对话摘要（summary）。

2.3 gen_mem 与 gen_mem_with_history_msg_num

• gen_mem：为True时，写入消息后由 LLM 自动抽取并生成记忆片段；为False时，仅把消息本身持久化到消息存储，不做记忆抽取。记忆引擎在add_messages中对gen_mem=False会直接返回空的AddMemResult()（见 long_term_memory.py），适用于只需要"留档"不需要"提炼"的场景。
• gen_mem_with_history_msg_num：控制生成记忆时引用的历史消息条数。引擎会通过_get_history_messages(..., history_window_size=gen_mem_with_history_msg_num)拉取该数量的历史消息作为生成记忆的上下文（见 long_term_memory.py）。

三、组件 API：构造与方法

3.1 构造

MemoryWriteComponent(component_config: Optional[MemoryWriteCompConfig] = None)

• component_config：可选的MemoryWriteCompConfig组件配置。传入None时组件仍可构造，但在执行前必须确保配置完整（尤其是memory实例）。

3.2 add_component：挂载到工作流图

add_component(graph: Graph, node_id: str, wait_for_all: bool = False) -> None

将本组件作为节点添加到工作流图中。源码实现为graph.add_node(node_id, self.to_executable(), wait_for_all=wait_for_all)（见 memory_write_comp.py）。wait_for_all=True时该节点会等待所有上游节点完成后才执行，适用于多路输入汇聚后再统一写入记忆的场景。

3.3 to_executable：转换为可执行单元

to_executable() -> MemoryWriteExecutable

将可组合组件转换为对应的可执行对象MemoryWriteExecutable，其内部持有_config与_memory引用（见 memory_write_comp.py）。工作流运行时会调用可执行对象的invoke方法完成写入。

四、输入与输出协议

4.1 输入（MemoryWriteInput）

注意：messages列表必须非空，否则会抛出参数校验错误。

源码中MemoryWriteInput继承pydantic.BaseModel，timestamp通过Field(default=None)声明可选，并允许extra="allow"的宽松模式（见 memory_write_comp.py）。

4.2 输出（MemoryWriteOutput）

写入成功时组件返回{"success": True}。若add_messages内部抛异常，组件不会返回失败结果，而是直接抛出封装后的错误（见下文错误处理章节）。

4.3 时间戳语义

timestamp缺省时，add_messages内部使用datetime.now(timezone.utc)作为当前时间，并在写入前统一转换为 UTC 时区（timestamp.astimezone(timezone.utc)）（见 long_term_memory.py）。当一条messages列表含多条消息时，引擎会为每条消息依次递增timedelta(milliseconds=i)生成毫秒级递增的时间戳，保证同一批次消息的时间顺序（见 long_term_memory.py）。

五、执行流程与底层实现原理

MemoryWriteExecutable.invoke是组件的核心执行方法（见 memory_write_comp.py），执行流程如下：

1. 绑定会话：_set_session(session)将当前执行会话绑定到组件；
2. 输入校验：validate_inputs(inputs)通过MemoryWriteInput.model_validate(inputs)校验输入，校验失败抛出COMPONENT_MEMORY_WRITE_INPUT_PARAM_ERROR；
3. 空列表检查：messages为空时抛出COMPONENT_MEMORY_WRITE_INPUT_PARAM_ERROR，错误信息为"Messages list cannot be empty"；
4. 开始日志：记录工作流组件开始事件（LogEventType.WORKFLOW_COMPONENT_START），附带message_count、scope_id、user_id、gen_mem、sensitive_mode等元数据；
5. 调用记忆引擎：await self._memory.add_messages(messages, agent_config, user_id, scope_id, session_id, timestamp, gen_mem, gen_mem_with_history_msg_num)；
6. 异常包装：引擎调用失败时抛出COMPONENT_MEMORY_WRITE_INVOKE_CALL_FAILED，错误信息为"Memory write call failed: {e}"，并以原始异常作为cause链式抛出；
7. 成功返回：返回{"success": True}，并记录完成事件日志。

5.1 底层调用链：add_messages

组件写入动作最终落到LongTermMemory.add_messages（见 long_term_memory.py），其内部关键步骤包括：

1. 校验scope_id合法性（_validate_id）；
2. 获取 scope 级 LLM（_get_scope_llm）与 scope 配置（_get_scope_config），应用 scope 级 embedding；
3. 获取用户级分布式锁，串行化同一用户的内存写入；
4. 按gen_mem_with_history_msg_num拉取历史消息作为生成上下文；
5. 将消息逐条持久化到消息管理器（MessageManager），记录user_id/scope_id映射；
6. 若gen_mem=True，调用Generator.gen_all_memory抽取变量、画像、语义记忆、事件记忆与摘要，再经WriteManager.add_memories写入对应存储后端；
7. 返回AddMemResult，按记忆类型分别汇总variables、user_profile、semantic_memory、episodic_memory、summary。

整个引擎支持 KV 存储、向量存储、数据库存储与消息存储四种后端（register_store中注册，见 long_term_memory.py），并可在注册时自动挂载SimpleMemoryIndex语义索引。

六、错误处理与状态码

组件涉及两类错误码，定义于 codes.py：

所有错误统一通过build_error构造，保留原始异常链（cause），便于排查底层失败原因。日志层面，开始/结束/失败分别使用WORKFLOW_COMPONENT_START、WORKFLOW_COMPONENT_END、WORKFLOW_COMPONENT_ERROR三类事件类型。

七、单元测试：行为契约验证

组件行为由 test_workflow_with_memory.py 中的TestMemoryWriteComponent覆盖，测试用例即组件行为契约：

• test_memory_write_success：构造MemoryWriteCompConfig(memory=mock_memory, scope_id="test_scope", user_id="test_user", session_id="test_session", gen_mem=True)，输入UserMessage("Hello")与AssistantMessage("Hi there!")，断言返回success=True且add_messages被调用；
• test_memory_write_with_timestamp：传入自定义datetime.now(tz=timezone.utc)时间戳，断言该时间戳被原样透传给add_messages的timestamp参数；
• test_memory_write_empty_messages_error：输入{"messages": []}，断言抛出COMPONENT_MEMORY_WRITE_INPUT_PARAM_ERROR且错误信息包含"Messages list cannot be empty"；
• test_memory_write_missing_messages_error：直接调用MemoryWriteExecutable.validate_inputs({})，断言缺失messages字段同样触发COMPONENT_MEMORY_WRITE_INPUT_PARAM_ERROR；
• test_memory_write_invoke_call_failed：mockadd_messages抛出Exception("DB connection failed")，断言抛出COMPONENT_MEMORY_WRITE_INVOKE_CALL_FAILED且错误信息包含"Memory write call failed"。

这些用例同时展示了组件的标准用法：config = MemoryWriteCompConfig(memory=...)→component = MemoryWriteComponent(config)→executable = component.to_executable()→await executable.invoke(inputs, fake_session, context)。

八、在工作流图中接入 MemoryWriteComponent

将记忆写入组件接入工作流的标准流程如下（参考组件挂载 API 与 预置组件使用指南 中的工作流构建模式）：

from openjiuwen.core.memory.long_term_memory import LongTermMemory from openjiuwen.core.workflow import Workflow, Start, End from openjiuwen.core.workflow.components.resource.memory_write_comp import ( MemoryWriteComponent, MemoryWriteCompConfig, ) # 1. 准备长期记忆实例（需先完成 store 注册与配置，见 LongTermMemory.register_store / set_config） memory = LongTermMemory() # 2. 配置记忆写入组件 write_config = MemoryWriteCompConfig( memory=memory, scope_id="customer_service", user_id="user_001", session_id="session_001", gen_mem=True, gen_mem_with_history_msg_num=2, ) mem_write = MemoryWriteComponent(write_config) # 3. 构建工作流图 workflow = Workflow() workflow.set_start_comp("start", Start(), inputs_schema={"query": "${user_inputs.query}"}) # 4. 将记忆写入组件挂载为节点，并建立边 mem_write.add_component(workflow, node_id="mem_write", wait_for_all=False) workflow.add_connection("start", "mem_write") end = End() workflow.set_end_comp("end", end) workflow.add_connection("mem_write", "end")

运行时，上游节点（如 LLM 组件产出的对话消息）通过inputs_schema把消息列表注入mem_write节点；节点执行时校验messages非空，调用add_messages完成持久化与记忆抽取，随后将{"success": True}传递到下游。

关于LongTermMemory实例的初始化，需要在调用前通过register_store(kv_store=..., vector_store=..., db_store=..., embedding_model=..., message_store=...)注册存储后端并调用set_config(MemoryEngineConfig())完成记忆引擎装配（见 long_term_memory.py）；未注册kv_store时引擎会直接报错，未提供db_store时 scope-user 映射将退化为 KV 后端，未启用消息管理器时历史上下文与来源追踪不可用（引擎会给出相应告警日志）。

九、最佳实践与注意事项

1. messages非空校验是硬性约束：组件在invoke入口与MemoryWriteInput校验两层都会拦截空列表，务必在装配上游节点时保证消息来源不为空。
2. 多租户隔离务必显式配置 ID：scope_id/user_id缺省值均为"__default__"，多场景、多用户部署时若不显式指定，所有数据将写入同一隔离域，建议在配置阶段统一传入。
3. gen_mem按需开启：高频消息流水（如日志记录类场景）可设gen_mem=False仅做消息落库，避免不必要的 LLM 抽取开销；需要"记住用户偏好/事实"的对话场景则保持True。
4. gen_mem_with_history_msg_num平衡上下文与成本：数值越大生成记忆时参考的历史越多、抽取质量越高，但 LLM 调用成本与延迟也随之上升，默认值 2 是兼顾质量与开销的起点。
5. 错误信息保留原始异常链：通过cause字段可追溯到底是参数问题（INPUT_PARAM_ERROR）还是引擎调用失败（INVOKE_CALL_FAILED），排查问题时优先查看组件日志中的WORKFLOW_COMPONENT_ERROR事件。
6. 组件与记忆引擎的生命周期：memory实例可在多个组件间复用（引擎本身为单例设计），同一LongTermMemory可同时服务于MemoryWriteComponent与 MemoryRetrievalComponent，形成"写入—检索"闭环，为智能体提供跨会话的长期记忆能力。

十、相关资源

• 组件源码：memory_write_comp.py
• 记忆引擎实现：long_term_memory.py
• 记忆配置模型：config.py
• 组件基类：component.py
• 单元测试：test_workflow_with_memory.py
• 错误码定义：codes.py
• 组件模块总览：components.README.md
• 记忆检索组件（配套使用）：memory_retrieval_comp.md
• 工作流预置组件使用指南：Using Preset Components.md

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习