APS 文件系统代理技能

该技能教会APS调度代理如何导航、查询和维护基于本地文件系统的知识库。文件系统是所有领域规则、客户记忆和问题模式的唯一真实来源。向量索引位于其上用于语义检索，Git跟踪每一次变更以实现可审计性。

知识库布局

apsknowledgebase/
├── .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/ ← 知识写入审计追踪

在执行任何操作之前，确认知识库根目录存在：
bash
ls apsknowledgebase/ 2>/dev/null || echo 知识库未初始化

如果不存在，则初始化它（参见下方初始化新知识库）。

读取知识

加载客户档案

始终先加载客户档案——它告诉你车间拓扑结构、计划流程和输出偏好，这些构成了所有其他决策的框架。

python
import json, pathlib

kb = pathlib.Path(apsknowledgebase)
profile = json.loads((kb / clientmemory/profile.json).read_text())
shop = profile[shop_floor] # 类型、阶段、每阶段机器数等
prefs = profile[preferences] # 主要目标、输出格式等

语义检索规则（推荐方法）

当你知道需要什么但不知道哪个文件包含它时，使用语义搜索。这需要向量索引已构建（参见维护向量索引）。

python
import chromadb

client = chromadb.PersistentClient(path=apsknowledgebase/.chromadb)
collection = client.getcollection(domainrules)

results = collection.query(
query_texts=[操作员 HSE 认证 机器 维护],
n_results=5,
where={status: active} # 仅检索活跃规则
)

results[ids], results[documents], results[metadatas]

按ID直接查找规则

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

python
rulepath = kb / fdomainrules/{category}/{rule_id}.json
rule = json.loads(rulepath.readtext())

加载调度会话的所有活跃规则

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

python
def getrelevantrules(query: str, top_k: int = 5) -> list[dict]:
collection = client.getcollection(domainrules)
results = collection.query(
query_texts=[query],
nresults=topk,
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

加载问题模式模板

python
problemtype = flowshop # 或 jobshop, rcpsp, reentrant
schema = json.loads((kb / fproblemschemas/{problemtype}.json).read_text())

读取会话决策历史

python
historydir = kb / clientmemory/decision_history
sessions = sorted(historydir.glob(session*.json), reverse=True)
lastsession = json.loads(sessions[0].readtext()) if sessions else {}

提议新知识（写入路径）

代理永远不会直接写入主知识目录。
所有新知识首先进入 pending_review/，然后由人工确认。

提议新APS规则

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

python
import json, pathlib, datetime

def proposerule(rulecontent: dict, sourcequote: str, sessionid: str):
kb = pathlib.Path(apsknowledgebase)
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,
createdby: aiagent,
confirmed_by: None,
sourcesession: sessionid,
sourcequote: sourcequote,
use_count: 0,
confidence: 0.9
}
}

outpath = pending / fproposed{rulecontent[id]}{ts}.json
outpath.writetext(json.dumps(proposal, ensure_ascii=False, indent=2))

# 返回摘要以向用户显示确认
return {
proposalfile: str(outpath),
ruleid: rulecontent[id],
name: rule_content[name],
description: rule_content[description]
}

调用此函数后，始终向用户展示提议并附带确认提示，然后再继续。格式如下：

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

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

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

等待明确确认后再执行 confirm_proposal()。

提议更新客户记忆

python
def proposememoryupdate(memory_type: str, updates: dict, reason: str):

memorytype: shopfloor | planning_process | preferences

pending = kb / pending_review
ts = datetime.datetime.utcnow().strftime(%Y%m%d_%H%M%S)
proposal = {
type: clientmemoryupdate,
memorytype: memorytype,
updates: updates,
reason: reason,
proposed_at: datetime.datetime.utcnow().isoformat() + Z
}
outpath = pending / fproposedmemory{memorytype}_{ts}.json
outpath.writetext(json.dumps(proposal, ensure_ascii=False, indent=2))
return str(out_path)

确认提议（人工批准后）

仅在用户在聊天中明确确认后调用这些函数。

python
def confirmproposal(proposalfile: str, confirmed_by: str):
将提议从pending_review移入活跃知识库。
kb = pathlib.Path(apsknowledgebase)
proposalpath = pathlib.Path(proposalfile)
proposal = json.loads(proposalpath.readtext())

if proposal.get(type) == clientmemoryupdate:
applymemoryupdate(proposal, confirmedby)
else:
applyrule(proposal, confirmed_by)

# 从pending中移除
proposal_path.unlink()

# 更新向量索引并提交
updatevector_index(proposal)
gitcommit(proposal, confirmed_by)

def applyrule(proposal: dict, confirmed_by: str):
rule_type = proposal.get(type, general)
category_map = {
machineconstraint: machinerules,
operatorconstraint: operatorrules,
materialconstraint: materialrules,
}
subdir = categorymap.get(ruletype, machine_rules)
dest = kb / fdomain_rules/{subdir}/{proposal[id]}.json
dest.parent.mkdir(parents=True, exist_ok=True)

proposal[status] = active
proposal[metadata][confirmedby] = confirmedby
proposal[metadata][confirmed_at] = (
datetime.datetime.utcnow().isoformat() + Z
)
dest.writetext(json.dumps(proposal, ensureascii=False, indent=2))

# 刷新索引文件
refreshrule_index()

def applymemoryupdate(proposal: dict, confirmedby: str):
profile_path