451 lines
23 KiB
Python
451 lines
23 KiB
Python
"""Fixed, typed tool registry for teacher conversations."""
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Callable, Literal
|
|
|
|
from fastapi import HTTPException
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from sqlalchemy import select
|
|
|
|
from .actions import prepare, require_scope
|
|
from .exports import batch_projects, project_result
|
|
from .models import (Conversation, Course, CourseVersion, IndexTask, KnowledgeDocument,
|
|
Project, RubricVersion, ScoringBatch, Student, SubmissionFile)
|
|
from .rag import retrieve
|
|
from .rules import totals
|
|
from .storage import directories, scan
|
|
|
|
|
|
class Args(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
|
|
class Empty(Args):
|
|
pass
|
|
|
|
|
|
class VersionId(Args):
|
|
course_version_id: int = Field(gt=0)
|
|
|
|
|
|
class RulesArgs(Args):
|
|
course_version_id: int | None = Field(default=None, gt=0)
|
|
batch_id: int | None = Field(default=None, gt=0)
|
|
criterion_id: str | None = None
|
|
|
|
|
|
class ProjectId(Args):
|
|
project_id: int = Field(gt=0)
|
|
|
|
|
|
class ProjectList(Args):
|
|
course_version_id: int = Field(gt=0)
|
|
student_no: str | None = None
|
|
|
|
|
|
class SearchCourse(VersionId):
|
|
query: str = Field(min_length=2, max_length=300)
|
|
|
|
|
|
class SearchProject(ProjectId):
|
|
query: str = Field(min_length=2, max_length=300)
|
|
|
|
|
|
class Directory(Args):
|
|
relative_dir: str = Field(min_length=1, max_length=500)
|
|
|
|
|
|
class ScanArgs(Directory):
|
|
offset: int = Field(default=0, ge=0)
|
|
limit: int = Field(default=10, ge=1, le=20)
|
|
|
|
|
|
class BatchId(Args):
|
|
batch_id: int = Field(gt=0)
|
|
|
|
|
|
class ScoresFilterArgs(BatchId):
|
|
project_id: int | None = Field(default=None, gt=0)
|
|
status: Literal["pending", "scored", "reviewed", "failed", "unscorable"] | None = None
|
|
|
|
|
|
class ScoresArgs(ScoresFilterArgs):
|
|
project_offset: int = Field(default=0, ge=0)
|
|
criterion_offset: int = Field(default=0, ge=0)
|
|
|
|
|
|
class CreateBatch(VersionId):
|
|
rubric_version_id: int = Field(gt=0)
|
|
|
|
|
|
class Review(Args):
|
|
score_id: int = Field(gt=0)
|
|
score: float = Field(ge=0)
|
|
comment: str = Field(min_length=1, max_length=1000)
|
|
reason: str = Field(min_length=1, max_length=500)
|
|
|
|
|
|
class Export(BatchId):
|
|
format: Literal["excel", "pdf"]
|
|
project_id: int | None = Field(default=None, gt=0)
|
|
|
|
|
|
class PrepareImport(VersionId):
|
|
relative_dir: str = Field(min_length=1, max_length=500)
|
|
offset: int = Field(default=0, ge=0)
|
|
limit: int = Field(default=10, ge=1, le=20)
|
|
|
|
|
|
class IndexRetry(Args):
|
|
task_id: int = Field(gt=0)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Tool:
|
|
description: str
|
|
schema: type[Args]
|
|
handler: Callable
|
|
|
|
|
|
def _version(db, conversation, version_id):
|
|
require_scope(db, conversation, version_id=version_id)
|
|
version = db.get(CourseVersion, version_id)
|
|
if not version:
|
|
raise HTTPException(404, "课程版本不存在")
|
|
return version
|
|
|
|
|
|
def _batch(db, conversation, batch_id):
|
|
batch, _ = require_scope(db, conversation, batch_id=batch_id)
|
|
if not batch:
|
|
raise HTTPException(404, "批次不存在")
|
|
return batch
|
|
|
|
|
|
def _project(db, conversation, project_id):
|
|
project = db.get(Project, project_id)
|
|
if not project:
|
|
raise HTTPException(404, "项目不存在")
|
|
require_scope(db, conversation, project_id=project_id, version_id=project.course_version_id)
|
|
return project
|
|
|
|
|
|
def courses(db, convo, args, _user):
|
|
rows = []
|
|
for course in db.scalars(select(Course).order_by(Course.id)):
|
|
if convo.course_version_id and db.get(CourseVersion, convo.course_version_id).course_id != course.id:
|
|
continue
|
|
versions = db.scalars(select(CourseVersion).where(CourseVersion.course_id == course.id).order_by(CourseVersion.version.desc())).all()
|
|
rows.append({"id": course.id, "name": course.name, "code": course.code,
|
|
"versions": [{"id": v.id, "version": v.version, "title": v.title, "published": v.published}
|
|
for v in versions if not convo.course_version_id or v.id == convo.course_version_id]})
|
|
return rows[:30]
|
|
|
|
|
|
def rules(db, convo, args, _user):
|
|
def compact_rubric(source):
|
|
criteria = source["criteria"]
|
|
if args.criterion_id:
|
|
criteria = [item for item in criteria if item["criterion_id"] == args.criterion_id]
|
|
else:
|
|
criteria = [{key: item[key] for key in ("criterion_id", "category", "name", "max_score")}
|
|
for item in criteria]
|
|
return {"version": source["version"], "criteria": criteria, "weights": source["weights"]}
|
|
batch_id = args.batch_id or convo.batch_id
|
|
if batch_id:
|
|
batch = _batch(db, convo, batch_id)
|
|
if args.course_version_id and args.course_version_id != batch.course_version_id:
|
|
raise HTTPException(400, "指定课程版本与批次快照不一致")
|
|
return {"batch_id": batch.id, "course_version_id": batch.course_version_id,
|
|
"course": {key: batch.course_snapshot.get(key) for key in ("version", "title", "objectives", "indicators", "directions")},
|
|
"rubric": compact_rubric(batch.rubric_snapshot),
|
|
"source": "batch_snapshot", "url": f"/?page=batches&batch_id={batch.id}"}
|
|
version_id = args.course_version_id or convo.course_version_id
|
|
if not version_id:
|
|
raise HTTPException(400, "请指定课程版本；有多个版本时不能猜测")
|
|
version = _version(db, convo, version_id)
|
|
rubrics = db.scalars(select(RubricVersion).where(RubricVersion.course_version_id == version.id).order_by(RubricVersion.version.desc())).all()
|
|
return {"course_version_id": version.id, "title": version.title, "version": version.version,
|
|
"objectives": version.objectives, "indicators": version.indicators,
|
|
"directions": version.directions, "published": version.published,
|
|
"rubrics": [{"id": r.id, "published": r.published,
|
|
**compact_rubric({"version": r.version, "criteria": r.criteria, "weights": r.weights})} for r in rubrics[:5]],
|
|
"source": "current_version", "url": f"/?page=courses&course_version_id={version.id}"}
|
|
|
|
|
|
def projects(db, convo, args, _user):
|
|
_version(db, convo, args.course_version_id)
|
|
query = select(Project).where(Project.course_version_id == args.course_version_id).order_by(Project.id)
|
|
rows = []
|
|
for project in db.scalars(query):
|
|
if convo.project_id and project.id != convo.project_id:
|
|
continue
|
|
student = db.get(Student, project.student_id)
|
|
if args.student_no and student.student_no != args.student_no:
|
|
continue
|
|
rows.append({"id": project.id, "student_no": student.student_no, "name": student.name,
|
|
"topic": project.topic, "status": project.status,
|
|
"url": f"/?page=projects&project_id={project.id}"})
|
|
return {"count": len(rows), "projects": rows[:30], "truncated": len(rows) > 30}
|
|
|
|
|
|
def project(db, convo, args, _user):
|
|
item = _project(db, convo, args.project_id)
|
|
student = db.get(Student, item.student_id)
|
|
files = db.scalars(select(SubmissionFile).where(SubmissionFile.project_id == item.id)).all()
|
|
return {"id": item.id, "student_no": student.student_no, "name": student.name,
|
|
"topic": item.topic, "direction": item.direction, "course_version_id": item.course_version_id,
|
|
"facts": item.facts, "files": [{"id": f.id, "name": f.relative_path.rsplit("/", 1)[-1],
|
|
"category": f.category, "parse_status": f.parse_status, "parse_error": f.parse_error} for f in files],
|
|
"url": f"/?page=projects&project_id={item.id}"}
|
|
|
|
|
|
def _search(db, version_id, project_id, query, *, public_only=False):
|
|
chunks = retrieve(db, course_version_id=version_id, project_id=project_id or 0, query=query,
|
|
limit=6, include_public=False, public_only=public_only)
|
|
citations = []
|
|
for chunk in chunks:
|
|
doc = db.get(KnowledgeDocument, chunk["document_id"])
|
|
file_id = doc.submission_file_id if doc else None
|
|
citations.append({"chunk_id": chunk["chunk_id"], "document_id": chunk["document_id"],
|
|
"title": doc.title if doc else "资料", "page": chunk["page"],
|
|
"quote": chunk["text"][:500], "project_id": chunk["project_id"],
|
|
"url": f"/api/knowledge/documents/{chunk['document_id']}/pages/{chunk['page']}",
|
|
"image_url": f"/api/assets/{file_id}/{chunk['page']}" if file_id else None})
|
|
return {"citations": citations, "found": bool(citations)}
|
|
|
|
|
|
def search_course(db, convo, args, _user):
|
|
_version(db, convo, args.course_version_id)
|
|
return _search(db, args.course_version_id, None, args.query, public_only=True)
|
|
|
|
|
|
def search_project(db, convo, args, _user):
|
|
item = _project(db, convo, args.project_id)
|
|
return _search(db, item.course_version_id, item.id, args.query)
|
|
|
|
|
|
def batch_list(db, convo, args, _user):
|
|
query = select(ScoringBatch).order_by(ScoringBatch.id.desc())
|
|
rows = []
|
|
for batch in db.scalars(query):
|
|
if convo.batch_id and batch.id != convo.batch_id or convo.course_version_id and batch.course_version_id != convo.course_version_id:
|
|
continue
|
|
if convo.project_id and batch.project_ids is not None and convo.project_id not in batch.project_ids:
|
|
continue
|
|
rows.append({"id": batch.id, "course_version_id": batch.course_version_id,
|
|
"rubric_version_id": batch.rubric_version_id, "status": batch.status,
|
|
"done": batch.progress_done, "total": batch.progress_total,
|
|
"project_count": len(batch.project_ids) if batch.project_ids is not None else None,
|
|
"legacy_membership": batch.project_ids is None})
|
|
return rows[:30]
|
|
|
|
|
|
def progress(db, convo, args, _user):
|
|
batch = _batch(db, convo, args.batch_id)
|
|
project_ids = [convo.project_id] if convo.project_id else batch.project_ids
|
|
return {"id": batch.id, "status": batch.status, "done": batch.progress_done,
|
|
"total": batch.progress_total, "error": batch.error,
|
|
"project_ids": project_ids, "legacy_membership": batch.project_ids is None,
|
|
"url": f"/?page=batches&batch_id={batch.id}"}
|
|
|
|
|
|
def scores(db, convo, args, _user):
|
|
batch = _batch(db, convo, args.batch_id)
|
|
selected = batch_projects(db, batch)
|
|
if args.project_id:
|
|
require_scope(db, convo, batch_id=batch.id, project_id=args.project_id)
|
|
selected = [p for p in selected if p.id == args.project_id]
|
|
if convo.project_id:
|
|
selected = [p for p in selected if p.id == convo.project_id]
|
|
results = []
|
|
for item in selected:
|
|
result = project_result(db, batch, item)
|
|
if args.status:
|
|
result["rows"] = [row for row in result["rows"] if row["status"] == args.status]
|
|
if result["rows"]:
|
|
results.append((item, result))
|
|
page = results[args.project_offset:args.project_offset + 1]
|
|
output = []
|
|
citations = []
|
|
for item, result in page:
|
|
url = f"/?page=scores&batch_id={batch.id}&project_id={item.id}"
|
|
rows = result["rows"][args.criterion_offset:args.criterion_offset + 6]
|
|
output.append({"project_id": item.id, "student_no": str(result["student_no"])[:50], "name": str(result["name"])[:100],
|
|
"totals": result["totals"], "url": url, "total_rows": len(result["rows"]),
|
|
"rows": [{"criterion_id": str(row["criterion"]["criterion_id"])[:100],
|
|
"criterion_name": str(row["criterion"]["name"])[:150],
|
|
"max_score": row["criterion"]["max_score"], "score_id": row["score_id"],
|
|
"model_score": row["model_score"], "teacher_score": row["teacher_score"],
|
|
"final_score": row["final_score"], "status": row["status"],
|
|
"error": str(row["error"] or "")[:300], "evidence": [{"page": ev["page"],
|
|
"quote": ev["quote"][:160], "document": str(ev["document"])[:200]}
|
|
for ev in row["evidence"][:2]]} for row in rows]})
|
|
for row in rows:
|
|
for evidence in row["evidence"][:2]:
|
|
citations.append({"title": str(evidence["document"])[:200], "page": evidence["page"],
|
|
"quote": evidence["quote"][:320],
|
|
"url": evidence["image_url"] or url})
|
|
next_criterion_offset = args.criterion_offset + 6 if page and args.criterion_offset + 6 < len(page[0][1]["rows"]) else None
|
|
next_project_offset = args.project_offset + 1 if args.project_offset + 1 < len(results) else None
|
|
return {"batch_id": batch.id, "status_filter": args.status, "count": len(results),
|
|
"project_offset": args.project_offset, "criterion_offset": args.criterion_offset,
|
|
"next_project_offset": next_project_offset, "next_criterion_offset": next_criterion_offset,
|
|
"results": output, "citations": citations}
|
|
|
|
|
|
def summary(db, convo, args, _user):
|
|
batch = _batch(db, convo, args.batch_id)
|
|
if args.project_id:
|
|
require_scope(db, convo, batch_id=batch.id, project_id=args.project_id)
|
|
result = {"batch_id": batch.id, "status": batch.status, "project_count": 0, "complete_count": 0,
|
|
"incomplete_count": 0, "criteria_status_counts": {}, "overall_distribution": {},
|
|
"model_complete_count": 0, "model_average": None, "final_average": None,
|
|
"status_filter": args.status, "matched_criteria_count": 0,
|
|
"legacy_membership": batch.project_ids is None,
|
|
"url": f"/?page=scores&batch_id={batch.id}"}
|
|
model_sum = 0.0
|
|
final_sum = 0.0
|
|
selected = batch_projects(db, batch)
|
|
if args.project_id:
|
|
selected = [item for item in selected if item.id == args.project_id]
|
|
if convo.project_id:
|
|
selected = [item for item in selected if item.id == convo.project_id]
|
|
for item in selected:
|
|
row = project_result(db, batch, item)
|
|
matched = [criterion for criterion in row["rows"] if not args.status or criterion["status"] == args.status]
|
|
if not matched:
|
|
continue
|
|
result["project_count"] += 1
|
|
for criterion in matched:
|
|
status = criterion["status"]
|
|
result["criteria_status_counts"][status] = result["criteria_status_counts"].get(status, 0) + 1
|
|
result["matched_criteria_count"] += 1
|
|
model_scores = {criterion["criterion"]["criterion_id"]: criterion["model_score"] for criterion in row["rows"]}
|
|
model_overall = totals(model_scores, batch.rubric_snapshot["criteria"], batch.rubric_snapshot["weights"])["overall"]
|
|
if model_overall is not None:
|
|
result["model_complete_count"] += 1
|
|
model_sum += model_overall
|
|
overall = row["totals"]["overall"]
|
|
if overall is None:
|
|
result["incomplete_count"] += 1
|
|
else:
|
|
result["complete_count"] += 1
|
|
bucket = f"{int(overall // 10) * 10}-{min(100, int(overall // 10) * 10 + 9)}"
|
|
result["overall_distribution"][bucket] = result["overall_distribution"].get(bucket, 0) + 1
|
|
final_sum += overall
|
|
if result["model_complete_count"]:
|
|
result["model_average"] = round(model_sum / result["model_complete_count"], 2)
|
|
if result["complete_count"]:
|
|
result["final_average"] = round(final_sum / result["complete_count"], 2)
|
|
return result
|
|
|
|
|
|
def index_tasks(db, convo, args, _user):
|
|
_version(db, convo, args.course_version_id)
|
|
tasks = db.scalars(select(IndexTask).where(IndexTask.course_version_id == args.course_version_id).order_by(IndexTask.id.desc()).limit(30)).all()
|
|
rows = []
|
|
for task in tasks:
|
|
if convo.project_id and task.submission_file_id:
|
|
file = db.get(SubmissionFile, task.submission_file_id)
|
|
if not file or file.project_id != convo.project_id:
|
|
continue
|
|
rows.append({"id": task.id, "kind": task.kind, "file_id": task.submission_file_id,
|
|
"status": task.status, "error": task.error, "result": task.result})
|
|
return rows
|
|
|
|
|
|
def browse_directories(db, convo, args, _user):
|
|
if convo.project_id or convo.batch_id:
|
|
raise HTTPException(400, "请在未限定学生或批次的会话中扫描目录")
|
|
return directories(args.relative_dir)
|
|
|
|
|
|
def scan_directories(db, convo, args, _user):
|
|
if convo.project_id or convo.batch_id:
|
|
raise HTTPException(400, "请在未限定学生或批次的会话中扫描目录")
|
|
data = scan(args.relative_dir)
|
|
return {**data, "projects": data["projects"][args.offset:args.offset + args.limit],
|
|
"count": len(data["projects"]), "offset": args.offset, "limit": args.limit}
|
|
|
|
|
|
def prepare_kind(kind):
|
|
def handler(db, convo, args, user_text):
|
|
payload = args.model_dump(exclude_none=True)
|
|
if kind == "review":
|
|
explicit = [float(value) for value in re.findall(r"(?<!\d)(\d+(?:\.\d+)?)\s*分", user_text)]
|
|
digits = {char: index for index, char in enumerate("零一二三四五六七八九")}
|
|
for value in re.findall(r"([零一二三四五六七八九十]+)\s*分", user_text):
|
|
if "十" in value:
|
|
left, right = value.split("十", 1)
|
|
explicit.append(float((digits[left] if left else 1) * 10 + (digits[right] if right else 0)))
|
|
elif len(value) == 1:
|
|
explicit.append(float(digits[value]))
|
|
if args.score not in explicit:
|
|
raise HTTPException(400, "教师须在当前消息中明确给出分数，例如“给 8 分”")
|
|
if kind == "export" and args.format == "pdf" and not args.project_id:
|
|
raise HTTPException(400, "PDF 导出须指定学生项目")
|
|
action = prepare(db, convo, kind, payload)
|
|
return {"pending_action": {"id": action.id, "kind": kind, "status": action.status,
|
|
"requires_edit": action.preview.get("requires_edit", False),
|
|
"expires_at": action.expires_at.isoformat()}}
|
|
return handler
|
|
|
|
|
|
def prepare_import(db, convo, args, _user):
|
|
data = scan(args.relative_dir)
|
|
candidates = data["projects"][args.offset:args.offset + args.limit]
|
|
if not candidates:
|
|
raise HTTPException(400, "扫描范围内没有学生项目")
|
|
projects = []
|
|
for candidate in candidates:
|
|
unique = {category for category in [file["category"] for file in candidate["files"]]
|
|
if category != "unknown" and sum(file["category"] == category for file in candidate["files"]) == 1}
|
|
projects.append({"relative_dir": candidate["relative_dir"],
|
|
"student_no": candidate["student_no"] or "00000", "name": candidate["name"] or "待确认",
|
|
"topic": candidate["topic"] or "待确认", "direction": "",
|
|
"files": [{"relative_path": file["relative_path"], "category": file["category"],
|
|
"sha256": file["sha256"], "status": "recognized" if file["category"] in unique and file["status"] == "recognized" else "ignored"}
|
|
for file in candidate["files"]]})
|
|
action = prepare(db, convo, "import", {"course_version_id": args.course_version_id, "projects": projects})
|
|
return {"pending_action": {"id": action.id, "kind": "import", "status": action.status,
|
|
"requires_edit": action.preview.get("requires_edit", False),
|
|
"project_count": len(projects), "total_scanned": len(data["projects"]),
|
|
"offset": args.offset, "expires_at": action.expires_at.isoformat()}}
|
|
|
|
|
|
TOOLS: dict[str, Tool] = {
|
|
"list_courses": Tool("列出课程、版本和发布状态；名称或版本不清楚时先调用此工具。", Empty, courses),
|
|
"get_course_rules": Tool("查询课程目标和规则。指定批次时只返回批次保存的快照。", RulesArgs, rules),
|
|
"search_course_material": Tool("检索指定课程版本的课程资料，返回原文片段和引用。", SearchCourse, search_course),
|
|
"list_directories": Tool("只读浏览 /data/inbox 下的相对目录。", Directory, browse_directories),
|
|
"scan_directory": Tool("只读分页扫描学生总目录，返回材料候选、缺失、重复和识别异常。", ScanArgs, scan_directories),
|
|
"list_projects": Tool("按课程版本及可选学号查询学生项目。", ProjectList, projects),
|
|
"get_project": Tool("读取单个学生项目的元数据、材料和解析状态。", ProjectId, project),
|
|
"search_project_material": Tool("只检索指定学生项目报告，返回原文片段和引用。", SearchProject, search_project),
|
|
"list_batches": Tool("列出可访问的评分批次。", Empty, batch_list),
|
|
"get_batch_progress": Tool("查询评分批次进度、状态和固定项目范围。", BatchId, progress),
|
|
"get_scores": Tool("分页查询单学生或单批次评分详情，可按状态筛选。每次返回一个项目、最多六项评分；如需其余评分项，用 next_criterion_offset 作为 criterion_offset 再查；如需下个项目，用 next_project_offset 作为 project_offset，并将 criterion_offset 设为 0。", ScoresArgs, scores),
|
|
"summarize_batch": Tool("用数据库计算批次或单学生完成数、未完成数和最终分布，可按评分项状态筛选，不补零。", ScoresFilterArgs, summary),
|
|
"list_index_tasks": Tool("查询课程版本的材料解析与索引任务进度。", VersionId, index_tasks),
|
|
"prepare_import": Tool("从只读扫描结果准备导入确认卡片；每页最多20个项目，文件有歧义时要求教师在卡片中修正。", PrepareImport, prepare_import),
|
|
"prepare_course_ingest": Tool("准备课程资料入库确认卡片。", VersionId, prepare_kind("course_ingest")),
|
|
"prepare_create_batch": Tool("准备评分批次创建确认卡片，固定项目清单。", CreateBatch, prepare_kind("create_batch")),
|
|
"prepare_start_batch": Tool("准备启动或重试评分批次确认卡片。", BatchId, prepare_kind("start_batch")),
|
|
"prepare_review": Tool("准备教师复核确认卡片；教师必须在当前消息明确给出分数。", Review, prepare_kind("review")),
|
|
"prepare_export": Tool("准备 Excel 成绩表或逐生 PDF 反馈导出确认卡片。", Export, prepare_kind("export")),
|
|
"prepare_index_retry": Tool("准备失败索引任务重试确认卡片。", IndexRetry, prepare_kind("index_retry")),
|
|
}
|
|
|
|
|
|
def tool_specs():
|
|
return [{"type": "function", "function": {"name": name, "description": tool.description,
|
|
"parameters": tool.schema.model_json_schema()}} for name, tool in TOOLS.items()]
|
|
|
|
|
|
def execute(db, conversation, name: str, arguments: dict, user_text: str):
|
|
tool = TOOLS.get(name)
|
|
if not tool:
|
|
raise HTTPException(400, "未注册的工具")
|
|
return tool.handler(db, conversation, tool.schema.model_validate(arguments), user_text)
|