ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

从静态画像到动态代码:构建智能体可执行记忆的工程实践

从静态画像到动态代码:构建智能体可执行记忆的工程实践 1. 项目概述从“用户画像”到“可执行记忆”最近在折腾个性化智能体Personalized Agents时我一直在思考一个核心问题我们给智能体提供的“用户信息”到底应该是什么形态是传统的一堆静态标签比如“喜欢科技”、“常驻北京”还是更动态、更可操作的东西直到我反复琢磨“User as Code: Executable Memory for Personalized Agents”这个标题才豁然开朗。这说的不就是把用户从一堆描述性的“数据”变成一段段可以运行、可以推理、可以组合的“代码”吗简单来说“User as Code”是一种构建智能体长期记忆Long-term Memory的新范式。它不再满足于将用户的历史对话、行为记录简单地存储为文本然后靠检索Retrieval来回忆。而是试图将这些记忆“编译”成一种结构化的、蕴含逻辑的、甚至可以直接被智能体“执行”的表示形式。你可以把它想象成不是给智能体一本关于你的、按时间排序的日记传统记忆而是给了它一个关于你的“行为函数库”和“知识图谱推理引擎”可执行记忆。当新场景出现时智能体不是去翻日记找相似片段而是直接“运行”相关的用户代码片段推导出你可能的态度或偏好。这个概念解决了一个关键痛点静态记忆的僵化和低效。比如你告诉智能体“我周三晚上通常要健身所以那个时间段的会议尽量别排”。传统记忆库只是存下这句话。下周三它需要先检索到这条记录再理解“尽量别排”的含义。而如果这条记忆被编码为“可执行”的规则IF (时间是周三晚上) AND (事件类型是会议) THEN (建议优先级低建议操作提议改期)那么智能体在规划日程时可以直接“触发”这条规则无需复杂的语义理解和上下文匹配反应更直接、决策更精准。这篇文章我就结合自己搭建个性化助理的实践拆解一下“User as Code”这个理念如何落地。我会重点聊清楚为什么需要可执行记忆它的核心技术栈是什么如何一步步把零散的用户信息“编译”成代码以及在实际操作中会遇到哪些坑怎么绕过去。无论你是AI产品经理、应用层开发者还是对智能体架构感兴趣的研究者相信这些从一线踩坑得来的经验都能给你带来些实实在在的参考。2. 核心理念与架构设计拆解2.1 从静态数据到动态能力的范式转变要理解“User as Code”首先得看清我们正在告别什么。传统的用户建模无论是早期的用户画像User Profile还是现在基于向量数据库的记忆系统本质都是“静态快照”或“历史档案”。它们擅长回答“用户过去做过什么”、“用户说过喜欢什么”。但当智能体需要主动为用户规划、决策甚至创造时这种静态性就成了瓶颈。举个例子你的智能体知道你喜欢某位导演的电影静态事实。但当它需要为你推荐一部周末放松的电影时仅仅知道这个事实不够。它需要结合“今天是周五晚上”时间上下文、“你刚完成一个高强度项目”实时状态推断、“该导演新片是悬疑片”外部知识以及“你压力大时更爱看轻松喜剧”潜在偏好规则来做出推荐。后两者——状态推断和偏好规则——就是典型的“可执行”知识。它们不是事实而是处理事实的逻辑。“User as Code”的核心主张就是将用户的偏好、习惯、决策逻辑、社交规范等用结构化的、机器可解释甚至可执行的方式表述出来。这种表述形式可以是规则RulesIF-THEN语句明确直接。函数Functions封装了更复杂逻辑的代码块可以接受输入参数。约束Constraints定义用户行为的边界条件。目标Goals与效用函数Utility Functions描述用户想要什么以及如何衡量不同结果的优劣。这样一来用户的“记忆”就变成了智能体“大脑”的一部分扩展程序库。智能体在思考时可以像调用内部API一样调用这些“用户代码”来辅助推理和规划。2.2 可执行记忆系统的核心组件设计基于上述理念一个可执行记忆系统至少需要三层架构第一层记忆采集与原子化这是原料入口。数据源包括显式对话用户直接陈述的偏好、隐式行为点击、停留、购买、跨平台活动日历、邮件、智能家居日志。关键任务是将这些原始数据“原子化”分解为最小可用的信息单元。例如从对话“我讨厌在早上开会”中可以提取原子事实[用户, 厌恶, 早上开会]并尝试抽象为规则模板[时间段, 事件类型] - 情感倾向。第二层记忆编译与代码生成这是核心引擎。它接收原子化的事实运用以下技术生成“代码”模式归纳从多个相似事实中归纳出通用规则。比如从“周一早上不想开会”、“周三早上推迟了会议”等事实中归纳出IF 时间包含“早上” THEN 对会议提议持消极态度。逻辑编程集成将归纳出的规则用形式化逻辑如Prolog风格的逻辑语句、一阶逻辑表示使其可被逻辑推理引擎处理。代码合成对于更复杂的习惯可能需要生成真正的代码片段如Python函数。例如从多次餐厅选择行为中合成一个choose_restaurant(cuisine_list, budget, distance)函数内含用户个性化的选择权重。第三层记忆执行与推理接口这是使用层。为智能体提供统一的API来调用这些记忆代码。例如query_preference(event_detail): 返回用户对该事件的偏好评分。check_constraint(proposed_plan): 检查提议计划是否违反任何用户约束。simulate_user_decision(options): 模拟用户在一组选项中可能做出的选择。这个架构的优势在于解耦和可组合性。记忆的生成、存储和执行相对独立不同的记忆片段代码可以像乐高积木一样被智能体按需组合调用以应对复杂场景。注意设计时要警惕“过度编译”。不是所有用户数据都适合或有必要变成代码。一些具体的、一次性的历史事实如“去年在XX餐厅过生日”用传统向量检索可能更合适。可执行记忆应聚焦于可重复使用的模式、习惯和逻辑。3. 关键技术栈与实现路径3.1 规则提取与逻辑表示从自然语言到可执行代码把用户随口说的话变成严谨的规则是第一步也是最难的一步。直接做自然语言到代码NL2Code的转换目前还不成熟。更可行的路径是分两步走先做信息抽取再做逻辑组装。第一步结构化信息抽取我们利用大语言模型LLM的强大概括和结构化输出能力。不是让LLM直接写代码而是让它按照预定格式提取关键要素。操作示例用户输入“如果邮件来自我的老板而且是在非工作时间收到的请标记为重要但别在晚上10点后提醒我。”给LLM的Prompt“请将以下用户指令解析为结构化规则。请提取触发条件列表、执行动作、例外条件如果有。以JSON格式输出。”期望的LLM输出{ rule_name: boss_after_hours_mail, conditions: [ {field: sender, operator: contains, value: 老板姓名}, {field: arrival_time, operator: not_between, value: [09:00, 18:00]} ], actions: [ {action: tag_email, params: {tag: 重要}}, {action: schedule_notification, params: {delay_until: next_working_day 09:00}} ], exceptions: [ {condition: {field: current_time, operator: , value: 22:00}, override_action: no_notification} ] }这一步将非结构化的自然语言转换成了半结构化的数据。第二步逻辑代码生成有了结构化数据我们就可以用确定的模板或代码生成器将其转化为可执行代码。例如将上面的JSON转化为一段Python函数def rule_boss_after_hours_mail(email, current_time): 处理老板非工作时间邮件的用户规则 # 检查触发条件 if (老板姓名 in email.sender) and (not is_working_hours(email.arrival_time)): # 执行动作 email.add_tag(重要) # 安排通知考虑例外 if current_time 22:00: schedule_notification(email, delay0) else: schedule_notification(email, delay_untilnext_working_day 09:00) return True return False这个过程可以自动化。你可以为不同类型的规则过滤规则、排序规则、提醒规则编写对应的代码模板然后用抽取出的JSON数据去填充模板生成最终的函数。3.2 记忆的存储、索引与版本管理生成的“用户代码”需要被存储和高效检索。这里不能简单用向量数据库因为我们要检索的不是相似文本而是适用的逻辑。存储方案代码仓库化直接使用Git仓库来管理用户的规则函数文件.py或.json。这天然支持版本历史、回滚和diff查看。每个规则一个文件目录按领域如work/,life/,communication/组织。元数据索引为每条规则/函数创建丰富的元数据存入一个轻量级数据库如SQLite或DuckDB。元数据包括rule_id: 规则唯一标识。trigger_fields: 规则依赖的输入字段如sender,time,event_type。这是索引的关键。output_type: 规则输出类型如boolean_filter,priority_score,action_sequence。domain: 适用领域工作、邮件、日历等。confidence: 规则置信度基于推导出该规则的数据量和支持度。created_at/last_triggered: 创建和最后触发时间。检索流程 当智能体遇到一个新情境例如处理一封新邮件它会提取该情境的特征sender“老板”, arrival_time“20:30”, type“邮件”。然后它不去做向量相似度搜索而是去元数据索引中做精确或条件匹配“找出所有trigger_fields包含sender和arrival_time的规则”。这比向量检索更精准、更快且结果直接是可执行的函数。版本管理的重要性 用户的偏好会变。上个月讨厌早上开会可能这个月因为项目紧急就接受了。我们需要管理规则的“生命周期”。实现上可以为每条规则附加一个“活性”分数根据近期是否被触发、用户是否明确否决其推荐等因素动态调整。长期未被使用或经常被覆盖的规则可以自动降级或归档。3.3 与智能体决策循环的集成可执行记忆不是孤立的它必须深度嵌入智能体的“感知-思考-行动”循环中。集成点一在规划阶段提供约束当智能体例如基于ReAct或Plan-and-Execute框架制定一个计划时它首先应该调用get_all_constraints(domain)接口获取所有相关领域的用户约束规则并将它们作为硬性条件或软性优化目标纳入规划器的考量范围。这能确保生成的计划从一开始就是用户可接受的。集成点二在推理阶段提供效用函数当智能体需要在多个备选方案中抉择时例如推荐几个不同的晚餐选项它可以调用simulate_user_choice(options)函数。这个函数内部会运行相关的用户偏好规则为每个选项计算一个“用户效用”分数从而指导智能体做出更个性化的推荐。集成点三在行动后用于验证与学习智能体采取行动后如代用户回复了一封邮件可以将行动结果和用户后续的反馈如用户手动修改了回复作为新的训练数据反馈给“记忆编译层”。如果用户频繁修改智能体的某个固定行为模式系统可以自动触发规则归纳流程修正或生成新的用户代码。实操心得初期集成时建议采用“影子模式”Shadow Mode。即让可执行记忆系统并行运行输出它的决策建议但不实际影响智能体的主流程。将它的建议与智能体原有决策、用户最终行为进行对比分析用以校准规则准确性和集成效果避免“坏规则”直接导致用户体验下降。4. 核心环节实现构建一个最小可行系统4.1 数据采集与原子化处理实战我们从一个具体场景开始构建一个理解用户会议偏好的可执行记忆模块。假设我们有以下几个数据源日历事件标题、时间、参与者、用户接受/拒绝状态。用户与聊天助理的对话记录“我不想把会排在午饭后立刻”。邮件中关于会议安排的沟通。第一步设计原子事实结构我们定义“原子事实”的JSON Schema用于承载从原始数据中提取的核心信息。atomic_fact_schema { type: object, properties: { id: {type: string}, source_type: {type: string, enum: [calendar, chat, email]}, source_id: {type: string}, # 原始数据ID extracted_relation: {type: string}, # 如 prefers_meeting_time, avoids_meeting_with subject: {type: string}, # 通常是 user object: {type: string}, # 如 morning sentiment: {type: number, minimum: -1, maximum: 1}, # 情感倾向-1反对1喜欢 context: {type: object}, # 额外上下文如 {duration_hours: 1.5} timestamp: {type: string, format: date-time} }, required: [source_type, extracted_relation, subject, object] }第二步使用LLM进行批量提取编写一个Prompt引导LLM从原始文本中提取信息并填充上述结构。# 示例Prompt extraction_prompt 你是一个信息提取助手。请从以下用户对话中提取关于会议偏好的结构化事实。 对话记录{conversation_snippet} 请根据以下字段提取信息 - extracted_relation: 关系类型。必须是以下之一prefers_meeting_time, avoids_meeting_time, prefers_meeting_duration, avoids_meeting_with_person, prefers_meeting_day。 - object: 关系的对象。例如如果关系是prefers_meeting_time对象可能是“morning”、“14:00”或“not_after_17:00”。 - sentiment: 用户对此对象的情感倾向。1表示非常喜欢/偏好-1表示非常讨厌/避免0表示中性或复杂。 - context: 其他相关信息如会议时长、是否远程等以JSON对象表示。 请以JSON列表形式输出每个元素是一个原子事实。 运行这个Prompt我们可以从“我不想把会排在午饭后立刻”中得到类似{extracted_relation: avoids_meeting_time, object: immediately_after_lunch, sentiment: -1, context: {}}的原子事实。第三步存储原子事实将所有提取出的原子事实按时间顺序存入一个时序数据库如InfluxDB或简单的文档数据库如Elasticsearch中便于后续按时间和关系类型进行查询分析。4.2 规则归纳与代码生成实战有了上百条关于会议的原子事实后我们就可以开始归纳规则。第一步聚类与模式发现首先我们可以按extracted_relation对事实进行分组。例如所有avoids_meeting_time的事实为一组。然后对同一组内的事实分析其object字段。对象是具体值如“14:00”可能代表用户固定这个时间有空。对象是模糊描述如“morning”,“immediately_after_lunch”需要将其标准化。我们可以建立一个“时间段标签”映射将模糊描述映射到具体的时间区间。例如定义“immediately_after_lunch”为[13:00, 14:00]。这个映射可以手动定义也可以用LLM辅助生成。第二步统计分析与规则生成对于avoids_meeting_time组我们统计每个标准化时间段如[13:00,14:00]出现的次数和平均sentiment值。如果某个时间段出现了显著多次如超过5次且平均情感倾向为负如 -0.5我们就可以考虑生成一条避免规则。 规则生成器是一个预定义的模板def generate_time_avoidance_rule(time_range, confidence): rule_id favoid_meeting_{time_range[0].replace(:, )}_{time_range[1].replace(:, )} code f def {rule_id}(meeting_proposal): \\\用户倾向于避免在{time_range[0]}-{time_range[1]}开会。置信度{confidence:.2f}\\\ proposed_time meeting_proposal.get(start_time) if proposed_time: hour_min proposed_time.hour proposed_time.minute/60 if {time_range[0]} hour_min {time_range[1]}: # 返回一个负分数表示不偏好 return -0.8 * {confidence} # 分数可根据置信度加权 return 0.0 # 不影响 metadata { rule_id: rule_id, trigger_fields: [meeting_proposal.start_time], output_type: preference_score, domain: scheduling, confidence: confidence, conditions: [{field: start_time, operator: between, value: time_range}] } return rule_id, code, metadata第三步规则编译与注册生成的函数代码字符串被写入到文件系统如rules/scheduling/avoid_meeting_1300_1400.py。同时其元数据被注册到前面提到的元数据索引数据库中。这样一个可执行的用户记忆单元就诞生了。4.3 记忆执行引擎的搭建我们需要一个轻量级引擎来加载、管理并执行这些分散的规则文件。引擎核心类设计class UserCodeExecutionEngine: def __init__(self, rules_repo_path, metadata_db_path): self.repo_path rules_repo_path self.metadata_db self._connect_to_metadata_db(metadata_db_path) self.rule_cache {} # 内存缓存已加载的规则函数 def _connect_to_metadata_db(self, db_path): # 连接SQLite等数据库 import sqlite3 return sqlite3.connect(db_path) def get_applicable_rules(self, context): 根据上下文检索适用的规则元数据 # 从context中提取关键字段 trigger_fields extract_fields_from_context(context) query SELECT rule_id, domain, confidence FROM rule_metadata WHERE ? LIKE % || trigger_fields || % AND domain ? ORDER BY confidence DESC # 简化查询实际应根据索引优化 cursor self.metadata_db.execute(query, (str(trigger_fields), context.get(domain))) return cursor.fetchall() def load_and_execute_rule(self, rule_id, context): 加载并执行单个规则 if rule_id not in self.rule_cache: # 动态导入规则模块 module_path os.path.join(self.repo_path, f{rule_id}.py) with open(module_path, r) as f: code f.read() # 注意生产环境需使用更安全的执行沙箱如PyPy的沙箱或Docker隔离 exec_globals {} exec(code, exec_globals) self.rule_cache[rule_id] exec_globals[rule_id] # 假设函数名与rule_id相同 rule_func self.rule_cache[rule_id] return rule_func(context) def evaluate_context(self, context): 评估一个上下文如会议提议 against 所有相关规则 applicable_rules self.get_applicable_rules(context) total_score 0.0 triggered_rules [] for rule_id, domain, conf in applicable_rules: try: score self.load_and_execute_rule(rule_id, context) total_score score if abs(score) 0.01: # 规则被触发 triggered_rules.append((rule_id, score)) except Exception as e: logging.error(f执行规则 {rule_id} 失败: {e}) continue return { aggregated_preference_score: total_score, triggered_rules: triggered_rules, recommendation: accept if total_score 0 else suggest_alternative }这个引擎提供了核心的检索和执行能力。智能体在需要做会议安排决策时只需将会议提议context传给evaluate_context方法即可得到一个基于用户历史“代码”的综合评分和建议。5. 常见陷阱、挑战与优化策略5.1 规则冲突与优先级仲裁随着规则数量增长冲突不可避免。例如规则A说“避免早上开会”规则B说“如果与会者是某重要客户则优先安排”。当为一个重要客户安排早上会议时就产生了冲突。解决方案实现优先级仲裁机制静态优先级为每条规则元数据增加一个priority字段如0-10。数值越高优先级越高。在规则生成时根据规则来源显式用户指令 隐式行为归纳和置信度赋予初始优先级。动态权重规则的最终影响力 priority * confidence * recency_factor。recency_factor基于规则最后被验证或触发的时间让新近产生的规则有更高权重。冲突检测与消解在执行引擎的evaluate_context中如果检测到两条触发规则输出符号相反一正一负且绝对值都较大则触发冲突处理子程序。该子程序可以记录冲突供后续人工或高级AI审核。上下文感知仲裁引入更细粒度的上下文判断。例如检查当前会议是否在“重要客户”列表内如果是则临时调高规则B的权重。向用户询问对于高优先级任务且冲突无法自动解决时让智能体生成一个澄清问题询问用户。5.2 规则过时与概念漂移问题用户的偏好会随时间变化概念漂移。去年喜欢安静今年可能喜欢热闹。陈旧的规则会降低智能体的准确性。建立规则生命周期管理活性监控每条规则被触发时记录触发时间和结果用户最终是否遵从了规则的隐含建议。定义一个“活性衰减”函数例如activity_score base_confidence * exp(-decay_rate * days_since_last_trigger)。有效性验证定期如每周进行“规则审计”。选取近期触发的规则检查在触发后用户的真实反馈。例如规则建议“拒绝此时间段会议”但用户实际接受了会议则此规则的有效性存疑。自动退役与重新训练当规则的activity_score低于阈值或近期有效性验证失败率过高时系统自动将其状态置为deprecated并从主检索索引中移除。同时触发针对该规则相关领域如“会议时间偏好”的重新归纳流程使用最新的原子事实数据生成新的规则候选集。用户显式反馈通道提供快捷方式让用户可以直接对智能体的某个具体决策说“不对”或“以后别这样”。这个反馈应直接关联到触发该决策的底层规则并大幅降低其置信度或直接禁用。5.3 安全、隐私与可控性考量“用户即代码”意味着用户最核心的行为逻辑被外化和程序化了这带来了独特的安全隐私挑战。安全与隐私设计要点本地优先可执行记忆引擎应尽可能在用户设备端如手机、个人电脑运行所有原始数据、原子事实、规则代码都存储在本地。只有经过高度抽象和脱敏的、非个人可识别的模式例如“某类用户普遍存在午饭后效率低谷”在用户明确同意后才可匿名上传用于改进公共模型。代码沙箱执行用户生成的规则代码必须在严格的沙箱环境中进行限制其文件系统访问、网络访问和系统调用能力防止恶意规则或规则生成过程中的漏洞导致安全问题。可解释性与审计追踪智能体的每一个基于用户规则的决策都必须能追溯到是哪些具体规则被触发、贡献了多少分数。这需要执行引擎输出详细的“决策日志”让用户随时可以审查“为什么我的助理会这样建议”。用户拥有最终编辑权系统应提供一个“用户规则面板”以可视化的方式如流程图、自然语言描述展示所有活跃的规则。用户应能随时查看、禁用、编辑或删除任何一条规则。这是确保用户主导权的关键。5.4 性能优化与工程实践当规则数量达到成千上万时检索和执行效率成为瓶颈。性能优化策略索引优化元数据数据库的trigger_fields需要建立高效的索引。可以考虑将字段列表序列化为位图Bitmap进行索引实现快速的集合包含查询。规则分组与预加载根据domain等字段将规则分组。智能体在进入特定领域如“处理邮件”时可以预加载该领域的所有高置信度规则到内存缓存中避免频繁的磁盘I/O。规则编译与JIT对于用Python等解释型语言生成的规则可以考虑在首次加载时使用PyPy的JIT编译器或者将其编译为字节码以加速后续执行。对于性能极度敏感的规则甚至可以用Cython或Rust重写核心逻辑。异步与批量执行对于实时性要求不高的场景如每日日程预排可以异步执行规则评估。对于需要评估大量相似项目如从100个新闻中推荐10条可以设计批量评估接口减少函数调用的开销。工程实践建议版本控制务必使用Git管理规则代码库。每一次规则的新增、更新、退役都应是一个commit便于回滚和追踪变化。单元测试为生成的规则函数编写单元测试。测试用例可以从推导出该规则的原子事实中采样确保规则行为符合预期。监控与告警监控规则引擎的调用延迟、错误率、规则冲突频率等指标。设置告警当规则大量失效或冲突激增时通知开发者进行干预。构建“User as Code”系统是一个持续迭代的过程。它不是一个一劳永逸的解决方案而是一个需要精心设计数据流水线、规则引擎和反馈循环的复杂系统。从最小场景如会议安排开始验证整个流程的可行性再逐步扩展到邮件过滤、内容推荐、购物建议等更多领域是稳妥且有效的推进方式。这个过程中最大的收获可能不是最终那个无所不能的智能体而是通过“编译”用户行为让我们对自己或用户的决策模式有了前所未有的、结构化的深刻理解。
返回列表