36 lines
1.6 KiB
Python
36 lines
1.6 KiB
Python
 
|
"""Tools scoped to an immutable authorized risk run, not the production database."""
|
||
|
|
import json
|
||
|
|
from copy import deepcopy
|
||
|
|
from langchain_core.tools import tool
|
||
|
|
|
||
|
|
|
||
|
|
def create_risk_tools(context, evidence, native_skills):
|
||
|
|
snapshot = deepcopy(context)
|
||
|
|
sources = deepcopy(evidence)
|
||
|
|
skills = deepcopy(native_skills)
|
||
|
|
|
||
|
|
@tool
|
||
|
|
def list_data_cards() -> str:
|
||
|
|
"""List authorized peer data-card snapshots available for this risk task."""
|
||
|
|
return json.dumps([{"index":i,"name":d["card"],"version":d["version"],"records":len(d["records"])} for i,d in enumerate(snapshot.get("peer_datasets",[]))],ensure_ascii=False)
|
||
|
|
|
||
|
|
@tool
|
||
|
|
def read_data_card(index: int) -> str:
|
||
|
|
"""Read one authorized data-card snapshot, including periods and raw evidence."""
|
||
|
|
cards = snapshot.get("peer_datasets", [])
|
||
|
|
if index < 0 or index >= len(cards):
|
||
|
|
return json.dumps({"error":"数据卡不在本次授权范围"},ensure_ascii=False)
|
||
|
|
return json.dumps(cards[index],ensure_ascii=False)
|
||
|
|
|
||
|
|
@tool
|
||
|
|
def read_checks_and_rules() -> str:
|
||
|
|
"""Read deterministic Skill outputs, human rules and rule findings for this run."""
|
||
|
|
return json.dumps({"values":snapshot.get("scalar_values",{}),"human_rules":snapshot.get("human_rules",[]),"findings":snapshot.get("rule_findings",[]),"evidence":sources},ensure_ascii=False)
|
||
|
|
|
||
|
|
@tool
|
||
|
|
def read_selected_skill(skill_id: str) -> str:
|
||
|
|
"""Read only a native SKILL.md explicitly selected in the published plan."""
|
||
|
|
return skills.get(skill_id,"此 Skill 未被所选方案授权")
|
||
|
|
|
||
|
|
return [list_data_cards, read_data_card, read_checks_and_rules, read_selected_skill]
|