hz_score_items
Score content items using AI to evaluate and categorize data in the Horizon pipeline, enabling automated quality assessment and structured analysis.
Instructions
对指定阶段内容执行 AI 打分，写入 scored 阶段。
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| run_id | Yes | ||
| source_stage | No | raw | |
| horizon_path | No | ||
| config_path | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- horizon_mcp/service.py:249-290 (handler)Implementation of the 'score_items' service method which is invoked by the 'hz_score_items' MCP tool.
async def score_items( self, run_id: str, source_stage: str = "raw", horizon_path: str | None = None, config_path: str | None = None, ) -> dict[str, Any]: items, ctx = self._load_stage_items( run_id=run_id, stage=source_stage, horizon_path=horizon_path, config_path=config_path, ) if not items: raise HorizonMcpError(code="HZ_EMPTY_INPUT", message="待评分内容为空。") ai_client = ctx.runtime.create_ai_client(ctx.config.ai) analyzer = ctx.runtime.ContentAnalyzer(ai_client) scored_items = await analyzer.analyze_batch(items) self.run_store.save_items(run_id, "scored", items_to_dicts(scored_items)) score_threshold = ctx.config.filtering.ai_score_threshold above_threshold = [x for x in scored_items if x.ai_score and x.ai_score >= score_threshold] meta = self.run_store.update_meta( run_id, { "scored_count": len(scored_items), "scored_threshold": score_threshold, "scored_above_threshold": len(above_threshold), }, ) return { "run_id": run_id, "scored": len(scored_items), "above_threshold": len(above_threshold), "score_distribution": self._score_distribution(scored_items), "artifact": str((self.run_store.run_dir(run_id) / "scored_items.json").resolve()), "meta": meta, } - horizon_mcp/server.py:173-190 (registration)MCP tool registration for 'hz_score_items', mapping the tool name to the service.score_items handler.
@mcp.tool() async def hz_score_items( run_id: str, source_stage: str = "raw", horizon_path: str | None = None, config_path: str | None = None, ) -> dict[str, Any]: """对指定阶段内容执行 AI 打分，写入 scored 阶段。""" return await _run_tool( "hz_score_items", lambda: service.score_items( run_id=run_id, source_stage=source_stage, horizon_path=horizon_path, config_path=config_path, ), )