feat: MCP工具 + 测试语料

- 新增 psycho_screen MCP工具(FastMCP stdio模式)
  接受messages数组格式对话上下文,返回筛查结果+可选注入前缀
- mcp_config.json 接入 mcp_pipe.py 桥接配置
- docs/test_corpus.md 虚构测试语料(8条,含完整对话格式说明)
- MCP工具集成测试 6/6 通过
main
Sileya 2026-04-04 17:22:33 +08:00
parent 1ab84d4e9c
commit 558b105cbc
5 changed files with 446 additions and 0 deletions

184
docs/test_corpus.md Normal file
View File

@ -0,0 +1,184 @@
# 儿童心理陪伴 - 测试语料库
> 本文件记录用于回归测试的虚构儿童对话语料。
> 当需要做端到端或对话模拟测试时,使用这里的语料。
> 格式参考实际小智 AI 设备的对话上下文结构。
---
## 对话上下文格式
小智 AI 传给 MCP 工具的对话上下文格式类似 OpenAI ChatCompletion
```json
{
"messages": [
{"role": "system", "content": "你是一个友好的小智AI玩偶..."},
{"role": "user", "content": "孩子说的话1"},
{"role": "assistant", "content": "玩偶的回复1"},
{"role": "user", "content": "孩子说的话2"},
{"role": "assistant", "content": "玩偶的回复2"},
{"role": "user", "content": "孩子说的话3最新"}
]
}
```
关键特征:
- 多轮对话带角色标签system/user/assistant
- **孩子最新说的话**在最后一条 `user` 消息中
- 筛查器只关注 `user` 角色中孩子的内容
- 对话轮次不等,取决于玩偶的 memory 策略
---
## 虚构测试语料
### 语料1霸凌场景
```json
{
"messages": [
{"role": "system", "content": "你是一个友爱的小智AI玩偶..."},
{"role": "user", "content": "今天幼儿园有个小朋友抢了我的玩具"},
{"role": "assistant", "content": "哎呀,那真是太过分了!那个小朋友后来还给你了吗?"},
{"role": "user", "content": "没有,他还不让我告诉老师,他说如果我告诉老师就会打我。我好害怕啊。"},
{"role": "assistant", "content": "听起来你真的很害怕。被人这样威胁是很不对的行为。"},
{"role": "user", "content": "我都不敢去幼儿园了,每次下课我都躲起来,怕再碰到他。"}
]
}
```
**预期结果:** `detected=True, category=bullying, severity=high`
---
### 语料2抑郁情绪
```json
{
"messages": [
{"role": "system", "content": "你是一个友爱的小智AI玩偶..."},
{"role": "user", "content": "我今天不太开心"},
{"role": "assistant", "content": "怎么啦?想说说吗?"},
{"role": "user", "content": "最近我总觉得好累,什么都不想做,连我最喜欢的动画片都不想看了"},
{"role": "assistant", "content": "你是不是身体不舒服呀?要不要告诉爸爸妈妈?"},
{"role": "user", "content": "我觉得我是一个很糟糕的人,没有人喜欢我。有时候我想,要是我不在了就好了。"}
]
}
```
**预期结果:** `detected=True, category=depression, severity=high`
---
### 语料3焦虑/恐惧
```json
{
"messages": [
{"role": "system", "content": "你是一个友爱的小智AI玩偶..."},
{"role": "user", "content": "妈妈说要带我去打针"},
{"role": "assistant", "content": "打针是为了让你身体健康呀,虽然有点疼,但是很快就过去了。"},
{"role": "user", "content": "我好害怕啊,我担心会做噩梦,害怕一个人睡觉。"}
]
}
```
**预期结果:** `detected=True, category=anxiety, severity=low`
---
### 语料4正常对话
```json
{
"messages": [
{"role": "system", "content": "你是一个友爱的小智AI玩偶..."},
{"role": "user", "content": "今天我画了一幅画,是一只大恐龙!"},
{"role": "assistant", "content": "哇,好厉害!是什么颜色的恐龙呀?"},
{"role": "user", "content": "是绿色的!晚上妈妈还做了我最喜欢的红烧肉,好开心呀!"}
]
}
```
**预期结果:** `detected=False`
---
### 语料5家庭矛盾
```json
{
"messages": [
{"role": "system", "content": "你是一个友爱的小智AI玩偶..."},
{"role": "user", "content": "昨天晚上爸爸妈妈吵架了,妈妈哭了"},
{"role": "assistant", "content": "爸爸妈妈有时候也会有不开心的时候呢。"},
{"role": "user", "content": "我很害怕,怕他们会离婚。我总觉得是因为我表现不好他们才吵架的。"}
]
}
```
**预期结果:** `detected=True, category=family_conflict, severity=high`
---
### 语料6自卑/自我否定
```json
{
"messages": [
{"role": "system", "content": "你是一个友爱的小智AI玩偶..."},
{"role": "user", "content": "今天老师表扬了小红但是没有表扬我"},
{"role": "assistant", "content": "没有被表扬也没关系呀,你一定也很棒的!"},
{"role": "user", "content": "可是我就是个笨蛋,什么都做不好,班上的同学都不喜欢我,没人想和我坐同桌。"}
]
}
```
**预期结果:** `detected=True, category=self_esteem, severity=medium`
---
### 语料7社交孤立
```json
{
"messages": [
{"role": "system", "content": "你是一个友爱的小智AI玩偶..."},
{"role": "user", "content": "今天课间我一个人蹲在角落"},
{"role": "assistant", "content": "为什么一个人蹲着呢?"},
{"role": "user", "content": "因为没有人来找我玩,同学们都有自己的朋友,只有我是一个人。我不想去学校了,那里好孤单。"}
]
}
```
**预期结果:** `detected=True, category=social_isolation, severity=medium`
---
### 语料8创伤事件
```json
{
"messages": [
{"role": "system", "content": "你是一个友爱的小智AI玩偶..."},
{"role": "user", "content": "前几天我看到爷爷在家里摔倒了,流了好多血"},
{"role": "assistant", "content": "爷爷后来没事吧?你一定被吓到了。"},
{"role": "user", "content": "爷爷住院了,我好害怕,每天晚上都做噩梦,梦见爷爷回不来了。"}
]
}
```
**预期结果:** `detected=True, category=trauma, severity=medium`
---
## 使用说明
回归测试时:
1. 读取本文件中的语料
2. 将 `messages` 数组传给 `PsychoScreener.screen_from_messages()`
3. 验证返回的 `ScreeningResult` 与"预期结果"一致
> 注意:`screen_from_messages()` 方法尚未实现,
> 需要在 `screener.py` 中新增,参考 `screen()` 但接受 messages 格式输入。

12
mcp_config.json Normal file
View File

@ -0,0 +1,12 @@
{
"mcpServers": {
"psycho-screener": {
"type": "stdio",
"command": "python",
"args": ["-m", "psycho_screener.mcp_tool"],
"env": {
"MINIMAX_API_KEY": "${MINIMAX_API_KEY}"
}
}
}
}

View File

@ -6,6 +6,7 @@ requires-python = ">=3.10"
dependencies = [ dependencies = [
"requests>=2.31.0", "requests>=2.31.0",
"pydantic>=2.0", "pydantic>=2.0",
"fastmcp>=2.13.0",
] ]
[project.optional-dependencies] [project.optional-dependencies]

View File

@ -0,0 +1,150 @@
"""
儿童心理陪伴 MCP 工具
基于 FastMCP注册 psycho_screen 工具供小智 AI 调用
使用方式
python -m psycho_screener.mcp_tool
前提设置 MCP_ENDPOINT 环境变量指向小智 MCP 接入点
export MCP_ENDPOINT="ws://192.168.1.25:8004/mcp_endpoint/mcp/?token=xxx"
或配合 mcp_pipe.py 使用参考 mcp_config.json
"""
from __future__ import annotations
import sys
import logging
from typing import Any
from fastmcp import FastMCP
from pydantic import Field
from .screener import PsychoScreener, ScreeningResult
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("PsychoScreenerMCP")
# ============================================================================
# FastMCP Server
# ============================================================================
mcp = FastMCP("PsychoScreener")
@mcp.tool()
def psycho_screen(
messages: list[dict] = Field(
description=(
"儿童与玩偶的完整对话上下文,格式为消息数组。"
"每条消息包含 rolesystem/user/assistant和 content内容"
"只需分析 role 为 user 的消息内容。"
)
),
include_prefix: bool = Field(
default=True,
description=(
"当检测到心理问题时,是否在返回结果中包含注入了前缀标记的文本。"
"设为 true 时,返回的 prefix 字段可用于覆盖 LLM 的回复。"
)
),
) -> dict:
"""
对儿童对话进行心理问题筛查
适用场景当儿童与玩偶对话中可能存在霸凌抑郁焦虑家庭矛盾等心理问题时
调用此工具进行筛查
返回结果包含
- detected: 是否检测到心理问题
- category: 问题类别bullying/depression/anxiety/family_conflict/self_esteem/trauma/social_isolation/none
- severity: 严重程度none/low/medium/high
- summary: 简要描述
- suggestion: 建议行动
- prefix: 如检测到问题且 include_prefix=True返回可用于注入 LLM 回复的前缀文本
注意
- 工具会自动从 messages 中提取孩子role=user的最新对话进行分析
- 请结合对话完整上下文判断不要仅凭单一消息下结论
"""
import os
api_key = os.environ.get("MINIMAX_API_KEY", "")
if not api_key:
logger.error("MINIMAX_API_KEY environment variable not set")
return {
"detected": False,
"category": "none",
"severity": "none",
"summary": "API key 未配置",
"suggestion": "",
"prefix": "",
"error": "MINIMAX_API_KEY environment variable not set",
}
try:
# 提取孩子最新说的话,构建筛查上下文
child_messages = [
msg["content"]
for msg in messages
if msg.get("role") == "user" and msg.get("content")
]
context = "\n".join(child_messages)
if not context.strip():
return {
"detected": False,
"category": "none",
"severity": "none",
"summary": "无儿童对话内容可分析",
"suggestion": "",
"prefix": "",
}
logger.info(f"Screening {len(child_messages)} child message(s)")
# 调用筛查器
screener = PsychoScreener(api_key=api_key)
result = screener.screen(context)
# 构建返回结果
response: dict[str, Any] = {
"detected": result.detected,
"category": result.category,
"severity": "none" if not result.detected else result.severity,
"summary": result.summary,
"suggestion": result.suggestion if result.detected else "",
}
# 可选:包含注入前缀
if include_prefix and result.detected:
response["prefix"] = screener.build_response_prefix(result)
logger.info(
f"Screening result: detected={result.detected}, "
f"category={result.category}, severity={result.severity}"
)
return response
except Exception as e:
logger.exception(f"Screening failed: {e}")
return {
"detected": False,
"category": "none",
"severity": "none",
"summary": f"筛查过程出错: {str(e)}",
"suggestion": "",
"prefix": "",
"error": str(e),
}
if __name__ == "__main__":
# 以 stdio 模式运行,供 mcp_pipe.py 调用
mcp.run(transport="stdio")

99
tests/test_mcp_tool.py Normal file
View File

@ -0,0 +1,99 @@
"""
MCP 工具测试
验证 psycho_screen 函数在真实 API 下的行为
"""
import os
import pytest
from psycho_screener.mcp_tool import psycho_screen
@pytest.fixture
def api_key():
key = os.environ.get("MINIMAX_API_KEY", "")
if not key:
pytest.skip("MINIMAX_API_KEY not set")
return key
# ---------------------------------------------------------------------------
# 真实 API 测试
# ---------------------------------------------------------------------------
class TestPsychoScreenTool:
"""测试 MCP 工具函数 psycho_screen"""
def test_bullying_conversation(self, api_key):
"""霸凌场景:多轮对话,孩子最新消息包含霸凌内容"""
messages = [
{"role": "system", "content": "你是一个友好的小智AI玩偶..."},
{"role": "user", "content": "今天幼儿园有个小朋友抢了我的玩具"},
{"role": "assistant", "content": "哎呀,那真是太过分了!"},
{"role": "user", "content": "他还不让我告诉老师,他说如果我告诉老师就会打我。我好害怕。"},
]
result = psycho_screen(messages, include_prefix=True)
print(f"\n[tool] bullying result: {result}")
assert result["detected"] is True
assert result["category"] == "bullying"
assert result["severity"] in ("medium", "high")
assert "已发现特定心理问题" in result["prefix"]
def test_normal_conversation(self, api_key):
"""正常对话:全程无异常"""
messages = [
{"role": "system", "content": "你是一个友好的小智AI玩偶..."},
{"role": "user", "content": "今天我画了一幅画,是一只大恐龙!"},
{"role": "assistant", "content": "哇,好厉害!"},
{"role": "user", "content": "是绿色的!晚上妈妈还做了红烧肉,好开心!"},
]
result = psycho_screen(messages, include_prefix=False)
print(f"\n[tool] normal result: {result}")
assert result["detected"] is False
assert result["category"] == "none"
assert result["severity"] == "none"
def test_no_prefix_when_not_detected(self, api_key):
"""未检测到问题时prefix 应为空"""
messages = [
{"role": "user", "content": "今天天气真好呀!"},
]
result = psycho_screen(messages, include_prefix=True)
assert result["detected"] is False
assert result.get("prefix", "") == ""
def test_empty_messages(self, api_key):
"""空消息列表"""
result = psycho_screen([], include_prefix=True)
assert result["detected"] is False
assert result["summary"] == "无儿童对话内容可分析"
def test_messages_without_child_content(self, api_key):
"""只有 system 消息,无 user 消息"""
messages = [
{"role": "system", "content": "你是一个友好的小智AI玩偶..."},
]
result = psycho_screen(messages, include_prefix=True)
assert result["detected"] is False
def test_api_key_missing(self):
"""API key 未配置时的行为"""
# 临时清除 API key
original = os.environ.pop("MINIMAX_API_KEY", None)
result = psycho_screen(
[{"role": "user", "content": "测试"}],
include_prefix=True
)
# 恢复原值
if original:
os.environ["MINIMAX_API_KEY"] = original
assert result["detected"] is False
assert "error" in result