|
|
@@ -0,0 +1,539 @@
|
|
|
+"""主通风机固定工作流：完整 PDF 提取 + 现成监测数据 → 一次 Qwen JSON 研判。"""
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import asyncio
|
|
|
+from datetime import datetime
|
|
|
+import json
|
|
|
+import math
|
|
|
+from pathlib import Path
|
|
|
+import time
|
|
|
+from typing import Literal
|
|
|
+from uuid import uuid4
|
|
|
+from zoneinfo import ZoneInfo
|
|
|
+
|
|
|
+import httpx
|
|
|
+from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator
|
|
|
+
|
|
|
+import config
|
|
|
+from agents.model_factory import _is_qwen35_model, _qwen_thinking_payload, normalize_thinking
|
|
|
+from billing.meter import UsageMeter
|
|
|
+from db import business_store, chat_store, db_conn, main_fan_store
|
|
|
+from extensions.main_fan import exploration, history_compaction, history_processing, workflow_json_repair, workflow_monitoring
|
|
|
+from extensions.main_fan.batch_context import scope_for
|
|
|
+from extensions.main_fan.schemas import MainFanResult
|
|
|
+from extensions.main_fan.workflow_documents import OCRSettings, PDFExtractor, service_url
|
|
|
+from extensions.main_fan.workflow_page_filter import select_pages
|
|
|
+from extensions.registry import load_extensions
|
|
|
+from obs import get_logger, get_trace_id, log_context
|
|
|
+from runtime import business, execution
|
|
|
+from runtime.context_meter import estimate_tokens
|
|
|
+from runtime.engine import ExecutionResult
|
|
|
+from runtime.python_sandbox import digest, workspace_file
|
|
|
+from runtime.result_models import PlatformJudgment, PlatformResultPayload
|
|
|
+
|
|
|
+logger = get_logger(__name__)
|
|
|
+VERSION = 'main-fan-workflow-v5'
|
|
|
+LOCAL_ZONE = ZoneInfo('Asia/Shanghai')
|
|
|
+JUDGMENT_POLICY = {
|
|
|
+ 'version': 'main-fan-precision-first-v2-monitoring-rules',
|
|
|
+ 'priority': 'precision',
|
|
|
+ 'monitoring': '固定代码按本轮冻结规则生成monitoring_assessment；已确认的监测告警不能被模型降级，不增加稳定窗口等新条件。',
|
|
|
+ 'performance': '辅助性粗略核对；只有清晰、可比且确凿的异常证据才告警，识别不清或资料不足只说明限制。',
|
|
|
+ 'normal': '有可用数据且未发现确凿异常；不表示未核实项目已被证明合格或安全。',
|
|
|
+ 'undetermined': '仅当报告、开停、风量三类均没有可用数据。',
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+class WorkflowJudgment(PlatformJudgment):
|
|
|
+ # 让请求 schema 与本地约束一致；本地文件没有公开证据链接。
|
|
|
+ evidenceLink: Literal[''] = Field(...)
|
|
|
+
|
|
|
+
|
|
|
+class WorkflowDecision(BaseModel):
|
|
|
+ """模型填写业务内容；平台身份、任务标识和完成时刻由宿主填写。"""
|
|
|
+ model_config = ConfigDict(extra='forbid', strict=True)
|
|
|
+ objectDetail: str = Field(min_length=1)
|
|
|
+ dataDetail: str
|
|
|
+ judgment: WorkflowJudgment
|
|
|
+
|
|
|
+ @field_validator('objectDetail')
|
|
|
+ @classmethod
|
|
|
+ def nonblank_object(cls, value):
|
|
|
+ if not value.strip():
|
|
|
+ raise ValueError('研判对象不能为空白')
|
|
|
+ return value
|
|
|
+
|
|
|
+ @field_validator('judgment')
|
|
|
+ @classmethod
|
|
|
+ def complete_judgment(cls, value):
|
|
|
+ if not value.result.strip() or not value.reason.strip():
|
|
|
+ raise ValueError('研判摘要和正文不能为空白')
|
|
|
+ if value.evidenceLink:
|
|
|
+ raise ValueError('本地工作流没有公开证据页，evidenceLink 须为空字符串')
|
|
|
+ return value
|
|
|
+
|
|
|
+
|
|
|
+class WorkflowError(ValueError):
|
|
|
+ def __init__(self, code, message, *, details=None):
|
|
|
+ self.code = code
|
|
|
+ self.details = details or {}
|
|
|
+ super().__init__(message)
|
|
|
+
|
|
|
+
|
|
|
+def publish_enabled():
|
|
|
+ # 新工作流默认用于服务器本地核验；不能因为沿用旧环境的总开关而直接上报。
|
|
|
+ return (config.get_bool('MAIN_FAN_WORKFLOW_PUBLISH_ENABLED', False)
|
|
|
+ and config.get_bool('RESULT_PLATFORM_ENABLED', False))
|
|
|
+
|
|
|
+
|
|
|
+def qwen_http_client():
|
|
|
+ return httpx.AsyncClient(trust_env=False)
|
|
|
+
|
|
|
+
|
|
|
+def full_monitoring(context, user, session):
|
|
|
+ prepared = context['prepared_data']
|
|
|
+ compact = exploration._read_json(workspace_file(user, session, prepared['agent_data_path']))
|
|
|
+ details_path = compact.get('details_path')
|
|
|
+ if details_path:
|
|
|
+ if details_path not in context.get('query_exports', {}):
|
|
|
+ raise ValueError('监测详情不属于本轮已核验输入')
|
|
|
+ details = exploration._read_json(workspace_file(user, session, details_path))
|
|
|
+ # 不重新取数或计算切换/比例，只展开固定预处理已经产生的全部结果。
|
|
|
+ view = history_compaction.agent_view(details, details_path, full=True)
|
|
|
+ compact = history_processing.agent_input(view)
|
|
|
+ compact['state_period_semantics'] = view['state_period_semantics']
|
|
|
+ if compact.get('switch_candidates_truncated') or any(row.get('truncated') for row in compact['states']):
|
|
|
+ raise ValueError('工作流需要完整监测结果，不能将截断预览交给唯一一次模型调用')
|
|
|
+ return compact
|
|
|
+
|
|
|
+
|
|
|
+def build_input(workflow, monitoring, reports, scope, *, availability=None, monitoring_assessment=None):
|
|
|
+ context = workflow['context']
|
|
|
+ availability = data_availability(reports, monitoring) if availability is None else availability
|
|
|
+ monitoring_assessment = (workflow_monitoring.assess(monitoring, context['policy'])
|
|
|
+ if monitoring_assessment is None else monitoring_assessment)
|
|
|
+ references = [{'id': row['id'], 'from_fan': row['from_fan'], 'to_fan': row['to_fan'],
|
|
|
+ 'flow': row.get('flow', []), 'comparison': row.get('comparison', {})}
|
|
|
+ for row in monitoring['switch_candidates']]
|
|
|
+ return {'workflow_version': VERSION, 'workflow_id': workflow['id'],
|
|
|
+ 'identity': {'mine_code': scope.mine['mine_id'], 'mine_name': scope.mine['mine_name'],
|
|
|
+ 'dept_id': scope.mine['dept_id'], 'shaft': scope.shaft.model_dump(mode='json'),
|
|
|
+ 'fan_pair': context['task_input'].get('fan_pair', [])},
|
|
|
+ 'rules': context['policy'],
|
|
|
+ 'judgment_policy': JUDGMENT_POLICY,
|
|
|
+ 'data_availability': availability,
|
|
|
+ 'monitoring_assessment': monitoring_assessment,
|
|
|
+ 'pdf': reports,
|
|
|
+ 'airflow': {'range': monitoring['range'], 'flows': monitoring['flows'],
|
|
|
+ 'switch_references': references, 'notes': monitoring.get('notes', [])},
|
|
|
+ 'switching': {'range': monitoring['range'], 'states': monitoring['states'],
|
|
|
+ 'candidates': [{**row, 'duration_minutes': row.get('duration_minutes'),
|
|
|
+ 'duration_status': row.get('duration_status') or (
|
|
|
+ 'unknown' if row.get('duration_minutes') is None else 'provided')}
|
|
|
+ for row in monitoring['switch_candidates']],
|
|
|
+ 'time_range_semantics': '整个候选的采样定位范围，不用起止相减；直接采用duration_minutes及其采样上下界。'
|
|
|
+ 'estimated表示按实际开停变化观测估算，只有无法配对时间时才为unknown。',
|
|
|
+ 'notes': monitoring.get('notes', []),
|
|
|
+ 'state_period_semantics': monitoring.get('state_period_semantics', '')},
|
|
|
+ 'history_coverage': context.get('history_manifest', {}) and {
|
|
|
+ key: context['history_manifest'][key] for key in ('cutoff', 'lookback_days', 'point_count',
|
|
|
+ 'expected_segments', 'completed_segments', 'failed_segments', 'coverage_complete')},
|
|
|
+ 'output_schema': decision_schema(availability, monitoring_assessment)}
|
|
|
+
|
|
|
+
|
|
|
+def decision_schema(availability=None, monitoring_assessment=None):
|
|
|
+ schema = WorkflowDecision.model_json_schema()
|
|
|
+ if availability is not None:
|
|
|
+ schema['$defs']['WorkflowJudgment']['properties']['resultCode']['enum'] = (
|
|
|
+ ['NORMAL', 'WARNING', 'ABNORMAL'] if availability['any_available'] else ['UNDETERMINED'])
|
|
|
+ if monitoring_assessment and monitoring_assessment['minimum_result_code'] != 'NORMAL':
|
|
|
+ minimum = workflow_monitoring.RANK[monitoring_assessment['minimum_result_code']]
|
|
|
+ schema['$defs']['WorkflowJudgment']['properties']['resultCode']['enum'] = [
|
|
|
+ code for code, rank in workflow_monitoring.RANK.items() if rank >= minimum]
|
|
|
+ return schema
|
|
|
+
|
|
|
+
|
|
|
+def response_format(availability=None, monitoring_assessment=None):
|
|
|
+ return {'type': 'json_schema', 'json_schema': {'name': 'main_fan_workflow_decision',
|
|
|
+ 'strict': True, 'schema': decision_schema(availability, monitoring_assessment)}}
|
|
|
+
|
|
|
+
|
|
|
+def data_availability(reports, monitoring):
|
|
|
+ def finite(value):
|
|
|
+ return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)
|
|
|
+ readable_pages = sum(page['status'] == 'completed' and bool(page.get('text', '').strip())
|
|
|
+ for report in reports if not report.get('is_mock') for page in report['pages'])
|
|
|
+ known_states = {'开', '停', '开机', '停机', '运行', '停止', 'running', 'stopped', 'on', 'off'}
|
|
|
+ state_periods = sum(str(period.get('state', '')).strip().lower() in known_states
|
|
|
+ for point in monitoring.get('states', []) for period in point.get('periods', []))
|
|
|
+ candidates = monitoring.get('switch_candidates', [])
|
|
|
+ durations = sum(finite(row.get('duration_minutes')) and row['duration_minutes'] >= 0 for row in candidates)
|
|
|
+ samples = sum(max(0, row['samples']) for row in monitoring.get('flows', []) if finite(row.get('samples')))
|
|
|
+ values = sum(finite(value) for point in monitoring.get('flows', []) for value in point.get('range', []))
|
|
|
+ values += sum(finite(row.get('comparison', {}).get(key)) for row in candidates
|
|
|
+ for key in ('before_value', 'after_value', 'change_percent'))
|
|
|
+ values += sum(isinstance(sample, (list, tuple)) and len(sample) == 2 and finite(sample[1])
|
|
|
+ for row in candidates for point in row.get('flow', [])
|
|
|
+ for sample in (point.get('before'), point.get('after')))
|
|
|
+ comparisons = sum(all(finite(row.get('comparison', {}).get(key))
|
|
|
+ for key in ('before_value', 'after_value', 'change_percent')) for row in candidates)
|
|
|
+ return {'pdf_readable_pages': readable_pages, 'known_state_periods': state_periods,
|
|
|
+ 'known_switch_intervals': durations, 'flow_numeric_samples': samples, 'flow_numeric_values': values,
|
|
|
+ 'flow_comparisons': comparisons,
|
|
|
+ 'any_available': bool(readable_pages or state_periods or durations or samples or values)}
|
|
|
+
|
|
|
+
|
|
|
+def validate_result_policy(decision, availability):
|
|
|
+ undetermined = decision.judgment.resultCode == 'UNDETERMINED'
|
|
|
+ if undetermined != (not availability['any_available']):
|
|
|
+ raise WorkflowError('workflow_result_policy',
|
|
|
+ 'UNDETERMINED 仅适用于三类数据均不可用；有可用数据时按明确异常证据判定，资料局部缺失只说明限制',
|
|
|
+ details={'data_availability': availability, 'result_code': decision.judgment.resultCode})
|
|
|
+
|
|
|
+
|
|
|
+def parse_decision(content, *, source_name='qwen-response.json'):
|
|
|
+ if not isinstance(content, str) or not content.strip():
|
|
|
+ raise WorkflowError('qwen_empty_response', 'Qwen 没有返回研判 JSON，原始响应已保存')
|
|
|
+ value = workflow_json_repair.unfenced(content)
|
|
|
+ def invalid_number(number):
|
|
|
+ raise ValueError('研判 JSON 不接受非有限数值：' + number)
|
|
|
+ try:
|
|
|
+ raw = json.loads(value, parse_constant=invalid_number)
|
|
|
+ except json.JSONDecodeError as exc:
|
|
|
+ raise WorkflowError('qwen_invalid_json',
|
|
|
+ f'Qwen JSON 语法错误：{exc.msg}，第 {exc.lineno} 行第 {exc.colno} 列（偏移 {exc.pos}）；原文见 {source_name}',
|
|
|
+ details={'error_type': 'JSONDecodeError', 'line': exc.lineno, 'column': exc.colno,
|
|
|
+ 'offset': exc.pos, 'message': exc.msg}) from exc
|
|
|
+ except ValueError as exc:
|
|
|
+ raise WorkflowError('qwen_invalid_json', str(exc), details={'error_type': type(exc).__name__}) from exc
|
|
|
+ try:
|
|
|
+ return WorkflowDecision.model_validate(raw)
|
|
|
+ except ValidationError as exc:
|
|
|
+ # 不把模型正文或整份输入塞入日志，只记录字段位置和约束错误。
|
|
|
+ errors = exc.errors(include_input=False, include_context=False, include_url=False)
|
|
|
+ raise WorkflowError('qwen_schema_validation',
|
|
|
+ f'Qwen JSON 可解析，但字段不符合研判契约；字段错误见 diagnostics.error_details，原文见 {source_name}',
|
|
|
+ details={'error_type': 'ValidationError', 'errors': errors[:8]}) from exc
|
|
|
+
|
|
|
+
|
|
|
+def build_payload(decision, workflow, scope):
|
|
|
+ binding = load_extensions().business('main_fan').result_binding
|
|
|
+ task_id = workflow['context']['job_id'] or workflow['id']
|
|
|
+ return PlatformResultPayload(agentCode=binding.agent_code, agentName=binding.agent_name,
|
|
|
+ taskId=task_id, checkedAt=datetime.now(LOCAL_ZONE).strftime('%Y-%m-%d %H:%M:%S'),
|
|
|
+ mineCode=scope.mine['mine_id'], mineName=scope.mine['mine_name'],
|
|
|
+ **decision.model_dump(mode='json')).model_dump(mode='json')
|
|
|
+
|
|
|
+
|
|
|
+async def execute(*, workflow_id, user_name, session_id, progress=None, qwen_client=None, ocr_client=None):
|
|
|
+ """一次业务研判；仅失败扫描页补识别和不合格 JSON 格式修复允许额外调用。"""
|
|
|
+ scope = scope_for(user_name, session_id)
|
|
|
+ workflow = exploration._load(workflow_id, user_name, session_id, allow_completed=True)
|
|
|
+ if workflow['status'] == 'completed':
|
|
|
+ if (workflow.get('result') or {}).get('assessment', {}).get('mode') != VERSION:
|
|
|
+ raise ValueError('旧执行版本结果不能冒充本版工作流结果，请重新研判')
|
|
|
+ result = workflow['result']
|
|
|
+ diagnostics = {**result['assessment'].get('diagnostics', {}), 'reused_result': True, 'qwen_calls_this_attempt': 0}
|
|
|
+ return ExecutionResult(workflow['context'].get('assessment_workflow', {}).get('run_id', ''),
|
|
|
+ session_id, 'success', result['result'], '', {}, result.get('sources', []), [], result, diagnostics)
|
|
|
+ context = workflow['context']
|
|
|
+ admission = None
|
|
|
+ run_id = uuid4().hex
|
|
|
+ started = time.monotonic()
|
|
|
+ prefix = '/workspace/exploration/' + workflow_id + '/workflow'
|
|
|
+ meters = {}
|
|
|
+ diagnostics = {'execution_mode': VERSION, 'qwen_calls': 0, 'judgment_calls': 0,
|
|
|
+ 'qwen_ocr_calls': 0, 'json_repair_calls': 0, 'tool_calls': 0}
|
|
|
+ artifacts = {}
|
|
|
+ stage_name = '初始化工作流'
|
|
|
+
|
|
|
+ def write(name, value):
|
|
|
+ virtual = prefix + '/' + name
|
|
|
+ path = workspace_file(user_name, session_id, virtual)
|
|
|
+ exploration._write(path, value)
|
|
|
+ context.setdefault('query_exports', {})[virtual] = digest(path)
|
|
|
+ artifacts[name] = virtual
|
|
|
+ return virtual
|
|
|
+
|
|
|
+ def stage(name, status, result=None):
|
|
|
+ nonlocal stage_name
|
|
|
+ stage_name = name
|
|
|
+ rows = workflow['stages']
|
|
|
+ key = 'workflow:' + name
|
|
|
+ found = next((row for row in rows if row['name'] == key), None)
|
|
|
+ value = {'name': key, 'status': status, 'result': result}
|
|
|
+ if found is None:
|
|
|
+ rows.append(value)
|
|
|
+ else:
|
|
|
+ found.clear(); found.update(value)
|
|
|
+ context['assessment_workflow'] = {'version': VERSION, 'run_id': run_id,
|
|
|
+ 'stage': name, 'status': status, 'diagnostics': diagnostics, 'files': artifacts}
|
|
|
+ main_fan_store.update_workflow(workflow_id, context=context, stages=rows)
|
|
|
+ if progress:
|
|
|
+ progress({'stage': name, 'status': status, 'workflow_id': workflow_id})
|
|
|
+ logger.info('研判工作流阶段', extra={'kv': {'event': 'workflow.stage', 'stage': name,
|
|
|
+ 'status': status, 'workflow_id': workflow_id, 'run_id': run_id}})
|
|
|
+
|
|
|
+ def record_usage(model, raw):
|
|
|
+ meter = meters.setdefault(model, UsageMeter(user_name, session_id, 'business', model,
|
|
|
+ trace_id=get_trace_id(), usage_id=run_id if model == config.MODEL else None))
|
|
|
+ if raw:
|
|
|
+ meter.add(raw.get('prompt_tokens', 0), raw.get('completion_tokens', 0))
|
|
|
+
|
|
|
+ def record_qwen_call(kind):
|
|
|
+ diagnostics['qwen_calls'] += 1
|
|
|
+ diagnostics[kind] += 1
|
|
|
+
|
|
|
+ def usage():
|
|
|
+ rows = {name: {'input_tokens': meter.input_tokens, 'output_tokens': meter.output_tokens,
|
|
|
+ 'total_tokens': meter.total_tokens, 'usage_source': meter.usage_source}
|
|
|
+ for name, meter in meters.items()}
|
|
|
+ return {'input_tokens': sum(row['input_tokens'] for row in rows.values()),
|
|
|
+ 'output_tokens': sum(row['output_tokens'] for row in rows.values()),
|
|
|
+ 'total_tokens': sum(row['total_tokens'] for row in rows.values()), 'models': rows}
|
|
|
+
|
|
|
+ with log_context(run_id=run_id, workflow_id=workflow_id, execution_mode=VERSION):
|
|
|
+ try:
|
|
|
+ admission = await execution.admit(user_name, session_id)
|
|
|
+ admission.business_lease = business.BusinessLease('main_fan')
|
|
|
+ admission.business_lease.bind()
|
|
|
+ await asyncio.to_thread(business_store.bind_session, session_id, 'main_fan')
|
|
|
+ exploration.validate_inputs(workflow, user_name, session_id)
|
|
|
+ if not _is_qwen35_model(config.MODEL):
|
|
|
+ raise WorkflowError('workflow_model_config', '本工作流请将 DEEPAGENT_MODEL 配置为实际 Qwen3.5 服务模型名')
|
|
|
+ endpoint = service_url(config.OPENAI_BASE_URL)
|
|
|
+ maximum = config.get_int('MAIN_FAN_WORKFLOW_MAX_TOKENS', 16384)
|
|
|
+ timeout = config.get_int('MAIN_FAN_WORKFLOW_TIMEOUT_SECONDS', 600)
|
|
|
+ repair_limit = config.get_int('MAIN_FAN_JSON_REPAIR_MAX_ATTEMPTS', workflow_json_repair.MAX_ATTEMPTS)
|
|
|
+ if maximum <= 0 or timeout <= 0:
|
|
|
+ raise ValueError('工作流输出长度及超时必须大于 0')
|
|
|
+ if not 0 <= repair_limit <= workflow_json_repair.MAX_ATTEMPTS:
|
|
|
+ raise ValueError('JSON 格式修复次数须为 0～3，不能超过 3 次')
|
|
|
+ ocr_settings = OCRSettings.from_config()
|
|
|
+ stage('PDF逐页提取', 'running')
|
|
|
+ extractor = PDFExtractor(user_name, session_id, prefix, scope.cache_root,
|
|
|
+ settings=ocr_settings, client=ocr_client, qwen_client=qwen_client, progress=progress,
|
|
|
+ cache_policy=context['task_input'].get('document_cache_policy', 'auto'),
|
|
|
+ on_usage=lambda raw: record_usage(ocr_settings.model, raw),
|
|
|
+ on_qwen_usage=lambda raw: record_usage(config.MODEL, raw),
|
|
|
+ on_qwen_call=lambda: record_qwen_call('qwen_ocr_calls'))
|
|
|
+ materials = context['prepared_data']['reports']['reports']
|
|
|
+ reports = await extractor.extract(materials)
|
|
|
+ context.setdefault('query_exports', {}).update(extractor.files)
|
|
|
+ write('pdf-extraction.json', reports)
|
|
|
+ diagnostics.update(pdf_count=len(reports), pdf_pages=sum(row['page_count'] for row in reports),
|
|
|
+ failed_pages=sum(page['status'] != 'completed' for row in reports for page in row['pages']))
|
|
|
+ stage('PDF逐页提取', 'completed', {key: diagnostics[key] for key in ('pdf_count', 'pdf_pages', 'failed_pages')})
|
|
|
+
|
|
|
+ stage('过滤明确无关PDF页', 'running')
|
|
|
+ selected_reports, selection = select_pages(reports)
|
|
|
+ write('pdf-page-selection.json', selection)
|
|
|
+ diagnostics.update(retained_pdf_pages=sum(len(row['pages']) for row in selected_reports),
|
|
|
+ excluded_pdf_pages=sum(len(row['excluded_pages']) for row in selection))
|
|
|
+ stage('过滤明确无关PDF页', 'completed', {key: diagnostics[key]
|
|
|
+ for key in ('retained_pdf_pages', 'excluded_pdf_pages')})
|
|
|
+
|
|
|
+ stage('合并本轮三类输入', 'running')
|
|
|
+ monitoring = full_monitoring(context, user_name, session_id)
|
|
|
+ availability = data_availability(reports, monitoring)
|
|
|
+ stage('监测阈值固定判定', 'running')
|
|
|
+ monitoring_assessment = workflow_monitoring.assess(monitoring, context['policy'])
|
|
|
+ write('monitoring-assessment.json', monitoring_assessment)
|
|
|
+ diagnostics['monitoring_result_code'] = monitoring_assessment['minimum_result_code']
|
|
|
+ stage('监测阈值固定判定', 'completed')
|
|
|
+ inputs = build_input(workflow, monitoring, selected_reports, scope, availability=availability,
|
|
|
+ monitoring_assessment=monitoring_assessment)
|
|
|
+ diagnostics['data_availability'] = availability
|
|
|
+ diagnostics['judgment_policy'] = JUDGMENT_POLICY['version']
|
|
|
+ write('qwen-input.json', inputs)
|
|
|
+ system_prompt = Path(__file__).with_name('workflow_prompt.md').read_text(encoding='utf-8')
|
|
|
+ messages = [{'role': 'system', 'content': system_prompt},
|
|
|
+ {'role': 'user', 'content': json.dumps(inputs, ensure_ascii=False, allow_nan=False)}]
|
|
|
+ payload = {'model': config.MODEL, 'messages': messages, 'stream': False,
|
|
|
+ 'max_tokens': maximum, 'response_format': response_format(availability, monitoring_assessment),
|
|
|
+ **_qwen_thinking_payload(normalize_thinking(None))}
|
|
|
+ stage('合并本轮三类输入', 'completed')
|
|
|
+
|
|
|
+ headers = {'Authorization': 'Bearer ' + config.OPENAI_API_KEY} if config.OPENAI_API_KEY else {}
|
|
|
+ async def invoke(client, request, label, kind):
|
|
|
+ # 无 HTTP 自动重试；补识别、格式修复均有明确计数和独立原始响应。
|
|
|
+ write(label + '-request.json', request)
|
|
|
+ record_qwen_call(kind)
|
|
|
+ response = await client.post(endpoint + '/chat/completions', json=request, headers=headers, timeout=timeout)
|
|
|
+ name = label + '-response.json'
|
|
|
+ virtual = prefix + '/' + name
|
|
|
+ local = workspace_file(user_name, session_id, virtual)
|
|
|
+ local.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ local.write_text(response.text, encoding='utf-8')
|
|
|
+ context['query_exports'][virtual] = digest(local)
|
|
|
+ artifacts[name] = virtual
|
|
|
+ diagnostics['qwen_http_status'] = response.status_code
|
|
|
+ if response.is_error:
|
|
|
+ raise WorkflowError('qwen_http_error', f'Qwen HTTP {response.status_code}，原始错误见 {name}')
|
|
|
+ raw = response.json()
|
|
|
+ if not isinstance(raw, dict):
|
|
|
+ raise WorkflowError('qwen_unexpected_response', f'Qwen 响应结构无效，详见 {name}')
|
|
|
+ choice = raw['choices'][0]
|
|
|
+ if not isinstance(choice, dict):
|
|
|
+ raise WorkflowError('qwen_unexpected_response', f'Qwen 候选消息结构无效，详见 {name}')
|
|
|
+ message = choice['message']
|
|
|
+ if not isinstance(message, dict):
|
|
|
+ raise WorkflowError('qwen_unexpected_response', f'Qwen 消息结构无效，详见 {name}')
|
|
|
+ content = message.get('content')
|
|
|
+ if raw.get('usage'):
|
|
|
+ record_usage(config.MODEL, raw['usage'])
|
|
|
+ else:
|
|
|
+ record_usage(config.MODEL, {})
|
|
|
+ meters[config.MODEL].add(estimate_tokens(json.dumps(request['messages'], ensure_ascii=False)),
|
|
|
+ estimate_tokens((content if isinstance(content, str) else '')
|
|
|
+ + str(message.get('reasoning') or message.get('reasoning_content') or '')), 'estimate')
|
|
|
+ if kind == 'judgment_calls':
|
|
|
+ diagnostics.update(response_id=raw.get('id'), finish_reason=choice.get('finish_reason'))
|
|
|
+ if choice.get('finish_reason') == 'length':
|
|
|
+ raise WorkflowError('qwen_response_truncated', f'Qwen 输出达到长度上限，不能补造丢失内容；详见 {name}')
|
|
|
+ if message.get('tool_calls') or choice.get('finish_reason') not in {'stop', None}:
|
|
|
+ raise WorkflowError('qwen_unexpected_response', f'Qwen 未返回正常结束的正文，详见 {name}')
|
|
|
+ return raw, content
|
|
|
+
|
|
|
+ async def judge(client):
|
|
|
+ stage('Qwen一次研判', 'running')
|
|
|
+ original_raw, original = await invoke(client, payload, 'qwen', 'judgment_calls')
|
|
|
+ stage('Qwen一次研判', 'completed', {'response_id': original_raw.get('id'), 'usage': original_raw.get('usage')})
|
|
|
+ stage('JSON校验与格式修复', 'running')
|
|
|
+ attempts = []
|
|
|
+ write('json-repair-history.json', attempts)
|
|
|
+ try:
|
|
|
+ decision = parse_decision(original)
|
|
|
+ except WorkflowError as exc:
|
|
|
+ if not repair_limit or not workflow_json_repair.can_repair(original, exc.code):
|
|
|
+ raise
|
|
|
+ error = {'error_code': exc.code, 'message': str(exc), 'details': exc.details}
|
|
|
+ else:
|
|
|
+ write('qwen-final-response.json', original_raw)
|
|
|
+ return decision
|
|
|
+ current = original
|
|
|
+ for attempt in range(1, repair_limit + 1):
|
|
|
+ label = f'qwen-repair-{attempt}'
|
|
|
+ entry = {'attempt': attempt, 'request': prefix + '/' + label + '-request.json',
|
|
|
+ 'response': prefix + '/' + label + '-response.json', 'status': 'running'}
|
|
|
+ attempts.append(entry)
|
|
|
+ write('json-repair-history.json', attempts)
|
|
|
+ logger.warning('研判 JSON 未通过校验，进行格式修复', extra={'kv': {
|
|
|
+ 'event': 'workflow.json.repair.started', 'attempt': attempt, 'max_attempts': repair_limit,
|
|
|
+ 'error_code': error['error_code'], 'error_details': error['details']}})
|
|
|
+ if progress:
|
|
|
+ progress({'stage': 'json_format_repair', 'attempt': attempt, 'max_attempts': repair_limit})
|
|
|
+ repair_request = {'model': config.MODEL, 'stream': False, 'max_tokens': maximum,
|
|
|
+ # 格式修复保留原编码；随后由固定监测规则统一合并，修复器不能改业务内容。
|
|
|
+ 'messages': workflow_json_repair.messages(current, decision_schema(availability), error),
|
|
|
+ 'response_format': response_format(availability), **_qwen_thinking_payload(normalize_thinking(None))}
|
|
|
+ repaired = None
|
|
|
+ try:
|
|
|
+ repaired_raw, repaired = await invoke(client, repair_request, label, 'json_repair_calls')
|
|
|
+ candidate = parse_decision(repaired, source_name=label + '-response.json')
|
|
|
+ changed = workflow_json_repair.preservation_errors(original, candidate)
|
|
|
+ if changed:
|
|
|
+ current = original
|
|
|
+ raise WorkflowError('qwen_repair_content_changed', '格式修复改动了业务内容，不能接受',
|
|
|
+ details={'changed_content': changed})
|
|
|
+ except (WorkflowError, httpx.HTTPError, ValueError, KeyError, IndexError, TypeError) as exc:
|
|
|
+ error = {'error_code': exc.code if isinstance(exc, WorkflowError) else type(exc).__name__,
|
|
|
+ 'message': str(exc)[:500], 'details': getattr(exc, 'details', {})}
|
|
|
+ entry.update(status='failed', error=error)
|
|
|
+ # 继续修复上一轮字符串；最终始终与最初研判核对内容。
|
|
|
+ if isinstance(exc, WorkflowError) and exc.code in {'qwen_invalid_json', 'qwen_schema_validation'}:
|
|
|
+ if isinstance(repaired, str):
|
|
|
+ current = repaired
|
|
|
+ write('json-repair-history.json', attempts)
|
|
|
+ continue
|
|
|
+ entry.update(status='completed', response_id=repaired_raw.get('id'))
|
|
|
+ write('json-repair-history.json', attempts)
|
|
|
+ write('qwen-final-response.json', repaired_raw)
|
|
|
+ diagnostics['json_repaired'] = True
|
|
|
+ return candidate
|
|
|
+ raise WorkflowError('qwen_json_repair_exhausted',
|
|
|
+ f'JSON 格式修复 {repair_limit} 次仍未通过，原始响应及每次修复记录已保存',
|
|
|
+ details={'repair_attempts': repair_limit, 'last_error': error})
|
|
|
+
|
|
|
+ if qwen_client is None:
|
|
|
+ async with qwen_http_client() as client:
|
|
|
+ decision = await judge(client)
|
|
|
+ else:
|
|
|
+ decision = await judge(qwen_client)
|
|
|
+ stage('JSON校验与格式修复', 'completed', {'repair_attempts': diagnostics['json_repair_calls']})
|
|
|
+
|
|
|
+ stage('校验并保存平台JSON', 'running')
|
|
|
+ resolved, resolution = workflow_monitoring.resolve(decision.model_dump(mode='json'), monitoring_assessment)
|
|
|
+ decision = WorkflowDecision.model_validate(resolved)
|
|
|
+ write('decision-resolution.json', resolution)
|
|
|
+ diagnostics['monitoring_enforced'] = resolution['monitoring_enforced']
|
|
|
+ diagnostics['model_result_code'] = resolution['model_result_code']
|
|
|
+ if resolution['monitoring_enforced']:
|
|
|
+ logger.info('固定监测规则生成正式告警结论', extra={'kv': {
|
|
|
+ 'event': 'workflow.monitoring.enforced', 'model_result_code': resolution['model_result_code'],
|
|
|
+ 'minimum_result_code': resolution['minimum_result_code'], 'final_result_code': resolution['final_result_code']}})
|
|
|
+ validate_result_policy(decision, availability)
|
|
|
+ exploration.validate_inputs(workflow, user_name, session_id)
|
|
|
+ platform = build_payload(decision, workflow, scope)
|
|
|
+ write('platform-payload.json', platform)
|
|
|
+ write('usage.json', usage())
|
|
|
+ body = '# ' + scope.mine['mine_name'] + ' 主备风机研判报告\n\n' + decision.judgment.result + '\n\n' + decision.judgment.reason
|
|
|
+ report_virtual = prefix + '/report.md'
|
|
|
+ report_path = workspace_file(user_name, session_id, report_virtual)
|
|
|
+ report_path.write_text(body, encoding='utf-8')
|
|
|
+ context['query_exports'][report_virtual] = digest(report_path)
|
|
|
+ artifacts['report.md'] = report_virtual
|
|
|
+ sources = [{'title': row.get('filename') or Path(row['path']).name, 'url': row['path'], 'type': 'file'} for row in materials]
|
|
|
+ sources.append({'title': '本轮完整工作流输入', 'url': artifacts['qwen-input.json'], 'type': 'file'})
|
|
|
+ value = {'status': 'completed', 'mine_code': scope.mine['mine_id'], 'mine_name': scope.mine['mine_name'],
|
|
|
+ 'check_object': decision.objectDetail, 'result_title': scope.mine['mine_name'] + ' 主备风机研判报告',
|
|
|
+ 'result': body, 'result_format': 'report', 'workflow_id': workflow_id, 'sources': sources,
|
|
|
+ 'check_process': '固定监测预处理 → Python/GLM OCR及Qwen补识别 → 无关页过滤 → 固定监测阈值判定 → Qwen一次研判 → JSON校验及最多3次格式修复 → 固定告警结论合并 → 本地归档',
|
|
|
+ 'platform_payload': platform, 'analysis': None, 'assessment': {'mode': VERSION,
|
|
|
+ 'decision': {'result_code': decision.judgment.resultCode, 'summary': decision.judgment.result},
|
|
|
+ 'monitoring_assessment': monitoring_assessment,
|
|
|
+ 'history_coverage': inputs['history_coverage'],
|
|
|
+ 'report_archives': [{key: row[key] for key in ('path', 'sha256', 'is_mock', 'binding',
|
|
|
+ 'page_count', 'complete', 'reused')} for row in reports],
|
|
|
+ 'workflow_files': artifacts, 'diagnostics': diagnostics},
|
|
|
+ 'artifacts': [{'filename': name, 'path': artifacts[name]} for name in (
|
|
|
+ 'report.md', 'platform-payload.json', 'qwen-input.json', 'pdf-extraction.json',
|
|
|
+ 'pdf-page-selection.json', 'qwen-response.json', 'qwen-final-response.json',
|
|
|
+ 'json-repair-history.json', 'monitoring-assessment.json', 'decision-resolution.json') if name in artifacts]}
|
|
|
+ with db_conn.transaction():
|
|
|
+ saved = main_fan_store.save_analysis(user_name, session_id, value)
|
|
|
+ result = MainFanResult.model_validate(saved).model_dump(mode='json')
|
|
|
+ stage('校验并保存平台JSON', 'completed', {'analysis_id': saved['analysis_id']})
|
|
|
+ main_fan_store.update_workflow(workflow_id, context=context, stages=workflow['stages'], result=result, status='completed')
|
|
|
+ try:
|
|
|
+ await asyncio.to_thread(chat_store.save_message, session_id, 'user', '执行主通风机固定工作流（本轮数据独立研判）')
|
|
|
+ await asyncio.to_thread(chat_store.save_message, session_id, 'assistant', body, citations=sources)
|
|
|
+ except Exception:
|
|
|
+ logger.warning('工作流结果已归档，但会话展示消息保存失败', exc_info=True)
|
|
|
+ logger.info('工作流研判已完成并保存', extra={'kv': {'event': 'workflow.completed',
|
|
|
+ 'result_code': decision.judgment.resultCode, 'qwen_calls': diagnostics['qwen_calls'],
|
|
|
+ 'duration_ms': round((time.monotonic() - started) * 1000)}})
|
|
|
+ return ExecutionResult(run_id, session_id, 'success', body, '', usage(), sources, [], result, diagnostics)
|
|
|
+ except asyncio.CancelledError:
|
|
|
+ main_fan_store.update_workflow(workflow_id, context=context, stages=workflow['stages'], status='interrupted')
|
|
|
+ raise
|
|
|
+ except Exception as exc:
|
|
|
+ code = exc.code if isinstance(exc, WorkflowError) else type(exc).__name__
|
|
|
+ diagnostics['error'] = str(exc)[:500]
|
|
|
+ if isinstance(exc, WorkflowError) and exc.details:
|
|
|
+ diagnostics['error_details'] = exc.details
|
|
|
+ stage(stage_name, 'failed', {'error_code': code, 'message': str(exc)[:500],
|
|
|
+ 'error_details': diagnostics.get('error_details', {})})
|
|
|
+ write('failure.json', {'error_code': code, 'message': str(exc)[:500], 'diagnostics': diagnostics})
|
|
|
+ main_fan_store.update_workflow(workflow_id, context=context, stages=workflow['stages'], status='failed')
|
|
|
+ logger.error('工作流研判失败', extra={'kv': {'event': 'workflow.failed', 'error_code': code,
|
|
|
+ 'stage': stage_name, 'qwen_calls': diagnostics['qwen_calls'], 'message': str(exc)[:500],
|
|
|
+ 'error_details': diagnostics.get('error_details', {})}})
|
|
|
+ return ExecutionResult(run_id, session_id, 'failed', '', code, usage(), [], [], None, diagnostics)
|
|
|
+ finally:
|
|
|
+ try:
|
|
|
+ for meter in meters.values():
|
|
|
+ await asyncio.shield(asyncio.to_thread(meter.settle))
|
|
|
+ finally:
|
|
|
+ if admission is not None:
|
|
|
+ admission.release()
|
