APS 文件系统代理技能

此技能教会 APS 调度代理如何导航、查询和维护基于本地文件系统的知识库。文件系统是所有领域规则、客户记忆和问题模型的唯一可信来源。上层设有向量索引以支持语义检索，同时使用 Git 记录每一次变更，确保可审计性。

知识库结构布局

aps_knowledge_base/
├── .git/                     ← 版本历史记录，切勿手动修改
├── domain_rules/             ← 从对话中提取的 APS 规则
│   ├── _index.json           ← 主规则注册表（始终需更新）
│   ├── machine_rules/
│   ├── operator_rules/
│   └── material_rules/
├── client_memory/            ← 客户的持久化理解
│   ├── _profile.json         ← 车间布局 + 计划流程 + 偏好设置
│   ├── shop_floor/
│   ├── planning_process/
│   └── decision_history/     ← 每次调度会话一个文件
├── problem_schemas/          ← 按问题类型划分的建模模板
├── solver_configs/           ← 求解器参数与路由阈值
├── pending_review/           ← 待人工审批的新知识提案
└── logs/
    ├── decisions/            ← 调度决策的审计日志
    └── knowledge_changes/    ← 知识写入操作的审计日志

在执行任何操作前，请确认知识库根目录存在：

ls aps_knowledge_base/ 2>/dev/null || echo "知识库未初始化"

若尚未存在，请进行初始化（参见下方“初始化新知识库”部分）。

读取知识

加载客户配置文件

始终优先加载客户配置文件——它定义了车间拓扑、计划流程及输出偏好，这些将影响后续所有决策。

import json, pathlib

kb = pathlib.Path("aps_knowledge_base")
profile = json.loads((kb / "client_memory/_profile.json").read_text())
shop = profile["shop_floor"]        # 类型、工序阶段、每阶段设备数量等
prefs = profile["preferences"]      # 主要目标、输出格式等

使用语义检索获取规则（推荐方式）

当你知道需要什么内容但不确定具体在哪一个文件时，应使用语义搜索。前提是必须已构建向量索引（参见“维护向量索引”部分）。

import chromadb

client = chromadb.PersistentClient(path="aps_knowledge_base/.chromadb")
collection = client.get_collection("domain_rules")

results = collection.query(
    query_texts=["操作员 HSE 认证与设备维护"],
    n_results=5,
    where={"status": "active"}          # 仅检索激活状态的规则
)

# results["ids"], results["documents"], results["metadatas"]
for doc, meta in zip(results["documents"][0], results["metadatas"][0]):
    print(f"[{meta['rule_id']}] {meta['name']}: {doc}")

根据规则 ID 直接查找规则

当已知规则 ID（例如来自决策日志）时使用：

rule_path = kb / f"domain_rules/{category}/{rule_id}.json"
rule = json.loads(rule_path.read_text())

加载当前调度会话所需的全部有效规则

将最相关的 Top-K 条规则注入调度上下文：

def get_relevant_rules(query: str, top_k: int = 5) -> list[dict]:
    collection = client.get_collection("domain_rules")
    results = collection.query(
        query_texts=[query],
        n_results=top_k,
        where={"status": "active"}
    )
    rules = []
    for rule_id, meta in zip(results["ids"][0], results["metadatas"][0]):
        path = kb / meta["file_path"]
        rules.append(json.loads(path.read_text()))
    return rules

加载问题模板

problem_type = "flow_shop"   # 或 job_shop, rcpsp, re_entrant
schema = json.loads((kb / f"problem_schemas/{problem_type}.json").read_text())

读取会话决策历史

history_dir = kb / "client_memory/decision_history"
sessions = sorted(history_dir.glob("session_*.json"), reverse=True)
last_session = json.loads(sessions[0].read_text()) if sessions else {}

提出新知识（写入路径）

代理绝不能直接写入主知识目录。 所有新知识必须先放入 pending_review/，经人工确认后方可生效。

提出新的 APS 规则

每当从对话中提取出新的约束或规则时调用此函数：

import json, pathlib, datetime

def propose_rule(rule_content: dict, source_quote: str, session_id: str):
    kb = pathlib.Path("aps_knowledge_base")
    pending = kb / "pending_review"
    pending.mkdir(exist_ok=True)

    ts = datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S")
    proposal = {
        **rule_content,
        "status": "proposed",
        "metadata": {
            **rule_content.get("metadata", {}),
            "created_at": datetime.datetime.utcnow().isoformat() + "Z",
            "created_by": "ai_agent",
            "confirmed_by": None,
            "source_session": session_id,
            "source_quote": source_quote,
            "use_count": 0,
            "confidence": 0.9
        }
    }

    out_path = pending / f"proposed_{rule_content['id']}_{ts}.json"
    out_path.write_text(json.dumps(proposal, ensure_ascii=False, indent=2))

    # 返回摘要供用户确认
    return {
        "proposal_file": str(out_path),
        "rule_id": rule_content["id"],
        "name": rule_content["name"],
        "description": rule_content["description"]
    }

调用该函数后，必须将提案展示给用户并等待明确确认，再继续下一步。格式如下：

建议将以下内容加入知识库：

规则ID: {rule_id}
名称: {name}
描述: {description}
来源: "{source_quote}"

[确认入库] [修改后入库] [忽略本次]

在收到用户明确确认前，不得执行 confirm_proposal()。

def propose_memory_update(memory_type: str, updates: dict, reason: str):
    """
    memory_type: 'shop_floor' | 'planning_process' | 'preferences'
    """
    pending = kb / "pending_review"
    ts = datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S")
    proposal = {
        "type": "client_memory_update",
        "memory_type": memory_type,
        "updates": updates,
        "reason": reason,
        "proposed_at": datetime.datetime.utcnow().isoformat() + "Z"
    }
    out_path = pending / f"proposed_memory_{memory_type}_{ts}.json"
    out_path.write_text(json.dumps(proposal, ensure_ascii=False, indent=2))
    return str(out_path)

确认提案（用户确认后执行）

仅在用户通过聊天明确确认后调用以下函数。

def confirm_proposal(proposal_file: str, confirmed_by: str):
    """将提案从 pending_review 移动到实时知识库中。"""
    kb = pathlib.Path("aps_knowledge_base")
    proposal_path = pathlib.Path(proposal_file)
    proposal = json.loads(proposal_path.read_text())

    if proposal.get("type") == "client_memory_update":
        _apply_memory_update(proposal, confirmed_by)
    else:
        _apply_rule(proposal, confirmed_by)

    # 从待审列表中移除
    proposal_path.unlink()

    # 更新向量索引并提交 Git
    _update_vector_index(proposal)
    _git_commit(proposal, confirmed_by)

def _apply_rule(proposal: dict, confirmed_by: str):
    rule_type = proposal.get("type", "general")
    category_map = {
        "machine_constraint": "machine_rules",
        "operator_constraint": "operator_rules",
        "material_constraint": "material_rules",
    }
    subdir = category_map.get(rule_type, "machine_rules")
    dest = kb / f"domain_rules/{subdir}/{proposal['id']}.json"
    dest.parent.mkdir(parents=True, exist_ok=True)

    proposal["status"] = "active"
    proposal["metadata"]["confirmed_by"] = confirmed_by
    proposal["metadata"]["confirmed_at"] = (
        datetime.datetime.utcnow().isoformat() + "Z"
    )
    dest.write_text(json.dumps(proposal, ensure_ascii=False, indent=2))

    # 刷新规则索引文件
    _refresh_rule_index()

def _apply_memory_update(proposal: dict, confirmed_by: str):
    profile_path = kb / "client_memory/_profile.json"
    profile = json.loads(profile_path.read_text())
    memory_type = proposal["memory_type"]

    if memory_type not in profile:
        profile[memory_type] = {}
    profile[memory_type].update(proposal["updates"])
    profile["last_updated"] = datetime.datetime.utcnow().isoformat() + "Z"

    profile_path.write_text(json.dumps(profile, ensure_ascii=False, indent=2))

维护向量索引

向量索引必须与文件系统保持同步。每当规则被添加、更新或停用时，都需重新构建。

增量更新（单个规则变更后）

def _update_vector_index(rule: dict):
    import chromadb
    client = chromadb.PersistentClient(path="aps_knowledge_base/.chromadb")

    try:
        collection = client.get_or_create_collection("domain_rules")
    except Exception:
        collection = client.create_collection("domain_rules")

    text = f"{rule['name']} {rule['description']} {' '.join(rule.get('metadata', {}).get('tags', []))}"
    meta = {
        "rule_id": rule["id"],
        "name": rule["name"],
        "status": rule.get("status", "active"),
        "constraint_type": rule.get("constraint_type", "soft"),
        "file_path": f"domain_rules/{_infer_subdir(rule)}/{rule['id']}.json"
    }
    collection.upsert(ids=[rule["id"]], documents=[text], metadatas=[meta])

完全重建（批量变更或首次设置时使用）

python aps_knowledge_base/scripts/rebuild_index.py

完整脚本内容请参见 references/scripts.md。

Git 版本管理

每次确认的知识变更都会自动触发 Git 提交。

import subprocess

def _git_commit(item: dict, confirmed_by: str):
    kb_path = "aps_knowledge_base"
    item_id = item.get("id", item.get("memory_type", "unknown"))
    item_type = item.get("type", "update")

    action = "add" if item.get("status") == "active" else "update"
    msg = f"{action}: {item_id} {item_type} ({confirmed_by})"

    subprocess.run(["git", "-C", kb_path, "add", "-A"], check=True)
    subprocess.run(["git", "-C", kb_path, "commit", "-m", msg], check=True)

提交信息规范：

add: rule_003 operator_constraint (big_boss)
update: client_memory shop_floor topology (plant_manager)
deprecate: rule_002 machine_a3 calibration - operator left (admin)
restore: rule_002 machine_a3 calibration (admin)

查看特定规则的历史记录：

git -C aps_knowledge_base log --oneline -- domain_rules/operator_rules/rule_003.json

更新知识状态

停用规则（软禁用 — 保留记录）

def deprecate_rule(rule_id: str, reason: str, deprecated_by: str):
    # 查找对应文件
    for f in (kb / "domain_rules").rglob(f"{rule_id}.json"):
        rule = json.loads(f.read_text())
        rule["status"] = "deprecated"
        rule["metadata"]["deprecated_at"] = datetime.datetime.utcnow().isoformat() + "Z"
        rule["metadata"]["deprecation_reason"] = reason
        f.write_text(json.dumps(rule, ensure_ascii=False, indent=2))

        # 从向量索引中移除，避免被检索到
        client = chromadb.PersistentClient(path="aps_knowledge_base/.chromadb")
        col = client.get_collection("domain_rules")
        col.update(ids=[rule_id], metadatas=[{**col.get(ids=[rule_id])["metadatas"][0], "status": "deprecated"}])

        _git_commit({"id": rule_id, "type": "deprecation"}, deprecated_by)
        _refresh_rule_index()
        return True
    return False

记录排程决策（审计日志）

每次排程会话结束后，应持久化保存该决策以供后续参考：

def log_decision(session_id: str, decision: dict, rules_used: list[str]):
    log_entry = {
        "session_id": session_id,
        "timestamp": datetime.datetime.utcnow().isoformat() + "Z",
        "decision_summary": decision,
        "triggered_by_rules": rules_used,
        "human_confirmed": True
    }
    log_path = kb / f"client_memory/decision_history/{session_id}.json"
    log_path.write_text(json.dumps(log_entry, ensure_ascii=False, indent=2))

    # 同时增加每个被触发规则的使用次数
    for rule_id in rules_used:
        _increment_use_count(rule_id)

知识库健康检查

请定期运行这些检查，或在重大排程会话前执行。

def check_knowledge_health() -> dict:
    issues = []
    profile = json.loads((kb / "client_memory/_profile.json").read_text())

    # 检查规则中引用了已不存在的人或机器
    known_operators = profile.get("operators", {}).get("active", [])
    for f in (kb / "domain_rules").rglob("*.json"):
        rule = json.loads(f.read_text())
        if rule.get("status") != "active":
            continue
        for op in rule.get("scope", {}).get("operators", []):
            if op not in known_operators:
                issues.append({
                    "rule_id": rule["id"],
                    "issue": f"引用了不在活跃名单中的操作员 '{op}'"
                })

    # 标记超过 180 天未使用的规则
    cutoff = datetime.datetime.utcnow() - datetime.timedelta(days=180)
    for f in (kb / "domain_rules").rglob("*.json"):
        rule = json.loads(f.read_text())
        if rule.get("status") != "active":
            continue
        last_used = rule.get("metadata", {}).get("last_used_at")
        if last_used and datetime.datetime.fromisoformat(last_used[:-1]) < cutoff:
            issues.append({
                "rule_id": rule["id"],
                "issue": "超过 180 天未使用 — 建议弃用"
            })

    return {"issues": issues, "checked_at": datetime.datetime.utcnow().isoformat()}

初始化新的知识库

如果 aps_knowledge_base/ 目录不存在，请进行初始化：

mkdir -p aps_knowledge_base/{domain_rules/{machine_rules,operator_rules,material_rules},client_memory/{shop_floor,planning_process,decision_history},problem_schemas,solver_configs,pending_review,logs/{decisions,knowledge_changes},.chromadb}

cd aps_knowledge_base && git init && git commit --allow-empty -m "init: knowledge base"

然后创建 client_memory/_profile.json 文件，根据对话内容填充其结构（使用 propose_memory_update + 确认流程）。

详见 references/schemas.md 中各类文件的完整 JSON Schema 定义。

每次排程会话前的决策检查清单

1. 加载 client_memory/_profile.json — 确认车间拓扑结构为最新状态
2. 使用订单批次描述通过语义搜索获取最相关的前 5 条规则
3. 检查 pending_review/ 目录 — 若有待处理的提案，需向用户展示
4. 加载对应的 problem_schemas/<type>.json 模板文件
5. 排程完成后，调用 log_decision() 并传入实际触发的规则列表
6. 若对话中出现新约束，调用 propose_rule() 并等待用户确认

参考文件

需要时可查阅以下文件以获取详细模式定义和重建脚本：

• references/schemas.md — 规则、客户端记忆、提案等的完整 JSON Schema
• references/scripts.md — rebuild_index.py 的完整源代码