Personalized Agent Swarms Analyzer 深度解析：从对话历史到个性化 Mini-Agent 群组的模式提取与生成-平芜编程栈

Personalized Agent Swarms Analyzer 深度解析：从对话历史到个性化 Mini-Agent 群组的模式提取与生成

【免费下载链接】generative-aiSample code and notebooks for Generative AI on Google Cloud, with Gemini Enterprise Agent Platform项目地址: https://gitcode.com/GitHub_Trending/ge/generative-ai

本篇技术指南聚焦generative-ai仓库中agents/personalized-agent-swarms项目的Analyzer 模块（analyzer/README.md）。它是整条「个性化 Mini-Agent 群组」流水线的第二阶段：读取用户会话历史，用 Gemini 3.1 Pro 提取重复出现的意图模式，并为每个模式自动生成带触发器（trigger）、作用域向量（scope embedding）与 Python 代码的 mini-agent。读完本文，你将掌握该模块的完整流水线、核心源码调用链、特征匹配与软惩罚逻辑，以及从命令行运行到产出物解读的端到端实操方法。

Analyzer 在流水线中的定位

personalized-agent-swarms是一条完整的研究型实验流水线：先由harvest/模拟多轮人机对话生成历史（history/{user_id}/），再由Analyzer从历史中提炼模式并生成群组（swarms/{user_id}/），最后由augmented_assistant_agent/在运行时加载这些 mini-agent 提供个性化回答。仓库根目录的 README.md 明确注明这是一套experimental research pipeline（实验性研究管道），并非生产级产品，运行前需要在.env中配置自己的 Google Cloud 项目与模型。

Analyzer 的核心职责可以概括为一句官方描述：Processes conversational history to extract recurring intent patterns and generate personalized mini-agent swarms——处理会话历史，提取重复意图模式，并生成个性化 mini-agent 群组。它向上承接harvest/orchestrator.py产出的会话 JSON，向下为运行时匹配提供triggers.json、user_style.json和agents/*.py。

工作原理：一条 10 步生成流水线

analyzer/README.md 用 10 个步骤概括了从「历史」到「群组」的完整链路，以下逐一结合源码展开：

加载会话日志：从history/{user_id}/读取全部session_*.json。在 analyze_history.py 中，通过sorted(user_history_dir.glob("session_*.json"))按文件名排序加载，若目录不存在或为空则直接返回。
分批送审：每 10 个会话为一批（batch_size=10，见 pattern_extractor.py 的sessions[i:i+batch_size]分片逻辑），调用 Gemini 3.1 Pro 做模式提取。批处理时会把会话做瘦身（仅保留scenario_id、intent、follow_up_strategy、turns、turn_count），以控制上下文长度。
跨批合并去重：每批返回的模式按pattern_name归并，同名的调用Pattern.merge()叠加频次、合并触发词（dict.fromkeys保序去重）、偏好字典做浅合并；复杂度只要任一方为dynamic就升级为dynamic。此外还通过 Jaccard 相似度做跨名合并：当两个模式的 trigger signals 集合相似度> 0.5时，低频模式并入高频模式（pattern_extractor.py）。
频次过滤：只保留出现次数>= 3个会话的模式（min_frequency=3），按频次降序排列。
任务模式与行为模式分离：pattern_type == "task"的是「用户要做什么」（如调试代码、写营销文案），pattern_type == "behavioral"的是「用户怎么沟通」（如消息经常发一半、被长文淹没）。行为模式被聚合进user_style.json作为共享风格画像，而不是生成独立 agent——这是为了避免把沟通风格误判为意图导致误触发。
行为模式 →user_style.json：由 swarm_generator.py 的_generate_user_style合成，输出response_length、language_level、format、tone、follow_up、avoid等键。Prompt 特别强调要用任务模式来校正行为模式的矛盾（比如用户偶尔觉得信息过载，但任务模式显示他要全面深入的输出时，不应写成 "short/concise"）。
任务模式逐个生成三件套：每个任务模式都会生成触发器定义（triggers.json，含结构化属性规则）、Python mini-agent 文件（agents/{pattern_name}.py）、768 维作用域向量（text-embedding-005）。生成时还有两道 V8 加固：并行双候选生成（温度 0.2 与 0.35 各出一份代码，都通过compile()后用 Flash 挑选更 grounded 的一份，swarm_generator.py）和生成后事实核查（_fact_check_agent扫描ENRICHED_PROMPT中的 API 名、CLI flag、URL 等声明并按 HIGH/MEDIUM/LOW 置信度分级，LOW 置信度声明会被改写为带 "verify the exact flag name" 之类的保守措辞）。
Critic/修订轮：用真实历史会话逐 agent 验证，最多 3 轮。每轮做compile()校验、确认execute()是 async、用匹配会话的首条用户消息运行 agent、把输出与历史中的 gold reference 一起交给 critic LLM 打分（0-10 分，10 个维度），不过 7 分就触发独立的 revise LLM 改写。评价与修订使用两次独立 LLM 调用以避免评分偏差；发现 >= 2 处事实矛盾（编造的 API 名、CLI flag、URL、函数签名）时触发捏造硬顶，得分封顶 4/10 并强制修订（swarm_generator.py）。
非参数化质量排名：critic 之后由 Gemini 3.1 Pro 用 5 维评分（VALUE×3、DISTINCTIVENESS×2、TRIGGER_CLARITY×2、QUALITY×2、FREQUENCY×1，满分 50）逐个打分，加权总分在 Python 侧计算，不设固定目标数量，凡 >= 25/50 全部保留；另有独立的 veto 轮剔除冗余或低价值 agent；生成前有 30 个的硬性安全上限（HARD_CAP_AGENTS，见 swarm_generator.py）防止极端情况下的成本失控。LLM 打分整体失败时回退到纯频次过滤（>= 4 个会话）。
覆盖率校验：排名淘汰 agent 后检查被淘汰模式是否仍被幸存者覆盖（swarm_generator.py）。存在缺口时，通过作用域向量余弦相似度找到最近的幸存 agent，把孤儿模式的域并入其规则（domain扩展、冲突exclude_keywords移除）；跨域用户（persona 同时含 software_engineering 与 devops_infrastructure 等）还会扩展最高频 agent 的域列表，最后重算所有作用域向量。

流水线末尾还有一个V8 验证门（validation gate）：对排名后幸存的每个 agent，从历史中选 3 个相似 + 3 个不同场景，用与 Phase 4 相同的多轮评测 harness（模拟用户 + 3 维 judge：accuracy/helpfulness/personalization）实测，满足avg_quality >= 2.5、trigger_rate > 0、false_positive_count <= 1、无捏造才保留（swarm_generator.py）。

核心组件与源码调用链

Analyzer 目录下共 5 个核心文件，职责高度内聚：

文件	职责
analyze_history.py	主入口：解析 CLI 参数、发现用户、串联「提取模式 → 生成群组」
pattern_extractor.py	基于 LLM 的模式聚类：批处理、合并、去重、频次过滤
swarm_generator.py	生成 mini-agent`.py`、触发器、嵌入向量，运行 critic/排名/覆盖率/验证门
trigger_schema.py	特征 schema、提取 Prompt、软惩罚匹配逻辑、规则校验、二元问卷
llm_util.py	对 Google Cloud 调用的重试 + 模型降级包装

入口的调用链非常清晰：analyze_history.main()→asyncio.run(async_main())→ 每个用户执行analyze_user()→extract_patterns()（模式提取）→generate_swarm()（群组生成）。generate_swarm内部按序执行：生成 user_style → 逐个生成 agent 代码 → 生成属性规则 → 生成二元问卷 → 计算作用域向量 → 写入triggers.json/manifest.json→ critic 轮 → 重叠 agent 合并 → 质量排名 → 覆盖率校验 → 验证门 → 最终写盘（swarm_generator.py）。

值得注意的设计细节：触发匹配逻辑在生成期与运行期共享。trigger_schema.py 的文档字符串写明它同时被swarm_generator.py（生成时构建 agent 规则）和active_mem.py（运行时特征提取 + 规则匹配）使用；trigger_matcher.py 则将「消息 → agent」的完整匹配管线抽成共享函数，验证门和运行时走同一套逻辑，确保验证质量与实际评测一致。

匹配逻辑：软惩罚取代硬拒绝

这是 README 单独成节强调的核心设计，值得重点展开。trigger_schema.py定义了完整的特征 schema：

domain 标签（13 个）：software_engineering、devops_infrastructure、data_science_ml、marketing_advertising、business_strategy、finance_accounting、food_beverage、travel_leisure、academic_research、consumer_lifestyle、human_resources、legal_compliance、other
task_type 标签（10 个）：create_generate、debug_troubleshoot、analyze_evaluate、explain_teach、plan_strategize、configure_setup、find_search、format_structure、negotiate_communicate、other
specificity：generic/domain_aware/specialist
output_format：code/structured_text/narrative/calculation/mixed
scope：single_item/comparison/comprehensive/iterative

运行时先用一次 Flash 调用抽取这 5 个枚举字段加topic_keywords（3-8 个关键词）和action_object（动作对象短语），再调用match_agent_rules_with_embedding（trigger_schema.py）做打分。其核心变化是：

domain 不匹配扣 0.15 分、task_type 不匹配扣 0.10 分，而不是一票否决（DOMAIN_MISMATCH_PENALTY = 0.15、TASK_TYPE_MISMATCH_PENALTY = 0.10，见 trigger_schema.py）。这样跨域请求只要语义相似度足够高仍能命中——例如一个 Dockerfile 请求可以匹配到用户的 code-example agent。
specificity 下限仍是硬拒绝（min_specificity），防止噪音；但默认generic，规则生成 Prompt 甚至明确警告「NEVER use specialist」以免误伤真实用户消息。
关键词匹配（require_any_keyword/exclude_keywords）已彻底移除，只保留纯语义匹配；相似度阈值默认 0.45（EMBEDDING_SIMILARITY_THRESHOLD），可由每个用户的_config.similarity_threshold覆盖。

生成规则时同样有validate_rules()（trigger_schema.py）把关，检查 domain/task_type 标签是否在合法集合内、require_any_keyword是否非空列表等，非法规则会打印警告。

运行时两阶段匹配（上下文补充）

README 虽未展开，但匹配逻辑的最终消费端是 trigger_matcher.py 的match_message_to_agent：Stage 1 并发执行特征抽取 + 消息嵌入，对所有 agent 计算带惩罚的余弦相似度并按分排序；Stage 2 对分数最高的至多 5 个候选并行执行各自 3-5 道二元问卷（yes/no，单次 Flash 调用），合成分为40% 嵌入 + 60% 问卷匹配率，问卷匹配率 >= 0.8 的单一候选直接胜出；若问卷无法区分，再升级到 Pro 模型做 tiebreaker（REVIEW_MODEL = "gemini-2.5-pro"）。这套「宽召回 + 精消歧」的层级设计正是 V8 版本误触发率大幅下降的关键。

命令行使用：从预览到全量生成

README 提供了 4 条核心命令，全部由 analyze_history.py 的argparse定义解析。完整用法如下：

source .venv/bin/activate # 分析所有用户（自动发现 history/ 下所有 user_* 目录） python analyzer/analyze_history.py # 单个用户，dry run：只打印模式，不生成任何文件 python analyzer/analyze_history.py --user user_1 --dry-run --verbose # 单个用户完整生成（含 critic 轮） python analyzer/analyze_history.py --user user_1 # 跳过 critic 轮（更快，但质量略低） python analyzer/analyze_history.py --user user_1 --skip-critic

四个 CLI 参数的含义与源码一一对应：

参数	源码位置	作用
`--user USER_ID`	analyze_history.py	只分析指定用户；不传则扫描`history/`下所有`user_*`目录
`--dry-run`	同上`#L109-L111`	提取并打印模式后立即返回，跳过`generate_swarm`
`--skip-critic`	`#L148-L149`	把`skip_critic`传入`generate_swarm`，manifest 中 critic_pass 标记为 skipped
`--verbose`/`-v`	`#L90-L107`	打印每个模式的触发词、典型流程、用户偏好，以及 critic/排名明细

Dry-run 的-v输出会按• pattern_name (freq=N, complexity) [behavioral]: description的格式列出所有模式，并展开 Triggers、Flow、Preferences 三行细节，便于在投入 LLM 成本前先人工审阅模式质量。

运行前提：需要先完成仓库根目录的 README.md 中的安装步骤（uv sync+gcloud auth application-default login），并在.env中配置GOOGLE_GENAI_USE_VERTEXAI=TRUE、GOOGLE_CLOUD_PROJECT、GOOGLE_CLOUD_LOCATION。会话历史必须先由python harvest/orchestrator.py（harvest/orchestrator.py）生成；如果history/下没有任何用户目录，主程序会提示先运行 harvest 并退出（analyze_history.py）。

输出产物：swarms/{user_id}/ 目录结构

README 给出了生成目录的标准结构，运行结束后会在swarms/{user_id}/下产出：

swarms/{user_id}/ ├── manifest.json # 全部 agent 的索引 + 元数据 + critic/排名结果 ├── triggers.json # 触发器规则 + 作用域向量（768 维）+ _config ├── user_style.json # 行为模式画像 └── agents/ ├── pattern_name_1.py # mini-agent，含 async execute() 函数 └── pattern_name_2.py # 数量由质量排名决定，通常 5-10 个

各文件在源码中的写入位置均可追溯：

manifest.json：在 swarm_generator.py 中构建，记录user_id、generated_at、total_sessions_analyzed、agent_count、behavioral_patterns_absorbed、has_user_style以及每个 agent 的name/file/complexity/frequency/trigger_type；critic、merge、ranking、coverage、validation 各阶段结果也会追加写入。
triggers.json：键为 agent 名，值为{trigger_type, rules, description, scope_embedding, binary_questions}；特殊键_config存similarity_threshold: 0.45，可在评测后按用户调优（swarm_generator.py）。
user_style.json：行为模式合成 +_sanitize_user_style净化后的风格画像（净化会剥离「检测消息截断」「逐条确认」「bite-sized」等会让一次性 agent 陷入多轮碎嘴的风格指令，并注入全局覆盖「Always provide a complete, direct, self-contained answer」，见 swarm_generator.py）。
agents/*.py：生成的 mini-agent 是独立可运行的 Python 模块，统一暴露async def execute(user_message: str, llm_client, history=None) -> str。complexity == "static"时核心是AGENT_META+ENRICHED_PROMPT（单次 LLM 调用）；complexity == "dynamic"时则是STEPS列表（最多 3 步：技术域 draft → verify → format，非技术域 draft → synthesize），execute()逐步前馈输出并只返回最后一步结果。生成 Prompt 强制要求调用形式await llm_client.aio.models.generate_content(...)（generate_content_async不存在），并在结尾注入「agent-specific instructions 优先于运行时注入的 STYLE INSTRUCTIONS」等约束。

关于 agent 数量：README 说明「typically 5-10」，但实际是质量驱动而非固定数量——V8 排名阈值 25/50 + veto + 验证门共同决定最终保留数。仓库根 README 的评测结果也印证了这一点：5 个模拟用户各 50 个会话，最终生成 23 个任务 agent（每用户 1-6 个不等）。

模型配置与容错

Analyzer 所有 LLM 调用都经过 llm_util.py 的generate_with_fallback：对主模型最多重试 2 次（30s、60s 退避），429 配额耗尽或 404 模型不可用时自动降级到后备模型，全部失败才抛错。模型清单集中定义在仓库根 config.py，Analyzer 相关的关键配置为：

配置项	默认值	用途
`ANALYSIS_MODEL`	`gemini-3.1-pro-preview`	模式提取、agent 生成、critic、排名
`ANALYSIS_FALLBACKS`	`["gemini-3-pro-preview", "gemini-2.5-pro"]`	分析类调用的降级链
`TRIGGER_MODEL`	`gemini-2.5-flash`	运行时特征抽取 + 问卷评估（低成本高吞吐）
`REVIEW_MODEL`	`gemini-2.5-pro`	问卷无法区分时的 Pro tiebreaker
`EMBED_MODEL`	`text-embedding-005`	作用域向量与消息嵌入（768 维）
`LOCATION`	`global`	所有模型走 global 端点，单一值即可

需要说明的是：config.py中所有 Gemini 3.x 与 text-embedding-005 都依赖 Google Cloud 的模型可用性，具体地区可用性请以你的 GCP 项目实际配额为准；若目标模型不在global端点提供服务，可通过.env的GOOGLE_CLOUD_LOCATION覆盖。

小结：一条可复现的个性化 agent 生产线

Analyzer 模块把「从用户历史到个性化 agent 群组」这一抽象目标落成了一条完全可复现、可审计的工程流水线：批式 LLM 模式提取 → 任务/行为分离 → 结构化触发器 + 语义作用域向量 → 双候选并行生成 + 事实核查 → 多轮 critic → 非参数化质量排名 → 覆盖率校验 → 多轮评测验证门。它在工程上最值得借鉴的三点设计是：软惩罚匹配（用扣分替代硬拒绝，兼顾召回与精度）、生成期与运行期共享同一套匹配逻辑（验证即真实评测）、以及分层质量闸门（编译检查、LLM 打分、向量覆盖率、端到端多轮评测逐层过滤）。对于想要构建「千人千面」式 agent 个性化系统的开发者，这份源码是研究运行时 agent 个性化这一方向的完整范本。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考