Liu_JB/fastapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

精简后端功能模块并补充教学互动

Browse Source
This commit is contained in:
刘金宝
2026-06-08 16:49:45 +08:00
parent 11b1712b01
commit f0cdc454b3
18 changed files with 1120 additions and 1194 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+86 -87
View File
@@ -1,34 +1,64 @@
# 医疗问诊 Agent FastAPI 后端
医疗问诊 Agent 是医疗教学平台中的问诊训练服务。后端负责 Django 用户身份校验、病例读取、多轮问诊、检查申请、诊断治疗提交、AI 评价、评分明细、PDF 报告和历史训练记录。
医疗问诊 Agent 是医疗教学平台中的训练服务。后端负责 Django 用户鉴权、病例读取、训练会话、流式问诊、练习提示、检查结果、诊断治疗提交、AI 评价、教学互动评价、训练记录和 PDF 下载。
病例库在本服务中为只读数据源。病例新增、解析、修改和删除由外部病例管理系统负责；本服务只读取已发布病例及其训练扩展、检查项和评分规则。
病例新增、病例解析、病例导入和病例删除不在本服务中实现。本服务只读取数据库中已经维护好的病例、检查项、教学题和评分规则。
## 当前保留功能
训练页面：
- 推荐配置信息
- 训练配置信息
- 新建会话
- 流式会话
- 王主任练习提示
- 体格检查列表获取
- 辅助检查列表获取
- 体格检查某项结果
- 辅助检查某项结果
- 完成问诊
- 提交诊断
- 提交治疗
- 生成评价
- 获取评价详情
- 下载 PDF
教学互动：
- 获取教学列表，包含题目、选项、答案、解析文本和视频
- 生成评价
- 获取评价详情
- 下载 PDF
个人中心：
- 训练记录列表
- 训练记录详情
基础能力：
- Django access token 鉴权
- MySQL 数据读取和训练记录写入
- Redis 短期会话 memory
- OpenAI-compatible LLM 调用
- Swagger / OpenAPI
- 健康检查
## 项目结构
仓库根目录可以直接部署为服务器的 `fastapi/` 目录：
```text
fastapi/
├── app/ # FastAPI 应用、Agent、服务、模型与提示词
├── scripts/ # 数据库迁移、结构检查与运维脚本
├── tests/ # 核心逻辑与接口测试
├── app/ # FastAPI 应用、Agent、服务、模型和提示词
├── scripts/ # 数据库初始化和检查脚本
├── tests/ # 当前功能测试
├── docs/03_api_design.md # 前端联调 API 文档
├── Dockerfile
├── requirements.txt
├── .env.example
└── .env.production.example
```
## 核心依赖
- Python 3.11
- FastAPI
- SQLAlchemy 2.x
- MySQL 8
- Redis 7
- OpenAI-compatible LLM API
- Django 用户中心 `/api/user/users/me/`
## 本地启动
```powershell
@@ -47,14 +77,14 @@ http://127.0.0.1:9000/docs
真实密码、LLM Key 和 access token 只写入本地 `.env` 或服务器环境变量，不提交到 Git。
## Docker Compose 部署
## 服务器部署
服务器目录：
服务器目录示例：
```text
/home/code/medical-ai/
├── django/
├── fastapi/ # 本仓库
├── fastapi/
├── vueapp/
├── vuecms/
└── docker-compose.yml
@@ -69,41 +99,28 @@ cp fastapi/.env.production.example fastapi/.env
vi fastapi/.env
```
必须在服务器 `.env` 中填写：
服务器 `.env` 至少配置：
- MySQL 密码和数据库名
- `LLM_API_KEY`
- 实际前端来源 `CORS_ALLOW_ORIGINS`
- Nginx 使用 `/fastapi/` 前缀时保留 `APP_ROOT_PATH=/fastapi`
父目录 `docker-compose.yml` 的 FastAPI 服务需要包含：
```yaml
fastapi:
build:
context: ./fastapi
container_name: fastapi
restart: always
ports:
- "9000:9000"
env_file:
- ./fastapi/.env
volumes:
- ./logs/fastapi:/app/logs
- ./data/fastapi-reports:/app/storage/reports
depends_on:
- mysql
- redis
- django
networks:
- medical
```env
APP_ENV=production
APP_ROOT_PATH=/fastapi
DATABASE_URL=mysql+pymysql://root:1822..@mysql:3306/medical?charset=utf8mb4
REDIS_URL=redis://redis:6379/0
RUNTIME_MEMORY_BACKEND=redis
AUTH_VALIDATE_ENABLED=true
AUTH_USER_ME_URL=http://django:8000/api/user/users/me/
LLM_BASE_URL=<模型服务地址>
LLM_API_KEY=<模型密钥>
LLM_MODEL=<模型名称>
LLM_FAST_MODEL=<模型名称>
LLM_REASON_MODEL=<模型名称>
CORS_ALLOW_ORIGINS=http://8.160.178.88
```
构建并启动：
```bash
cd /home/code/medical-ai
docker compose config
docker compose build fastapi
docker compose up -d fastapi
docker compose logs --tail=200 fastapi
@@ -119,35 +136,9 @@ docker compose build fastapi
docker compose up -d fastapi
```
## 数据库初始化与检查
## 验证
服务启动后先进行只读结构检查：
```bash
cd /home/code/medical-ai
docker compose exec fastapi python scripts/check_final_schema.py
docker compose exec fastapi python scripts/check_final_demo_readiness.py
```
以下迁移脚本只用于独立本地开发库或旧环境升级，不得用于共享生产病例库：
```bash
docker compose exec fastapi python scripts/migrate_to_new_schema.py
docker compose exec fastapi python scripts/migrate_user_department_score_detail.py
```
迁移脚本使用 `create_all` 补齐 Agent 所需表，不删除 Django 或现有业务表；`migrate_to_new_schema.py` 会写入 Demo 病例和基础数据。共享环境中的病例数据由外部病例管理系统维护。
## 部署验证
容器内部端口验证：
```bash
curl http://127.0.0.1:9000/health/live
curl http://127.0.0.1:9000/health/ready
```
使用 Nginx `/fastapi/` 代理后的公网验证：
公网验证：
```text
http://8.160.178.88/fastapi/docs
@@ -155,9 +146,7 @@ http://8.160.178.88/fastapi/openapi.json
http://8.160.178.88/fastapi/health/ready
```
`/health/live` 返回 200 表示 FastAPI 进程正常。`/health/ready` 返回 200 表示 MySQL、Redis 和关键配置已经就绪；返回 503 时查看响应中的检查项和容器日志。
验证 Django 用户中心联调：
Django 用户鉴权验证：
```bash
curl "http://8.160.178.88/fastapi/api/v1/auth/me" \
@@ -165,7 +154,7 @@ curl "http://8.160.178.88/fastapi/api/v1/auth/me" \
-H "X-Entry-Scene: production_vue"
```
验证 PDF 文件流下载：
PDF 下载验证：
```bash
curl -L "http://8.160.178.88/fastapi/api/v1/evaluations/<evaluation_id>/download-pdf" \
@@ -183,12 +172,22 @@ python tests\test_api_contract.py
python tests\test_demo_flow.py
```
当前测试覆盖训练页主要链路：
测试覆盖：
- Django token 鉴权与 user_id 隔离。
- 新建会话、流式问诊、练习提示。
- 体格检查列表、辅助检查列表、单项检查结果和重复申请幂等。
- 完成问诊、提交诊断、提交治疗、生成评价。
- 评价详情、历史评价、PDF 路径导出、PDF 文件流下载和跨用户访问拒绝。
- Django token 鉴权和 user_id 隔离
- 推荐配置和训练配置
- 新建会话、流式问诊、王主任练习提示
- 体格检查和辅助检查列表
- 单项检查结果和重复申请幂等
- 完成问诊、提交诊断、提交治疗、生成评价
- 教学互动列表和教学互动评价
- 训练记录列表、评价详情、PDF 下载
- 跨用户访问拦截
病例新增、解析、导入、删除不在本 FastAPI 服务中实现；本服务只读取数据库中已发布病例、检查项和评分规则。
## API 文档
前端联调文档见：
```text
docs/03_api_design.md
```
app/agents/orchestrator.py
+19
View File
@@ -57,6 +57,25 @@ class MedicalConsultationOrchestrator:
)
return self.report_agent.build_report(scoring_result)
async def evaluate_teaching(
self,
*,
case: CaseBase,
teaching_payload: dict,
scoring_rules: list,
guideline_refs: list[dict],
score_type: str,
) -> dict:
"""教学互动评价编排：调用 Scoring Agent 后复用 Report Agent 整理报告结构。"""
scoring_result = await self.scoring_agent.score_teaching(
case=case,
teaching_payload=teaching_payload,
scoring_rules=scoring_rules,
guideline_refs=guideline_refs,
score_type=score_type,
)
return self.report_agent.build_report(scoring_result)
async def generate_hints(
self,
session: TrainingSession,
app/agents/scoring_agent.py
+156
View File
@@ -58,6 +58,162 @@ class ScoringAgent:
}
return data
async def score_teaching(
self,
*,
case: CaseBase,
teaching_payload: dict,
scoring_rules: list,
guideline_refs: list[dict],
score_type: str,
) -> dict:
"""教学互动评价：根据题目、标准答案、学生作答和评分规则生成结构化评价。"""
start = time.perf_counter()
messages = self._build_teaching_messages(case, teaching_payload, scoring_rules, guideline_refs, score_type)
try:
response = await self.llm.chat(
messages,
settings.llm_fast_model,
thinking_enabled=settings.llm_fast_thinking_enabled,
reasoning_effort=None,
response_format={"type": "json_object"} if settings.llm_scoring_json_response else None,
max_tokens=min(settings.llm_scoring_max_tokens, 1600),
)
data = json.loads(response.content)
data = self._normalize_score_payload(data, score_type, guideline_refs)
data["_llm_model"] = response.model
data["_latency_metrics"] = {"scoring_latency_ms": response.latency_ms, "fallback_used": False}
return data
except (AppError, json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
logger.warning("teaching_scoring_agent.fallback case_id=%s error=%s", case.id, exc.__class__.__name__)
data = self._fallback_teaching_score(score_type, guideline_refs, teaching_payload)
data["_llm_model"] = f"local-fallback-{settings.llm_fast_model}"
data["_latency_metrics"] = {
"scoring_latency_ms": int((time.perf_counter() - start) * 1000),
"fallback_used": True,
"fallback_reason": exc.__class__.__name__,
}
return data
def _build_teaching_messages(
self,
case: CaseBase,
teaching_payload: dict,
scoring_rules: list,
guideline_refs: list[dict],
score_type: str,
) -> list[dict]:
"""教学评分提示词：只传教学互动评价需要的病例、题目、答案和评分规则。"""
payload = {
"score_type": score_type,
"case": {
"case_id": case.id,
"title": case.title,
"chief_complaint": case.chief_complaint,
"description": self._truncate(case.description, 320),
"knowledge_points": case.knowledge_points or [],
"key_points": case.key_points or [],
},
"teaching": teaching_payload,
"scoring_rules": self._compact_scoring_rules(scoring_rules),
"guidelines": self._compact_guidelines(guideline_refs),
}
system = (
"你是医学教学互动评价专家，只输出合法 JSON。"
"请根据病例、教学目标、选择题、标准答案、解析文本、学生作答和评分规则生成教学评价。"
"输出字段固定为 score_type,total_score,dimension_scores,errors,improvement_plan,"
"evidence_summary,guideline_refs,overall_comment,score_details。"
"dimension_scores 包含 知识掌握、临床推理、检查理解、治疗决策、人文沟通 维度，"
"每项包含 dimension,score,max_score,comment,evidence,deductions,improvement。"
"score_details 对应 scoring_rules 或题目维度，每项包含 rule_id,dimension,score,"
"deducted_reason,evidence_message_ids,ai_confidence,comment。"
"必须指出答对题目、答错题目、错误原因、下一步学习重点。"
"本评价仅用于医学教学训练，不替代真实临床诊疗。"
)
return [{"role": "system", "content": system}, {"role": "user", "content": json.dumps(payload, ensure_ascii=False)}]
def _fallback_teaching_score(self, score_type: str, guideline_refs: list[dict], teaching_payload: dict) -> dict:
"""教学评分兜底：LLM 不可用时按选择题正确率生成稳定评价结构。"""
results = teaching_payload.get("answer_results") or []
total = len(results)
correct = sum(1 for item in results if item.get("is_correct"))
accuracy = correct / total if total else 0
total_score = round(accuracy * 100, 1) if total else 0
incorrect = [item for item in results if not item.get("is_correct")]
incorrect_titles = [f"{item.get('question_id')}: {item.get('stem', '')}" for item in incorrect[:5]]
data = {
"score_type": "percentage",
"total_score": total_score,
"dimension_scores": [
{
"dimension": "知识掌握",
"score": round(total_score * 0.35, 1),
"max_score": 35,
"comment": f"共 {total} 题，答对 {correct} 题。",
"evidence": [f"正确率 {round(accuracy * 100, 1)}%"],
"deductions": incorrect_titles,
"improvement": "复习错题对应知识点和病例解析。",
},
{
"dimension": "临床推理",
"score": round(total_score * 0.25, 1),
"max_score": 25,
"comment": "根据选择题表现评估临床判断链路。",
"evidence": [item.get("stem", "") for item in results[:3]],
"deductions": incorrect_titles,
"improvement": "把题目选项与病例主诉、体征和检查结果逐项对应。",
},
{
"dimension": "检查理解",
"score": round(total_score * 0.15, 1),
"max_score": 15,
"comment": "重点关注检查项目与病情严重程度判断。",
"evidence": teaching_payload.get("scoring_focus", "").split("、")[:3],
"deductions": [],
"improvement": "理解血氧、胸片和炎症指标在肺炎评估中的作用。",
},
{
"dimension": "治疗决策",
"score": round(total_score * 0.15, 1),
"max_score": 15,
"comment": "根据题目表现评估治疗原则掌握情况。",
"evidence": teaching_payload.get("teaching_goal", "").split("、")[:3],
"deductions": [],
"improvement": "复习抗感染、平喘、氧合监测和风险预案。",
},
{
"dimension": "人文沟通",
"score": round(total_score * 0.10, 1),
"max_score": 10,
"comment": "教学互动中需继续强化家属沟通和健康教育。",
"evidence": ["教学互动题包含沟通与健康教育相关内容。"],
"deductions": [],
"improvement": "向家属说明病情、观察指标、复诊指征和用药注意事项。",
},
],
"score_details": [],
"errors": [
{
"title": "教学题目答题错误",
"description": "；".join(incorrect_titles) if incorrect_titles else "暂无明显错题。",
"severity": "medium" if incorrect_titles else "low",
"related_dimension": "知识掌握",
}
],
"improvement_plan": [
"复盘错题解析，明确每个选项与病例证据的对应关系。",
"把病例中的主诉、体征、检查和治疗原则整理成一条临床推理链。",
"针对血氧、胸片、炎症指标和医患沟通进行专项复习。",
],
"evidence_summary": [
f"教学互动共提交 {total} 题，答对 {correct} 题。",
"评分依据包括题目标准答案、解析文本、教学目标和评分规则。",
],
"guideline_refs": guideline_refs,
"overall_comment": f"本次教学互动正确率为 {round(accuracy * 100, 1)}%，请结合错题解析继续巩固病例关键知识点。",
}
return self._convert_to_five_point(data) if score_type == "five_point" else data
def _build_messages(
self,
session: TrainingSession,
app/api/knowledge.py
-24
View File
@@ -1,24 +0,0 @@
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from app.core.response import ApiResponse, ok
from app.core.user_context import UserContext, get_user_context
from app.db.session import get_db
from app.schemas.knowledge import KnowledgeSearchResponse
from app.services.knowledge_service import KnowledgeService
router = APIRouter()
@router.get("/search", response_model=ApiResponse[KnowledgeSearchResponse])
def search_knowledge(
_: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
department_id: int = Query(...),
training_type: str = Query(...),
q: str = Query(default=""),
):
"""知识检索：按科室、训练类别和关键词检索评分参考指南。"""
keywords = [item.strip() for item in q.split(",") if item.strip()]
result = KnowledgeService(db).search_guidelines(department_id, training_type, keywords)
return ok(KnowledgeSearchResponse(**result))
app/api/llm_test.py
-108
View File
@@ -1,108 +0,0 @@
import time
from fastapi import APIRouter, Depends
from app.agents.llm_adapter import OpenAICompatibleLLMClient
from app.core.config import settings
from app.core.exceptions import AppError
from app.core.response import ApiResponse, ok
from app.core.user_context import UserContext, get_user_context
from app.schemas.llm import LLMTestRequest, LLMTestResponse
router = APIRouter()
@router.post("/deepseek-fast", response_model=ApiResponse[LLMTestResponse])
async def test_deepseek_fast(
payload: LLMTestRequest,
_: UserContext = Depends(get_user_context),
):
"""Fast 模型测试：验证快速模型的非流式响应耗时。"""
client = OpenAICompatibleLLMClient()
response = await client.chat(
[{"role": "user", "content": payload.message}],
settings.llm_fast_model,
thinking_enabled=settings.llm_fast_thinking_enabled,
max_tokens=min(settings.llm_fast_max_tokens, 256),
)
return ok(
LLMTestResponse(
model=response.model,
total_latency_ms=response.latency_ms,
stream=False,
mock_mode=client.is_mock_mode,
fallback_used=response.model.startswith("mock-fallback"),
thinking_enabled=settings.llm_fast_thinking_enabled,
)
)
@router.post("/deepseek-reason", response_model=ApiResponse[LLMTestResponse])
async def test_deepseek_reason(
payload: LLMTestRequest,
_: UserContext = Depends(get_user_context),
):
"""Reason 模型测试：优先验证流式耗时，流式不兼容时降级为真实非流式测试。"""
client = OpenAICompatibleLLMClient()
messages = [{"role": "user", "content": payload.message}]
first_token_ms = None
start = time.perf_counter()
try:
async for chunk in client.stream_chat(
messages,
settings.llm_reason_model,
thinking_enabled=settings.llm_reason_thinking_enabled,
reasoning_effort=settings.llm_reasoning_effort if settings.llm_reason_thinking_enabled else None,
max_tokens=min(settings.llm_fast_max_tokens, 256),
):
if first_token_ms is None and chunk.first_token_ms is not None:
first_token_ms = chunk.first_token_ms
if chunk.done:
return ok(
LLMTestResponse(
model=chunk.model or (settings.llm_reason_model if not client.is_mock_mode else f"mock-{settings.llm_reason_model}"),
first_token_ms=first_token_ms,
total_latency_ms=chunk.total_latency_ms or int((time.perf_counter() - start) * 1000),
stream=True,
mock_mode=client.is_mock_mode,
fallback_used=chunk.fallback_used,
thinking_enabled=settings.llm_reason_thinking_enabled,
reasoning_effort=settings.llm_reasoning_effort if settings.llm_reason_thinking_enabled else None,
)
)
except AppError as exc:
if exc.code != "LLM_STREAM_FAILED":
raise
response = await client.chat(
messages,
settings.llm_reason_model,
thinking_enabled=settings.llm_reason_thinking_enabled,
reasoning_effort=settings.llm_reasoning_effort if settings.llm_reason_thinking_enabled else None,
max_tokens=min(settings.llm_fast_max_tokens, 256),
)
return ok(
LLMTestResponse(
model=response.model,
first_token_ms=None,
total_latency_ms=response.latency_ms,
stream=False,
mock_mode=client.is_mock_mode,
fallback_used=response.model.startswith("mock-fallback"),
thinking_enabled=settings.llm_reason_thinking_enabled,
reasoning_effort=settings.llm_reasoning_effort if settings.llm_reason_thinking_enabled else None,
)
)
return ok(
LLMTestResponse(
model=settings.llm_reason_model,
first_token_ms=first_token_ms,
total_latency_ms=int((time.perf_counter() - start) * 1000),
stream=True,
mock_mode=client.is_mock_mode,
fallback_used=False,
thinking_enabled=settings.llm_reason_thinking_enabled,
reasoning_effort=settings.llm_reasoning_effort if settings.llm_reason_thinking_enabled else None,
)
)
app/api/router.py
+2 -3
View File
@@ -1,6 +1,6 @@
from fastapi import APIRouter
from app.api import agent, auth, cases, evaluations, knowledge, llm_test, sessions, training_config
from app.api import agent, auth, cases, evaluations, sessions, teaching, training_config
api_router = APIRouter()
api_router.include_router(agent.router, tags=["agent"])
@@ -8,6 +8,5 @@ api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
api_router.include_router(cases.router, prefix="/cases", tags=["cases"])
api_router.include_router(training_config.router, prefix="/training-config", tags=["training-config"])
api_router.include_router(sessions.router, prefix="/sessions", tags=["sessions"])
api_router.include_router(teaching.router, prefix="/teaching", tags=["teaching"])
api_router.include_router(evaluations.router, prefix="/evaluations", tags=["evaluations"])
api_router.include_router(knowledge.router, prefix="/knowledge", tags=["knowledge"])
api_router.include_router(llm_test.router, prefix="/llm/test", tags=["llm-test"])
app/api/teaching.py
+32
View File
@@ -0,0 +1,32 @@
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.core.response import ApiResponse, ok
from app.core.user_context import UserContext, get_user_context
from app.db.session import get_db
from app.schemas.teaching import CreateTeachingEvaluationRequest, TeachingEvaluationResponse, TeachingItemsResponse
from app.services.teaching_service import TeachingService
router = APIRouter()
@router.get("/cases/{case_id}/items", response_model=ApiResponse[TeachingItemsResponse])
def get_teaching_items(
case_id: int,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""教学列表：返回病例、题目、选项、答案、解析文本和教学视频。"""
return ok(TeachingService(db).list_items(ctx, case_id))
@router.post("/evaluation", response_model=ApiResponse[TeachingEvaluationResponse])
async def create_teaching_evaluation(
payload: CreateTeachingEvaluationRequest,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""教学评价：根据教学互动作答生成 AI 评价并写入训练记录。"""
result = await TeachingService(db).create_evaluation(ctx, payload)
db.commit()
return ok(result)
app/core/config.py
+1 -1
View File
@@ -146,7 +146,7 @@ class Settings(BaseModel):
"stream_chat": self.llm_stream_enabled,
"score_types": ["percentage", "five_point"],
"pdf_export": True,
"knowledge_search": True,
"scoring_guideline_lookup": True,
"llm_mock_enabled": mock_enabled,
"llm_mode": "mock" if mock_enabled else "real",
"llm_fallback_to_mock": self.llm_fallback_to_mock,
app/prompts/knowledge/guideline_search_query.md
-37
View File
@@ -1,37 +0,0 @@
---
template_code: guideline_search_query
agent_type: knowledge
version: v1
scene: guideline_search
model_type: fast
output_format: json
---
# Role
你是评分参考指南检索 Query Agent。
# Task
根据病例、训练类别、诊断和治疗任务生成知识库检索关键词。
# Inputs
- 病例科室。
- 主诉和关键症状。
- 训练类别。
- 用户提交的诊断和治疗方案。
# Rules
- 关键词必须来自病例和任务本身。
- 不生成与病例无关的疾病关键词。
- 控制关键词数量，便于 MySQL 文本检索。
# Output Format
输出合法 JSON：`{"keywords": []}`。
# Safety Boundaries
检索词仅用于教学评分参考，不用于真实临床检索决策。
app/prompts/scoring/teaching_interaction_evaluation.md
+81
View File
@@ -0,0 +1,81 @@
---
template_code: scoring_teaching_interaction
agent_type: scoring
version: v1
scene: teaching_interaction
model_type: fast
output_format: json
---
# Role
你是医学教学互动评价专家，负责根据病例、教学题目、标准答案、解析文本、学生作答和评分规则生成教学训练评价。
# Task
对教学互动模式的选择题作答结果进行评分，指出学生对病例知识点、临床推理、检查理解、治疗决策和人文沟通的掌握情况。
# Inputs
- case_base 病例基础信息。
- teaching_case 教学目标、教师引导、评分重点。
- questions 题目、选项、标准答案、解析文本、视频资源。
- student_answers 学生作答。
- answer_results 后端计算的对错结果。
- scoring_rules 病例评分规则。
- guideline_refs 评分参考指南。
# Rules
- 只输出合法 JSON，不输出 Markdown。
- 必须指出答对题目、答错题目和错因。
- 不编造数据库中不存在的题目、答案、检查结果或视频。
- 评价仅用于医学教学训练，不替代真实临床诊疗。
- 评分要尽量复用 scoring_rules 的维度和权重。
# Output Format
```json
{
"score_type": "percentage",
"total_score": 85,
"dimension_scores": [
{
"dimension": "知识掌握",
"score": 30,
"max_score": 35,
"comment": "能够识别支气管肺炎核心诊断依据。",
"evidence": ["q2 选择正确"],
"deductions": ["q1 对严重程度指标理解不足"],
"improvement": "复习血氧、胸片和炎症指标的临床意义。"
}
],
"score_details": [
{
"rule_id": 1,
"dimension": "知识掌握",
"score": 30,
"deducted_reason": "严重程度判断题答错。",
"evidence_message_ids": ["q1", "q2"],
"ai_confidence": 0.86,
"comment": "基础诊断方向正确，严重程度评估需加强。"
}
],
"errors": [
{
"title": "严重程度评估不足",
"description": "未能优先识别血氧饱和度对病情判断的意义。",
"severity": "medium",
"related_dimension": "临床推理"
}
],
"improvement_plan": ["复习儿童肺炎严重程度评估。"],
"evidence_summary": ["共完成 5 题，答对 4 题。"],
"guideline_refs": [],
"overall_comment": "教学互动表现良好，需加强严重程度评估。"
}
```
# Safety Boundaries
本评价仅用于医学教学训练和学习反馈，不提供真实诊疗结论，不替代医生临床判断。
app/repositories/teaching_repository.py
+25
View File
@@ -0,0 +1,25 @@
from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload
from app.models.source_case import CaseBase
class TeachingRepository:
"""教学互动仓储：读取 case_base + teaching_case 以及评分相关扩展数据。"""
def __init__(self, db: Session) -> None:
self.db = db
def get_active_teaching_case(self, case_id: int) -> CaseBase | None:
"""教学病例读取：校验病例已发布、已启用且存在 teaching_case 扩展。"""
stmt = (
select(CaseBase)
.options(
selectinload(CaseBase.teaching_case),
selectinload(CaseBase.traditional_case),
selectinload(CaseBase.scoring_rules),
)
.where(CaseBase.id == case_id, CaseBase.status == 1, CaseBase.publish_status == 1)
)
case = self.db.scalar(stmt)
return case if case and case.teaching_case else None
app/schemas/knowledge.py
-9
View File
@@ -1,9 +0,0 @@
from pydantic import BaseModel
class KnowledgeSearchResponse(BaseModel):
"""知识检索响应：返回评分参考指南片段和来源。"""
matched_chunks: list[dict]
source_refs: list[dict]
no_match: bool
app/schemas/llm.py
-20
View File
@@ -1,20 +0,0 @@
from pydantic import BaseModel
class LLMTestRequest(BaseModel):
"""LLM 测试入参：用于快速模型和 reason 模型耗时验证。"""
message: str = "请用一句话说明医疗问诊训练 Demo 的用途。"
class LLMTestResponse(BaseModel):
"""LLM 测试响应：返回模型名、首 token 时间和总耗时。"""
model: str
first_token_ms: int | None = None
total_latency_ms: int
stream: bool
mock_mode: bool = False
fallback_used: bool = False
thinking_enabled: bool | None = None
reasoning_effort: str | None = None
app/schemas/teaching.py
+75
View File
@@ -0,0 +1,75 @@
from pydantic import BaseModel, Field
from app.schemas.evaluation import EvaluationResponse
class TeachingVideo(BaseModel):
"""教学视频：题目解析关联的视频资源。"""
title: str = ""
url: str = ""
class TeachingOption(BaseModel):
"""教学选项：单选题或多选题的选项结构。"""
key: str
text: str
class TeachingQuestion(BaseModel):
"""教学题目：从 teaching_case 解析出的互动题目。"""
question_id: str
question_type: str = "single_choice"
stem: str
options: list[TeachingOption] = Field(default_factory=list)
answer: str | list[str]
analysis: str = ""
video: TeachingVideo | None = None
knowledge_points: list[str] = Field(default_factory=list)
class TeachingCaseSummary(BaseModel):
"""教学病例摘要：教学互动页面展示的病例基础信息。"""
case_id: int
title: str
department_id: int | None = None
difficulty: str
chief_complaint: str
description: str
patient_age: int | None = None
patient_gender: str | None = None
knowledge_points: list[str] = Field(default_factory=list)
class TeachingItemsResponse(BaseModel):
"""教学列表响应：病例、教学目标、题目、答案、解析文本和视频。"""
case: TeachingCaseSummary
teaching_goal: str
teacher_guide: str
scoring_focus: str
questions: list[TeachingQuestion]
class TeachingAnswer(BaseModel):
"""教学作答：前端提交的单题选择结果。"""
question_id: str = Field(min_length=1, max_length=64)
selected_answer: str | list[str]
class CreateTeachingEvaluationRequest(BaseModel):
"""教学评价入参：提交教学互动题目作答并生成评价。"""
case_id: int
answers: list[TeachingAnswer] = Field(min_length=1)
score_type: str = Field(default="percentage", pattern="^(percentage|five_point)$")
class TeachingEvaluationResponse(EvaluationResponse):
"""教学评价响应：复用训练评价结构，并返回教学会话 ID。"""
session_id: int
app/services/teaching_service.py
+357
View File
@@ -0,0 +1,357 @@
from __future__ import annotations
import json
import uuid
from datetime import datetime
from typing import Any
from sqlalchemy.orm import Session
from app.agents.orchestrator import MedicalConsultationOrchestrator
from app.core.context import UserContext
from app.core.exceptions import AppError
from app.models.training import TrainingSession
from app.models.training_record import TrainingRecord
from app.repositories.evaluation_repository import EvaluationRepository
from app.repositories.source_case_repository import SourceCaseRepository
from app.repositories.teaching_repository import TeachingRepository
from app.schemas.teaching import (
CreateTeachingEvaluationRequest,
TeachingCaseSummary,
TeachingEvaluationResponse,
TeachingItemsResponse,
TeachingOption,
TeachingQuestion,
TeachingVideo,
)
from app.services.audit_service import AuditService
from app.services.evaluation_service import EvaluationService
from app.services.knowledge_service import KnowledgeService
class TeachingService:
"""教学互动服务：读取教学题目、提交作答并生成教学互动评价。"""
def __init__(self, db: Session) -> None:
self.db = db
self.repo = TeachingRepository(db)
self.source_repo = SourceCaseRepository(db)
self.eval_repo = EvaluationRepository(db)
self.audit = AuditService(db)
self.knowledge = KnowledgeService(db)
self.orchestrator = MedicalConsultationOrchestrator()
self.evaluation_service = EvaluationService(db)
def list_items(self, ctx: UserContext, case_id: int) -> TeachingItemsResponse:
"""教学列表：读取 case_base + teaching_case 并返回题目、选项、答案、解析和视频。"""
case = self.repo.get_active_teaching_case(case_id)
if not case:
raise AppError("TEACHING_CASE_NOT_FOUND", "teaching case not found or inactive", 404)
teaching = case.teaching_case
questions = self._parse_questions(case)
self.audit.log(ctx, "teaching.items", "case_base", str(case.id), None)
return TeachingItemsResponse(
case=TeachingCaseSummary(
case_id=case.id,
title=case.title,
department_id=case.department_id,
difficulty=case.difficulty,
chief_complaint=case.chief_complaint,
description=case.description,
patient_age=case.patient_age,
patient_gender=case.patient_gender,
knowledge_points=case.knowledge_points or [],
),
teaching_goal=teaching.teaching_goal,
teacher_guide=teaching.teacher_guide,
scoring_focus=teaching.scoring_focus,
questions=questions,
)
async def create_evaluation(self, ctx: UserContext, payload: CreateTeachingEvaluationRequest) -> TeachingEvaluationResponse:
"""教学评价：校验作答后调用 LLM 评分，并写入 training_record 与评分明细。"""
case = self.repo.get_active_teaching_case(payload.case_id)
if not case:
raise AppError("TEACHING_CASE_NOT_FOUND", "teaching case not found or inactive", 404)
teaching = case.teaching_case
questions = self._parse_questions(case)
if not questions:
raise AppError("TEACHING_QUESTION_EMPTY", "teaching questions are empty", 400)
answer_results = self._build_answer_results(questions, payload.answers)
session = self._create_teaching_session(ctx, case.id, payload.score_type, answer_results)
guideline_result = self.knowledge.search_guidelines(
case.department_id or 0,
case.case_type,
(case.knowledge_points or []) + (case.key_points or []),
)
scoring_rules = self.source_repo.get_scoring_rules(case.id)
teaching_payload = {
"teaching_goal": teaching.teaching_goal,
"teacher_guide": teaching.teacher_guide,
"scoring_focus": teaching.scoring_focus,
"questions": [item.model_dump() for item in questions],
"student_answers": [item.model_dump() for item in payload.answers],
"answer_results": answer_results,
"correct_count": sum(1 for item in answer_results if item["is_correct"]),
"total_count": len(answer_results),
}
report = await self.orchestrator.evaluate_teaching(
case=case,
teaching_payload=teaching_payload,
scoring_rules=scoring_rules,
guideline_refs=guideline_result["source_refs"],
score_type=payload.score_type,
)
record = self._build_training_record(ctx, session, case, teaching_payload, report, scoring_rules, guideline_result)
self.eval_repo.create_record(record)
score_details = self.evaluation_service._build_score_details(record.id, report, scoring_rules)
self.eval_repo.replace_score_details(record.id, score_details)
self.audit.log(ctx, "teaching.evaluation.generate", "training_record", str(record.id), session.id)
response = self.evaluation_service._to_response(record)
return TeachingEvaluationResponse(session_id=session.id, **response.model_dump())
def _create_teaching_session(self, ctx: UserContext, case_id: int, score_type: str, answer_results: list[dict]) -> TrainingSession:
"""教学会话：创建轻量 session 以复用评价详情、历史记录和 PDF 能力。"""
now = datetime.utcnow()
session_code = f"teach_{now.strftime('%Y%m%d%H%M%S')}_{uuid.uuid4().hex[:8]}"
session = TrainingSession(
session_code=session_code,
user_id=ctx.user_id,
tenant_id=ctx.tenant_id,
class_id=ctx.class_id,
entry_scene=ctx.entry_scene,
case_id=case_id,
training_type="teaching_interaction",
mode="teaching",
score_type=score_type,
status="completed",
started_at=now,
completed_at=now,
memory_key=None,
metadata_={"source": "teaching_interaction", "answer_results": answer_results},
)
self.db.add(session)
self.db.flush()
return session
def _build_training_record(
self,
ctx: UserContext,
session: TrainingSession,
case,
teaching_payload: dict,
report: dict,
scoring_rules: list,
guideline_result: dict,
) -> TrainingRecord:
"""教学记录：把教学互动评价沉淀为 training_record。"""
total_score = float(report.get("total_score") or 0)
structured = {
"score_type": report.get("score_type", session.score_type),
"total_score": total_score,
"dimension_scores": report.get("dimension_scores") or [],
"score_details": report.get("score_details") or [],
"errors": report.get("errors") or [],
"improvement_plan": report.get("improvement_plan") or [],
"evidence_summary": report.get("evidence_summary") or [],
"guideline_refs": report.get("guideline_refs") or [],
"overall_comment": report.get("overall_comment") or "",
"llm_model": report.get("_llm_model"),
"latency_metrics": report.get("_latency_metrics") or {},
"teaching_summary": {
"correct_count": teaching_payload.get("correct_count", 0),
"total_count": teaching_payload.get("total_count", 0),
},
}
return TrainingRecord(
training_mode="teaching",
case_type="teaching_interaction",
start_time=session.started_at or datetime.utcnow(),
end_time=session.completed_at or datetime.utcnow(),
duration_seconds=0,
total_score=total_score,
ai_score=total_score,
teacher_score=None,
evaluation_level=self.evaluation_service._evaluation_level(total_score, structured["score_type"]),
status="completed",
feedback=structured["overall_comment"],
thinking_chain=json.dumps(
{
"teaching_goal": teaching_payload.get("teaching_goal", ""),
"scoring_focus": teaching_payload.get("scoring_focus", ""),
"answer_results": teaching_payload.get("answer_results", []),
"scoring_rule_count": len(scoring_rules),
"guideline_refs": structured["guideline_refs"],
},
ensure_ascii=False,
),
diagnosis_path=json.dumps(
{
"case_title": case.title,
"question_count": teaching_payload.get("total_count", 0),
"correct_count": teaching_payload.get("correct_count", 0),
},
ensure_ascii=False,
),
wrong_points=structured["errors"],
missed_questions=[],
recommendation_result={"improvement_plan": structured["improvement_plan"]},
ai_feedback_structured=structured,
osce_station_score={},
interruption_count=0,
emotion_analysis={},
prompt_version="teaching_interaction_v1",
rag_context_version=self.evaluation_service._rag_context_version(guideline_result),
case_id=case.id,
teacher_id=None,
user_id=self.evaluation_service._numeric_user_id(ctx.user_id),
external_user_id=ctx.user_id,
session_id=session.id,
evaluation_record_id=None,
score_type=structured["score_type"],
pdf_file_path=None,
)
def _build_answer_results(self, questions: list[TeachingQuestion], answers: list) -> list[dict]:
"""作答校验：按 question_id 对比标准答案并生成评分证据。"""
question_map = {item.question_id: item for item in questions}
results: list[dict] = []
for answer in answers:
question = question_map.get(answer.question_id)
if not question:
raise AppError("TEACHING_QUESTION_NOT_FOUND", f"question {answer.question_id} not found", 404)
selected = self._normalize_answer(answer.selected_answer)
expected = self._normalize_answer(question.answer)
results.append(
{
"question_id": question.question_id,
"stem": question.stem,
"selected_answer": selected,
"correct_answer": expected,
"is_correct": selected == expected,
"analysis": question.analysis,
"knowledge_points": question.knowledge_points,
}
)
return results
def _parse_questions(self, case) -> list[TeachingQuestion]:
"""题目解析：从 teaching_case.discussion_questions 解析 JSON，失败时返回病例默认题目。"""
raw = case.teaching_case.discussion_questions if case.teaching_case else ""
payload: Any
try:
payload = json.loads(raw)
except (TypeError, json.JSONDecodeError):
payload = self._fallback_questions(case)
if isinstance(payload, dict):
payload = payload.get("questions") or []
if not isinstance(payload, list) or not payload:
payload = self._fallback_questions(case)
return [self._normalize_question(item, index) for index, item in enumerate(payload, start=1)]
def _normalize_question(self, item: dict, index: int) -> TeachingQuestion:
"""题目结构归一：补齐选项、答案、解析、视频等字段。"""
options = item.get("options") or []
normalized_options = [
TeachingOption(key=str(option.get("key") or ""), text=str(option.get("text") or ""))
for option in options
if isinstance(option, dict)
]
video = item.get("video")
return TeachingQuestion(
question_id=str(item.get("question_id") or f"q{index}"),
question_type=str(item.get("question_type") or "single_choice"),
stem=str(item.get("stem") or item.get("question") or ""),
options=normalized_options,
answer=item.get("answer") or "",
analysis=str(item.get("analysis") or ""),
video=TeachingVideo(**video) if isinstance(video, dict) else None,
knowledge_points=[str(value) for value in (item.get("knowledge_points") or [])],
)
def _normalize_answer(self, value: str | list[str]) -> list[str]:
"""答案归一：把单选和多选统一为排序后的大写字符串数组。"""
raw_values = value if isinstance(value, list) else [value]
return sorted(str(item).strip().upper() for item in raw_values if str(item).strip())
def _fallback_questions(self, case) -> list[dict]:
"""默认题目：当数据库暂未维护结构化题库时，为儿科肺炎 demo 生成可测试题目。"""
video = {"title": "儿童肺炎教学示例视频", "url": ""}
return [
{
"question_id": "q1",
"question_type": "single_choice",
"stem": "该患儿最需要优先关注的病情严重程度指标是？",
"options": [
{"key": "A", "text": "体温峰值"},
{"key": "B", "text": "血氧饱和度"},
{"key": "C", "text": "咳嗽天数"},
{"key": "D", "text": "食欲下降"},
],
"answer": "B",
"analysis": "血氧饱和度能帮助判断低氧和肺炎严重程度，是处置决策的重要依据。",
"video": video,
"knowledge_points": ["严重程度评估", "血氧判断"],
},
{
"question_id": "q2",
"question_type": "single_choice",
"stem": "结合发热、咳嗽、喘息和肺部湿啰音，最符合的诊断方向是？",
"options": [
{"key": "A", "text": "支气管肺炎"},
{"key": "B", "text": "急性胃肠炎"},
{"key": "C", "text": "泌尿系感染"},
{"key": "D", "text": "单纯过敏性鼻炎"},
],
"answer": "A",
"analysis": "呼吸道症状、肺部体征和影像/炎症指标共同支持儿童支气管肺炎。",
"video": video,
"knowledge_points": ["诊断依据", "肺部体征"],
},
{
"question_id": "q3",
"question_type": "single_choice",
"stem": "下列哪组检查最有助于完善本例肺炎诊断和严重程度评估？",
"options": [
{"key": "A", "text": "血常规、CRP、胸片、血氧饱和度"},
{"key": "B", "text": "肝功能、甲状腺功能、腹部超声"},
{"key": "C", "text": "胃镜、幽门螺杆菌、粪便常规"},
{"key": "D", "text": "骨龄片、维生素D、微量元素"},
],
"answer": "A",
"analysis": "炎症指标、胸部影像和血氧情况可共同支撑诊断和严重程度判断。",
"video": video,
"knowledge_points": ["检查选择", "辅助检查"],
},
{
"question_id": "q4",
"question_type": "single_choice",
"stem": "治疗方案中最需要覆盖的核心原则是？",
"options": [
{"key": "A", "text": "抗感染、止咳平喘、改善氧合、严密观察"},
{"key": "B", "text": "立即长期激素维持治疗"},
{"key": "C", "text": "只需补充维生素"},
{"key": "D", "text": "无需随访观察"},
],
"answer": "A",
"analysis": "儿童肺炎处置需围绕抗感染、呼吸症状缓解、氧合监测和病情变化预案展开。",
"video": video,
"knowledge_points": ["治疗原则", "风险预案"],
},
{
"question_id": "q5",
"question_type": "single_choice",
"stem": "向家属沟通时，最合适的内容是？",
"options": [
{"key": "A", "text": "说明病情、观察指标、用药注意事项和复诊/住院指征"},
{"key": "B", "text": "只告知已经开药即可"},
{"key": "C", "text": "不需要解释检查结果"},
{"key": "D", "text": "避免回答家属担心的问题"},
],
"answer": "A",
"analysis": "儿科场景需要重视家属知情、风险信号识别和家庭护理教育。",
"video": video,
"knowledge_points": ["人文沟通", "健康教育"],
},
]
docs/03_api_design.md
+149 -888
View File
File diff suppressed because it is too large Load Diff
scripts/init_demo_db.py
+96 -2
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import sys
import json
from pathlib import Path
from sqlalchemy import select
@@ -116,19 +117,111 @@ def _seed_traditional_case(db, case_id: int) -> None:
def _seed_teaching_case(db, case_id: int) -> None:
"""教学互动病例种子：教学互动模式读取 case_base + teaching_case。"""
if db.scalar(select(TeachingCase).where(TeachingCase.case_id == case_id)):
existing = db.scalar(select(TeachingCase).where(TeachingCase.case_id == case_id))
if existing:
try:
json.loads(existing.discussion_questions)
except (TypeError, json.JSONDecodeError):
existing.discussion_questions = _demo_teaching_questions_json()
return
questions_json = _demo_teaching_questions_json()
case = db.get(CaseBase, case_id)
if case and not case.multimodal_assets:
case.multimodal_assets = [{"type": "video", "title": "儿童肺炎教学示例视频", "url": ""}]
db.add(
TeachingCase(
case_id=case_id,
teaching_goal="围绕儿科肺炎问诊、检查选择、诊断依据、治疗决策和医患沟通完成互动训练。",
discussion_questions="如何判断病情严重程度？哪些检查是关键检查？治疗方案如何兼顾抗感染、平喘和氧合监测？",
discussion_questions=questions_json,
teacher_guide="观察学生是否完整追问发热、咳嗽、喘息、既往史、接触史，并能解释胸片、炎症指标和血氧。",
scoring_focus="问诊完整性、检查合理性、诊断准确性、治疗计划、风险预案、人文沟通。",
)
)
def _demo_teaching_questions_json() -> str:
"""教学题库种子：为儿科支气管肺炎病例生成可测试的选择题、答案、解析和视频字段。"""
video = {"title": "儿童肺炎教学示例视频", "url": ""}
questions = [
{
"question_id": "q1",
"question_type": "single_choice",
"stem": "该患儿最需要优先关注的病情严重程度指标是？",
"options": [
{"key": "A", "text": "体温峰值"},
{"key": "B", "text": "血氧饱和度"},
{"key": "C", "text": "咳嗽天数"},
{"key": "D", "text": "食欲下降"},
],
"answer": "B",
"analysis": "血氧饱和度能帮助判断低氧和肺炎严重程度，是处置决策的重要依据。",
"video": video,
"knowledge_points": ["严重程度评估", "血氧判断"],
},
{
"question_id": "q2",
"question_type": "single_choice",
"stem": "结合发热、咳嗽、喘息和肺部湿啰音，最符合的诊断方向是？",
"options": [
{"key": "A", "text": "支气管肺炎"},
{"key": "B", "text": "急性胃肠炎"},
{"key": "C", "text": "泌尿系感染"},
{"key": "D", "text": "单纯过敏性鼻炎"},
],
"answer": "A",
"analysis": "呼吸道症状、肺部体征和影像/炎症指标共同支持儿童支气管肺炎。",
"video": video,
"knowledge_points": ["诊断依据", "肺部体征"],
},
{
"question_id": "q3",
"question_type": "single_choice",
"stem": "下列哪组检查最有助于完善本例肺炎诊断和严重程度评估？",
"options": [
{"key": "A", "text": "血常规、CRP、胸片、血氧饱和度"},
{"key": "B", "text": "肝功能、甲状腺功能、腹部超声"},
{"key": "C", "text": "胃镜、幽门螺杆菌、粪便常规"},
{"key": "D", "text": "骨龄片、维生素D、微量元素"},
],
"answer": "A",
"analysis": "炎症指标、胸部影像和血氧情况可共同支撑诊断和严重程度判断。",
"video": video,
"knowledge_points": ["检查选择", "辅助检查"],
},
{
"question_id": "q4",
"question_type": "single_choice",
"stem": "治疗方案中最需要覆盖的核心原则是？",
"options": [
{"key": "A", "text": "抗感染、止咳平喘、改善氧合、严密观察"},
{"key": "B", "text": "立即长期激素维持治疗"},
{"key": "C", "text": "只需补充维生素"},
{"key": "D", "text": "无需随访观察"},
],
"answer": "A",
"analysis": "儿童肺炎处置需围绕抗感染、呼吸症状缓解、氧合监测和病情变化预案展开。",
"video": video,
"knowledge_points": ["治疗原则", "风险预案"],
},
{
"question_id": "q5",
"question_type": "single_choice",
"stem": "向家属沟通时，最合适的内容是？",
"options": [
{"key": "A", "text": "说明病情、观察指标、用药注意事项和复诊/住院指征"},
{"key": "B", "text": "只告知已经开药即可"},
{"key": "C", "text": "不需要解释检查结果"},
{"key": "D", "text": "避免回答家属担心的问题"},
],
"answer": "A",
"analysis": "儿科场景需要重视家属知情、风险信号识别和家庭护理教育。",
"video": video,
"knowledge_points": ["人文沟通", "健康教育"],
},
]
return json.dumps(questions, ensure_ascii=False)
def _seed_exam_items(db, case_id: int) -> None:
"""检查项目种子：写入病例可申请检查和固定返回结果。"""
if db.scalar(select(CaseExamItem).where(CaseExamItem.case_id == case_id)):
@@ -265,6 +358,7 @@ def _seed_prompts(db) -> None:
("patient_teaching", "patient", "teaching", "v1", "fast", "text", "app/prompts/patient/teaching.md"),
("novice_case_hint", "hint", "novice", "v1", "fast", "json", "app/prompts/hint/novice_case_hint.md"),
("scoring_pediatrics_pneumonia", "scoring", "pediatrics_pneumonia", "v1", "fast", "json", "app/prompts/scoring/pediatrics_pneumonia.md"),
("scoring_teaching_interaction", "scoring", "teaching_interaction", "v1", "fast", "json", "app/prompts/scoring/teaching_interaction_evaluation.md"),
("report_evaluation", "report", "evaluation", "v1", "fast", "json", "app/prompts/report/evaluation_report.md"),
]
for code, agent_type, scene, version, model_type, output_format, file_path in templates:
tests/test_api_contract.py
+41 -15
View File
@@ -86,9 +86,6 @@ def run_api_contract_tests() -> None:
auth_me_operation = openapi_payload["paths"]["/api/v1/auth/me"]["get"]
assert any("HTTPBearer" in item for item in auth_me_operation.get("security", []))
assert "HTTPBearer" in openapi_payload["components"]["securitySchemes"]
assert "/api/v1/imports/case-sql/preview" not in openapi_payload["paths"]
assert "/api/v1/imports/case-sql/apply" not in openapi_payload["paths"]
assert "/api/v1/cases/{case_id}/delete-preview" not in openapi_payload["paths"]
assert "delete" not in openapi_payload["paths"]["/api/v1/cases/{case_id}"]
assert "/api/v1/training-config/recommended" in openapi_payload["paths"]
assert "/api/v1/training-config/options" in openapi_payload["paths"]
@@ -98,11 +95,52 @@ def run_api_contract_tests() -> None:
assert "/api/v1/sessions/{session_id}/physical-exams/{item_code}" in openapi_payload["paths"]
assert "/api/v1/sessions/{session_id}/auxiliary-exams/{item_code}" in openapi_payload["paths"]
assert "/api/v1/evaluations/{evaluation_id}/download-pdf" in openapi_payload["paths"]
assert "/api/v1/teaching/cases/{case_id}/items" in openapi_payload["paths"]
assert "/api/v1/teaching/evaluation" in openapi_payload["paths"]
cases = client.get("/api/v1/cases", headers=headers)
assert cases.status_code == 200
case_id = cases.json()["data"]["items"][0]["id"]
teaching_items = client.get(f"/api/v1/teaching/cases/{case_id}/items", headers=headers)
assert teaching_items.status_code == 200
teaching_data = teaching_items.json()["data"]
assert teaching_data["case"]["case_id"] == case_id
assert teaching_data["questions"]
assert teaching_data["questions"][0]["options"]
assert teaching_data["questions"][0]["answer"]
assert "analysis" in teaching_data["questions"][0]
teaching_answers = [
{"question_id": item["question_id"], "selected_answer": item["answer"]}
for item in teaching_data["questions"]
]
teaching_evaluation = client.post(
"/api/v1/teaching/evaluation",
headers=headers,
json={"case_id": case_id, "answers": teaching_answers, "score_type": "percentage"},
)
assert teaching_evaluation.status_code == 200
teaching_evaluation_id = teaching_evaluation.json()["data"]["evaluation_id"]
assert teaching_evaluation.json()["data"]["session_id"]
assert teaching_evaluation.json()["data"]["score_details"]
teaching_detail = client.get(f"/api/v1/evaluations/{teaching_evaluation_id}", headers=headers)
assert teaching_detail.status_code == 200
assert teaching_detail.json()["data"]["evaluation_id"] == teaching_evaluation_id
teaching_pdf_download = client.get(f"/api/v1/evaluations/{teaching_evaluation_id}/download-pdf", headers=headers)
assert teaching_pdf_download.status_code == 200
assert teaching_pdf_download.headers["content-type"].startswith("application/pdf")
assert teaching_pdf_download.content.startswith(b"%PDF")
cross_user_teaching_detail = client.get(
f"/api/v1/evaluations/{teaching_evaluation_id}",
headers={"Authorization": "Bearer api_user_002_token", "X-Entry-Scene": "api_test"},
)
assert cross_user_teaching_detail.status_code == 404
assert cross_user_teaching_detail.json()["code"] == "EVALUATION_NOT_FOUND"
recommended_config = client.get(f"/api/v1/training-config/recommended?case_id={case_id}", headers=headers)
assert recommended_config.status_code == 200
assert recommended_config.json()["data"]["recommended"]["visit_environment"] == "outpatient"
@@ -320,18 +358,6 @@ def run_api_contract_tests() -> None:
assert teaching_hint.status_code == 400
assert teaching_hint.json()["code"] == "SESSION_STATUS_INVALID"
llm_fast = client.post("/api/v1/llm/test/deepseek-fast", headers=headers, json={"message": "hello"})
assert llm_fast.status_code == 200
assert llm_fast.json()["code"] == "OK"
assert llm_fast.json()["data"]["stream"] is False
llm_reason = client.post("/api/v1/llm/test/deepseek-reason", headers=headers, json={"message": "hello"})
assert llm_reason.status_code == 200
assert llm_reason.json()["code"] == "OK"
assert "total_latency_ms" in llm_reason.json()["data"]
if __name__ == "__main__":
run_api_contract_tests()
print("api contract tests passed")