qiaoqichaoandClaude Code
d56c699ce6
按 docs/v2/05-implementation-plan.md §1 候选迁移表与 claude_use/T00 白名单， 从 55e2cdb 抽取 224 个文件（2.48 MB）到独立目录 F:/mini-OpenClaw-v2。 - 排除：凭据(.env/server-deploy.conf/旧DELIVERY.md)、运行数据(xingbang_data 等 4 个 SQLite)、依赖(venv/node_modules 符号链接)、归档(_backup*/test-evidence/ ori_file)、Office 临时文件 - 目标目录独立 git 仓库，初始恢复点；Gitea 远程与分支待 OQ-06 Q2 定案 - 未部署、未启动服务、未调用模型 Co-Authored-By: Claude Code <noreply@anthropic.com>
210 lines
6.5 KiB
Python
210 lines
6.5 KiB
Python
"""历史评审案例向量库
|
||
|
||
功能：
|
||
1. 存储历史评审案例（项目特征+评审结果+评审员行为）
|
||
2. 相似案例检索（基于向量相似度）
|
||
3. 评审员行为学习（团队集体经验，非个人预测）
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
from typing import List, Dict, Optional, Tuple
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class HistoryStore:
|
||
"""历史评审案例存储（简化版，生产环境建议使用 Chroma/Qdrant）"""
|
||
|
||
def __init__(self, data_dir: str = "./xingbang_data/history"):
|
||
self.data_dir = Path(data_dir)
|
||
self.data_dir.mkdir(parents=True, exist_ok=True)
|
||
self.cases_file = self.data_dir / "review_cases.jsonl"
|
||
|
||
# 确保文件存在
|
||
if not self.cases_file.exists():
|
||
self.cases_file.touch()
|
||
|
||
def add_case(self, case_data: Dict) -> str:
|
||
"""添加一条历史评审案例"""
|
||
case_id = f"case_{datetime.now().strftime('%Y%m%d%H%M%S')}"
|
||
|
||
case_record = {
|
||
"case_id": case_id,
|
||
"timestamp": datetime.now().isoformat(),
|
||
**case_data
|
||
}
|
||
|
||
# 追加写入 JSONL
|
||
with open(self.cases_file, "a", encoding="utf-8") as f:
|
||
f.write(json.dumps(case_record, ensure_ascii=False) + "\n")
|
||
|
||
logger.info(f"案例已保存: {case_id}")
|
||
return case_id
|
||
|
||
def search_similar_cases(
|
||
self,
|
||
industry: str,
|
||
credit_amount: float,
|
||
risk_level: str = None,
|
||
limit: int = 5
|
||
) -> List[Dict]:
|
||
"""检索相似案例（简化版：基于行业+金额范围）"""
|
||
|
||
if not self.cases_file.exists():
|
||
return []
|
||
|
||
similar_cases = []
|
||
|
||
with open(self.cases_file, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
if not line.strip():
|
||
continue
|
||
|
||
try:
|
||
case = json.loads(line)
|
||
|
||
# 行业匹配
|
||
industry_match = industry in case.get("industry", "") or case.get("industry", "") in industry
|
||
|
||
# 金额范围匹配（±30%）
|
||
case_amount = case.get("credit_amount", 0)
|
||
amount_match = 0.7 * credit_amount <= case_amount <= 1.3 * credit_amount
|
||
|
||
# 风险等级匹配（可选）
|
||
risk_match = risk_level is None or case.get("risk_level") == risk_level
|
||
|
||
if industry_match and amount_match and risk_match:
|
||
similar_cases.append(case)
|
||
|
||
except json.JSONDecodeError:
|
||
continue
|
||
|
||
# 按时间倒序返回最近的案例
|
||
similar_cases.sort(key=lambda x: x.get("timestamp", ""), reverse=True)
|
||
return similar_cases[:limit]
|
||
|
||
def get_case_by_id(self, case_id: str) -> Optional[Dict]:
|
||
"""根据 ID 获取案例详情"""
|
||
|
||
if not self.cases_file.exists():
|
||
return None
|
||
|
||
with open(self.cases_file, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
if not line.strip():
|
||
continue
|
||
|
||
try:
|
||
case = json.loads(line)
|
||
if case.get("case_id") == case_id:
|
||
return case
|
||
except json.JSONDecodeError:
|
||
continue
|
||
|
||
return None
|
||
|
||
def get_team_patterns(self, industry: str = None) -> Dict:
|
||
"""提取团队集体经验模式（高频风险点、常见处理方式）"""
|
||
|
||
if not self.cases_file.exists():
|
||
return {"high_frequency_risks": [], "common_solutions": []}
|
||
|
||
risk_counter = {}
|
||
solution_counter = {}
|
||
|
||
with open(self.cases_file, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
if not line.strip():
|
||
continue
|
||
|
||
try:
|
||
case = json.loads(line)
|
||
|
||
# 行业过滤
|
||
if industry and industry not in case.get("industry", ""):
|
||
continue
|
||
|
||
# 统计高频风险点
|
||
for risk in case.get("identified_risks", []):
|
||
risk_key = risk.get("type", "unknown")
|
||
risk_counter[risk_key] = risk_counter.get(risk_key, 0) + 1
|
||
|
||
# 统计常见解决方案
|
||
solution = case.get("solution", "")
|
||
if solution:
|
||
solution_counter[solution] = solution_counter.get(solution, 0) + 1
|
||
|
||
except json.JSONDecodeError:
|
||
continue
|
||
|
||
# 排序
|
||
high_frequency_risks = sorted(
|
||
[{"risk": k, "count": v} for k, v in risk_counter.items()],
|
||
key=lambda x: x["count"],
|
||
reverse=True
|
||
)[:10]
|
||
|
||
common_solutions = sorted(
|
||
[{"solution": k, "count": v} for k, v in solution_counter.items()],
|
||
key=lambda x: x["count"],
|
||
reverse=True
|
||
)[:10]
|
||
|
||
return {
|
||
"high_frequency_risks": high_frequency_risks,
|
||
"common_solutions": common_solutions,
|
||
}
|
||
|
||
def get_statistics(self) -> Dict:
|
||
"""获取统计数据"""
|
||
|
||
if not self.cases_file.exists():
|
||
return {"total_cases": 0}
|
||
|
||
total_cases = 0
|
||
industry_dist = {}
|
||
approval_rate = {"approved": 0, "rejected": 0, "pending": 0}
|
||
|
||
with open(self.cases_file, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
if not line.strip():
|
||
continue
|
||
|
||
try:
|
||
case = json.loads(line)
|
||
total_cases += 1
|
||
|
||
# 行业分布
|
||
industry = case.get("industry", "未知")
|
||
industry_dist[industry] = industry_dist.get(industry, 0) + 1
|
||
|
||
# 通过率
|
||
result = case.get("review_result", "pending")
|
||
if result in approval_rate:
|
||
approval_rate[result] += 1
|
||
|
||
except json.JSONDecodeError:
|
||
continue
|
||
|
||
return {
|
||
"total_cases": total_cases,
|
||
"industry_distribution": industry_dist,
|
||
"approval_rate": approval_rate,
|
||
}
|
||
|
||
|
||
# 全局单例
|
||
_history_store: Optional[HistoryStore] = None
|
||
|
||
|
||
def get_history_store() -> HistoryStore:
|
||
"""获取历史案例存储单例"""
|
||
global _history_store
|
||
if _history_store is None:
|
||
_history_store = HistoryStore()
|
||
return _history_store
|