aps-filesystem-agent

# APS Filesystem Agent Skill This skill teaches an APS scheduling agent how to navigate, query, and maintain a local filesystem-based knowledge base. The filesystem is the single source of truth for all domain rules, client memory, and problem schemas. A vector index sits on top for semantic retrieval, and Git tracks every change for auditability. ## Knowledge base layout ``` aps_knowledge_base/ ├── .git/ ← version history, never touch manually ├── domain_rules/ ← APS rules extracted from conversations │ ├── _index.json ← master rule registry (always update this) │ ├── machine_rules/ │ ├── operator_rules/ │ └── material_rules/ ├── client_memory/ ← persistent understanding of this customer │ ├── _profile.json ← shop floor + planning process + preferences │ ├── shop_floor/ │ ├── planning_process/ │ └── decision_history/ ← one file per scheduling session ├── problem_schemas/ ← modeling templates by problem type ├── solver_configs/ ← solver parameters and routing thresholds ├── pending_review/ ← proposed knowledge awaiting human approval └── logs/ ├── decisions/ ← audit trail of scheduling decisions └── knowledge_changes/ ← audit trail of knowledge writes ``` Before doing anything, confirm the knowledge base root exists: ```bash ls aps_knowledge_base/ 2>/dev/null || echo "Knowledge base not initialized" ``` If it doesn't exist yet, initialize it (see "Initializing a new knowledge base" below). --- ## Reading knowledge ### Load client profile Always load the client profile first — it tells you the shop floor topology, planning process, and output preferences that frame every other decision. ```python import json, pathlib kb = pathlib.Path("aps_knowledge_base") profile = json.loads((kb / "client_memory/_profile.json").read_text()) shop = profile["shop_floor"] # type, stages, machines_per_stage, etc. prefs = profile["preferences"] # primary_objective, output_format, etc. ``` ### Semantic retrieval of rules (preferred method) Use semantic search when you know *what you need* but not *which file has it*. This requires the vector index to be built (see "Maintaining the vector index"). ```python import chromadb client = chromadb.PersistentClient(path="aps_knowledge_base/.chromadb") collection = client.get_collection("domain_rules") results = collection.query( query_texts=["operator HSE certification machine maintenance"], n_results=5, where={"status": "active"} # only retrieve active rules ) # 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}") ``` ### Direct rule lookup by ID When you already know the rule ID (e.g., from a decision log): ```python rule_path = kb / f"domain_rules/{category}/{rule_id}.json" rule = json.loads(rule_path.read_text()) ``` ### Load all active rules for a scheduling session Inject the Top-K most relevant rules into the scheduling context: ```python 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 ``` ### Load a problem schema template ```python problem_type = "flow_shop" # or job_shop, rcpsp, re_entrant schema = json.loads((kb / f"problem_schemas/{problem_type}.json").read_text()) ``` ### Read session decision history ```python 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 {} ``` --- ## Proposing new knowledge (write path) **The agent NEVER writes directly to the main knowledge directories.** All new knowledge goes to `pending_review/` first, then a human confirms. ### Propose a new APS rule Call this whenever you extract a new constraint or rule from a conversation: ```python 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 the summary to show the user for confirmation return { "proposal_file": str(out_path), "rule_id": rule_content["id"], "name": rule_content["name"], "description": rule_content["description"] } ``` After calling this, **always present the proposal to the user** with a confirmation prompt before moving on. Format it like this: ``` 建议将以下内容加入知识库： 规则ID: {rule_id} 名称: {name} 描述: {description} 来源: "{source_quote}" [确认入库] [修改后入库] [忽略本次] ``` Wait for explicit confirmation before proceeding to `confirm_proposal()`. ### Propose an update to client memory ```python 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) ``` --- ## Confirming proposals (after human approval) Only call these functions **after** the user has explicitly confirmed in chat. ```python def confirm_proposal(proposal_file: str, confirmed_by: str): """Move a proposal from pending_review into the live knowledge base.""" 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) # Remove from pending proposal_path.unlink() # Update vector index and commit _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 the index file _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)) ``` --- ## Maintaining the vector index The vector index must stay in sync with the filesystem. Rebuild it whenever rules are added, updated, or deprecated. ### Incremental update (after a single rule change) ```python 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]) ``` ### Full rebuild (use after bulk changes or first setup) ```bash python aps_knowledge_base/scripts/rebuild_index.py ``` See `references/scripts.md` for the full rebuild script content. --- ## Git version management Every confirmed knowledge change gets a Git commit automatically. ```python 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) ``` Commit message conventions: ``` 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) ``` To view history for a specific rule: ```bash git -C aps_knowledge_base log --oneline -- domain_rules/operator_rules/rule_003.json ``` --- ## Updating knowledge status ### Deprecate a rule (soft disable — keeps the record) ```python def deprecate_rule(rule_id: str, reason: str, deprecated_by: str): # find the file for f in (kb / "domain_rules").rglob(f"{rule_id}.json"): rule = json.loads(f.read_text()) rule["status"] = "deprecated"