Deploy MOFScreen-Agent FastAPI backend

agent/conversation.py

@@ -1,27 +1,42 @@
-"""Natural-language intent agent for selective MOF tool execution."""
+"""LLM-guided router for online CIF screening tools."""
 from __future__ import annotations
 
 import json
 import re
-from pathlib import Path
+import urllib.error
+import urllib.request
+from types import SimpleNamespace
+from typing import Any
 
-import yaml
+from tools.online_screening import ACTIVE_TOOLS, run_online_tools
 
 
-def parse_user_request(
-    user_request: str,
-    llm_provider: str,
-    llm_api_key: str | None = None,
-) -> dict:
-    """Parse a natural-language user request into a selective tool plan using an LLM only."""
-    if llm_provider == "rule_based" or not llm_api_key:
-        raise ValueError("Conversational intent recognition requires an LLM provider and API key.")
+TOOL_KEYWORDS = {
+    "predict_adsorption": ("adsorption", "benzene", "toluene", "吸附", "苯", "甲苯"),
+    "check_metals": ("metal", "heavy", "重金属", "金属"),
+    "identify_linker": ("ligand", "linker", "配体"),
+    "predict_sa_score": ("synth", "sa", "可合成", "合成性"),
+    "predict_aquatic_toxicity": ("tox", "toxicity", "aquatic", "lc50", "igc50", "ibc50", "毒性", "水生"),
+    "predict_price": ("price", "cost", "coprinet", "usd", "价格", "成本"),
+}
+
+FULL_KEYWORDS = (
+    "full", "complete", "overall", "comprehensive", "screening", "pass",
+    "完整", "综合", "筛选", "能不能通过", "是否通过", "评价一下", "评估一下",
+)
+
+ACTIVE_FULL_PLAN = list(ACTIVE_TOOLS)
 
-    llm_intent = _try_llm_intent(user_request, llm_provider, llm_api_key)
-    if isinstance(llm_intent, dict):
-        return _normalize_intent(llm_intent)
 
-    raise RuntimeError("LLM intent recognition failed. Check the API key, provider, and model access.")
+def parse_user_request(user_request: str, llm_provider: str, llm_api_key: str | None = None) -> dict:
+    """Return the intended online tool plan."""
+    rule_intent = _rule_based_intent(user_request)
+    if rule_intent["mode"] == "general_answer":
+        return rule_intent
+    llm_intent = _try_llm_tool_plan(user_request, {}, llm_provider, llm_api_key)
+    if llm_intent:
+        return _normalize_intent(llm_intent, user_request)
+    return rule_intent
 
 
 def run_conversational_screening(
@@ -29,350 +44,373 @@ def run_conversational_screening(
     cif_path: str | None,
     llm_provider: str,
     llm_api_key: str | None = None,
+    final_llm: bool = True,
 ) -> dict:
-    """Run only the tools required by the parsed user intent."""
-    try:
-        intent = parse_user_request(user_request, llm_provider, llm_api_key)
-    except Exception as exc:
+    if _is_pmt_only_request(user_request):
         return {
-            "intent": None,
-            "needs_api_key": not llm_api_key or llm_provider == "rule_based",
+            "intent": {"mode": "out_of_scope", "requires_cif": False, "tools": [], "user_goal": user_request},
             "needs_file": False,
-            "assistant_message": str(exc),
-            "agent_trace": [{
-                "agent": "Intent Agent",
-                "action": "blocked because LLM intent recognition is required",
-                "output": {"error": str(exc)},
-            }],
+            "assistant_message": "This request is outside the platform scope.",
+            "agent_trace": [],
             "results": {},
             "warnings": [],
-            "errors": [str(exc)],
+            "errors": [],
         }
 
-    trace = [{
-        "agent": "Intent Agent",
-        "action": "parsed natural-language request into a selective tool plan",
-        "output": intent,
-    }]
+    intent = parse_user_request(user_request, llm_provider, llm_api_key)
+    trace = [{"agent": "Orchestrator Agent", "action": "planned online tool calls", "output": intent}]
 
-    if intent["requires_cif"] and not cif_path:
+    if intent["mode"] == "general_answer":
+        message = _general_answer(user_request, llm_provider, llm_api_key)
         return {
             "intent": intent,
-            "needs_file": True,
-            "assistant_message": _missing_file_message(intent),
+            "needs_file": False,
+            "assistant_message": message,
             "agent_trace": trace,
             "results": {},
             "warnings": [],
             "errors": [],
         }
 
-    if not cif_path:
+    if intent["requires_cif"] and not cif_path:
         return {
             "intent": intent,
-            "needs_file": False,
-            "assistant_message": "This request does not require a CIF file, but no matching non-file task is implemented yet.",
+            "needs_file": True,
+            "assistant_message": "This request requires a CIF file before online calculations can run.",
             "agent_trace": trace,
             "results": {},
             "warnings": [],
             "errors": [],
         }
 
-    results: dict = {}
-    warnings: list[str] = []
-    errors: list[str] = []
+    result = run_online_tools(cif_path, intent["tools"])
+    result["intent"] = intent
+    result["needs_file"] = False
+    result["agent_trace"] = trace + result.get("agent_trace", [])
+    result["assistant_message"] = (
+        _try_llm_result_response(user_request, result, llm_provider, llm_api_key)
+        if final_llm else None
+    ) or _build_response(result)
+    return result
+
+
+def _normalize_intent(intent: dict, user_request: str) -> dict:
+    mode = str(intent.get("mode") or "selective_tools")
+    tools = intent.get("tools") or []
+    if isinstance(tools, str):
+        tools = [tools]
+    tools = [tool for tool in tools if tool in set(ACTIVE_TOOLS)]
+    if mode in {"full", "full_screening", "full_six_step_screening"}:
+        mode = "full_online_screening"
+        tools = ACTIVE_FULL_PLAN
+    elif mode == "general_answer":
+        tools = []
+    elif not tools:
+        return _rule_based_intent(user_request)
+    else:
+        mode = "selective_tools"
+    return {
+        "mode": mode,
+        "requires_cif": mode != "general_answer",
+        "tools": tools,
+        "user_goal": intent.get("user_goal") or user_request,
+        "rationale_summary": intent.get("rationale_summary"),
+    }
 
-    if intent["mode"] == "full_screening":
-        from agent.graph import screening_app
 
-        full = screening_app.invoke({
-            "cif_path": cif_path,
-            "llm_provider": llm_provider,
-            "llm_api_key": llm_api_key,
-            "warnings": [],
-            "errors": [],
-        })
-        full["intent"] = intent
-        full["agent_trace"] = trace + list(full.get("agent_trace", []))
-        return full
+def _rule_based_intent(user_request: str) -> dict:
+    text = (user_request or "").lower()
+    is_conceptual = any(keyword in text for keyword in ("是什么", "解释", "explain", "what is", "meaning", "含义"))
+    candidate_specific = any(
+        keyword in text
+        for keyword in ("风险", "预测", "筛选", "评价", "评估", "怎么样", "多少", "rank", "price", "toxicity", "screen", "计算")
+    )
+    if is_conceptual and not candidate_specific:
+        return {"mode": "general_answer", "requires_cif": False, "tools": [], "user_goal": user_request}
 
-    try:
-        if intent["targets"].get("adsorption"):
-            ads = _run_adsorption_tools(cif_path, intent["targets"]["adsorption"])
-            results["adsorption"] = ads["results"]
-            warnings.extend(ads["warnings"])
-            trace.extend(ads["trace"])
-
-        if intent["targets"].get("toxicity") or intent["targets"].get("safety"):
-            linker_bundle = _run_linker_tool(cif_path)
-            results["linker"] = linker_bundle["result"]
-            warnings.extend(linker_bundle["warnings"])
-            trace.extend(linker_bundle["trace"])
-
-            linker = linker_bundle["result"]
-            smiles = linker.get("linker_smiles") if linker else None
-            metals = linker.get("metals", []) if linker else []
-
-            if intent["targets"].get("toxicity"):
-                tox_bundle = _run_toxicity_tool(smiles)
-                results["toxicity"] = tox_bundle["result"]
-                warnings.extend(tox_bundle["warnings"])
-                trace.extend(tox_bundle["trace"])
-
-            if intent["targets"].get("safety"):
-                safety_bundle = _run_safety_tools(metals, smiles)
-                results["safety"] = safety_bundle["result"]
-                warnings.extend(safety_bundle["warnings"])
-                trace.extend(safety_bundle["trace"])
-    except Exception as exc:
-        errors.append(str(exc))
+    is_full = any(keyword in text for keyword in FULL_KEYWORDS)
+    tools = [name for name, keywords in TOOL_KEYWORDS.items() if any(keyword in text for keyword in keywords)]
+    if is_full or "综合" in text:
+        tools = ACTIVE_FULL_PLAN
+        mode = "full_online_screening"
+    elif tools:
+        mode = "selective_tools"
+    else:
+        mode = "general_answer"
+    return {"mode": mode, "requires_cif": mode != "general_answer", "tools": tools, "user_goal": user_request}
 
-    response = _build_selective_response(intent, results, warnings, errors)
-    trace.append({
-        "agent": "Response Agent",
-        "action": "generated targeted answer from selected tool outputs",
-        "output": {"assistant_message": response},
-    })
 
-    return {
-        "intent": intent,
-        "needs_file": False,
-        "assistant_message": response,
-        "agent_trace": trace,
-        "results": results,
-        "warnings": warnings,
-        "errors": errors,
-    }
+def _is_pmt_only_request(user_request: str) -> bool:
+    text = (user_request or "").lower()
+    if not any(keyword in text for keyword in ("pmt", "pbt", "vpvm")):
+        return False
+    return not any(any(keyword in text for keyword in keywords) for keywords in TOOL_KEYWORDS.values())
 
 
-def _normalize_intent(intent: dict) -> dict:
-    normalized = {
-        "mode": intent.get("mode", "selective_tools"),
-        "requires_cif": True,
-        "targets": intent.get("targets") or {},
-        "user_goal": intent.get("user_goal", ""),
-        "clarification": intent.get("clarification"),
-    }
-    targets = normalized["targets"]
-    targets.setdefault("adsorption", [])
-    targets.setdefault("toxicity", False)
-    targets.setdefault("safety", False)
-    targets["adsorption"] = [
-        target for target in targets.get("adsorption", [])
-        if target in {"benzene", "toluene"}
-    ]
-    targets["toxicity"] = bool(targets.get("toxicity"))
-    targets["safety"] = bool(targets.get("safety"))
-    normalized["targets"] = targets
-    normalized["tool_plan"] = _tool_plan_from_targets(targets, normalized.get("mode", "selective_tools"))
-    normalized["requires_cif"] = True
-    return normalized
-
-
-def _tool_plan_from_targets(targets: dict, mode: str) -> list[str]:
-    if mode == "full_screening":
-        return ["full_agentic_screening_graph"]
-    plan = []
-    adsorption = targets.get("adsorption") or []
-    if adsorption:
-        plan.append("compute_scm_descriptor")
-        for target in adsorption:
-            plan.append(f"predict_{target}_adsorption")
-    if targets.get("toxicity") or targets.get("safety"):
-        plan.append("extract_linker")
-    if targets.get("toxicity"):
-        plan.append("predict_toxicity")
-    if targets.get("safety"):
-        plan.append("apply_safety_rules")
-    return plan
-
-
-def _try_llm_intent(user_request: str, provider: str, api_key: str | None) -> dict | None:
+def _try_llm_tool_plan(user_request: str, current_results: dict, provider: str, api_key: str | None) -> dict | None:
     if provider == "rule_based" or not api_key:
-        raise ValueError("Conversational intent recognition requires an LLM provider and API key.")
+        return None
     try:
-        from langchain_openai import ChatOpenAI
-
-        cfg = {"temperature": 0.0, "max_tokens": 350, "api_key": api_key}
-        if provider == "openai":
-            cfg["model"] = "gpt-4o-mini"
-        elif provider == "deepseek":
-            cfg.update({"model": "deepseek-chat", "base_url": "https://api.deepseek.com/v1"})
-        elif provider == "qwen":
-            cfg.update({"model": "qwen-turbo", "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1"})
-        else:
+        response = _llm(provider, api_key, temperature=0.0, max_tokens=500).invoke(_orchestrator_prompt(user_request, current_results))
+        parsed = _extract_json(response.content)
+        if not parsed or parsed.get("action") == "final_answer":
             return None
+        return {
+            "mode": "selective_tools",
+            "tools": [item.get("name") for item in parsed.get("tools", []) if isinstance(item, dict)],
+            "rationale_summary": parsed.get("rationale_summary"),
+            "user_goal": user_request,
+        }
+    except Exception:
+        return None
+
+
+def _orchestrator_prompt(user_request: str, current_results: dict) -> str:
+    return f"""You are MOFScreen-Agent, a MOF research assistant with expert planning ability.
+Infer the user's real screening intent and call the minimum necessary tools to build an evidence chain.
+
+Principles:
+1. Do not answer only the literal wording. If the user asks about toxicity, gather toxicity data, linker structure, and metal information when needed.
+2. Respect dependencies:
+   - Toxicity, price, and synthesizability require linkers; call identify_linker first when linker data is missing.
+   - Environmental risk should combine metal information and toxicity predictions when relevant.
+   - Adsorption performance requires predict_adsorption.
+3. Quantitative values must come from tools. Never invent values.
+4. Before final_answer, collect enough evidence to support the requested conclusion.
+
+Available tools:
+parse_cif, predict_adsorption, identify_linker, check_metals, predict_sa_score, predict_aquatic_toxicity, predict_price, final_answer
 
-        prompt = f"""
-You are an Intent Agent for a MOF screening application.
-Parse the user request into JSON only.
-Allowed mode: "selective_tools" or "full_screening".
-Allowed adsorption targets: "benzene", "toluene".
-Set toxicity true only if the user asks toxicity/ecotoxicity.
-Set safety true only if the user asks safety, PMT, metal risk, PFAS, or full evaluation.
-Always requires_cif=true for property prediction.
+Context:
+- User request: {user_request}
+- Existing tool results: {json.dumps(current_results, ensure_ascii=False, default=str)}
+
+Output JSON only. No Markdown.
+Fields:
+- action: "tool_call" or "final_answer"
+- tools: tools to call; empty for final_answer
 
-JSON shape:
 {{
-  "mode": "selective_tools",
-  "requires_cif": true,
-  "targets": {{"adsorption": ["benzene"], "toxicity": false, "safety": false}},
-  "user_goal": "short description",
-  "clarification": null
+  "action": "tool_call",
+  "tools": [{{"name": "tool_name", "arguments": {{}}}}],
+  "rationale_summary": "One sentence explaining why these tools are needed.",
+  "final_response": null
 }}
-
-User request: {user_request}
 """
-        response = ChatOpenAI(**cfg).invoke(prompt)
-        return _extract_json(response.content)
-    except Exception:
-        return None
 
 
-def _extract_json(text: str) -> dict | None:
+def _general_answer(user_request: str, llm_provider: str, llm_api_key: str | None) -> str:
+    llm_answer = _try_llm_answer(user_request, llm_provider, llm_api_key)
+    if llm_answer:
+        return llm_answer
+    return "I can answer general MOF questions. For material-specific calculations, please upload a CIF file."
+
+
+def _try_llm_answer(user_request: str, provider: str, api_key: str | None) -> str | None:
+    if provider == "rule_based" or not api_key:
+        return None
     try:
-        return json.loads(text)
-    except Exception:
-        match = re.search(r"\{.*\}", text, flags=re.S)
-        if not match:
-            return None
-        try:
-            return json.loads(match.group(0))
-        except Exception:
-            return None
+        prompt = f"""You are MOFScreen-Agent, a research assistant focused on metal-organic frameworks (MOFs).
+You have strong materials chemistry knowledge, but you are strict: for specific material data such as adsorption, toxicity, or price, calculations take priority and you must not guess values.
 
+Task: Answer the user's general or methodological question in concise, professional English.
 
-def _run_adsorption_tools(cif_path: str, targets: list[str]) -> dict:
-    from tools.adsorption import predict_benzene, predict_toluene
-    from tools.descriptors import compute_scm_eigenvalues
-
-    cfg_path = Path(__file__).resolve().parent.parent / "configs" / "paths.yaml"
-    with open(cfg_path, "r", encoding="utf-8") as f:
-        cfg = yaml.safe_load(f)["adsorption_models"]
-
-    results = {}
-    warnings = []
-    trace = []
-    for target in targets:
-        dim = cfg[f"{target}_target_dim"]
-        desc = compute_scm_eigenvalues(cif_path, dim)
-        pred = predict_benzene(desc["eigenvalues"]) if target == "benzene" else predict_toluene(desc["eigenvalues"])
-        results[target] = {
-            "uptake_mg_g": pred["uptake_mg_g"],
-            "model": pred["model_version"],
-            "descriptor_raw_dim": desc["raw_dim"],
-            "descriptor_padded": desc["padded"],
-            "descriptor_truncated": desc["truncated"],
-        }
-        for warning in [desc["applicability_warning"], pred["applicability_warning"]]:
-            if warning:
-                warnings.append(warning)
-        trace.append({
-            "agent": "Tool Execution Agent",
-            "action": f"ran {target} adsorption prediction only",
-            "output": results[target],
-        })
-    return {"results": results, "warnings": warnings, "trace": trace}
-
-
-def _run_linker_tool(cif_path: str) -> dict:
-    from tools.linker_extraction import extract_linker
-
-    result = extract_linker(cif_path)
-    warnings = [result.get("extraction_note")] if result.get("extraction_level") == 3 else []
-    return {
-        "result": result,
-        "warnings": [w for w in warnings if w],
-        "trace": [{
-            "agent": "Tool Execution Agent",
-            "action": "ran extract_linker because toxicity or safety was requested",
-            "output": result,
-        }],
-    }
+Strategy:
+1. For MOF concepts, answer with relevant scientific context.
+2. If the user asks for specific material performance, ask them to upload a CIF file so the platform can run tools.
+3. Be helpful, rigorous, and avoid unsupported numerical claims.
 
+User question: {user_request}
+"""
+        return _llm(provider, api_key, temperature=0.2, max_tokens=500).invoke(prompt).content
+    except Exception:
+        return None
 
-def _run_toxicity_tool(smiles: str | None) -> dict:
-    from tools.toxicity import predict_toxicity
 
-    result = predict_toxicity(smiles)
-    return {
-        "result": result,
-        "warnings": [],
-        "trace": [{
-            "agent": "Tool Execution Agent",
-            "action": "ran predict_toxicity because toxicity was requested",
-            "output": result,
-        }],
+def _result_prompt(user_request: str, result: dict) -> str:
+    safe_result = {
+        "mof_id": result.get("mof_id"),
+        "intent": result.get("intent"),
+        "tool_results": result.get("tool_results") or result.get("results"),
+        "gate_status": result.get("gate_status"),
+        "recommendation": result.get("recommendation"),
+        "errors": result.get("errors"),
+        "warnings": result.get("warnings"),
     }
+    return f"""You are MOFScreen-Agent, a research assistant for MOF virtual screening.
+Your task is to produce an English evaluation report based strictly on backend tool results.
+Do not introduce yourself as an analysis module, and do not say that you are converting raw data into a report. Start directly with the conclusion.
+
+Input:
+- User request: {user_request}
+- Tool results: {json.dumps(safe_result, ensure_ascii=False, default=str)}
+
+Requirements:
+1. **Data fidelity**: all quantitative values, including LC50, price, adsorption uptake, and SA score, must come directly from tool results. Do not invent numbers.
+2. **Deep interpretation**:
+   - Do not only repeat a number such as "LC50 is 12.5 mg/L"; explain what the value implies for screening.
+   - Connect the result to structure where possible. For example, Zn nodes often suggest better biocompatibility than many heavy metals, while nitrogen heterocycle linkers may contribute to biological activity.
+   - Use cautious benchmark language when appropriate, and clearly mark it as interpretation rather than additional computation.
+3. **Structured output**: use Markdown and exactly these four sections.
+
+Output template:
+
+### 📊 Conclusion
+Summarize the core conclusion in 1-2 sentences and answer the user's question directly.
+
+### 🔬 Key Evidence
+List the key tool-returned metrics. A compact table or bullets are both acceptable.
+*   **Metric**: value + unit
+*   **Assessment**: Pass/Fail or low/medium/high risk where supported by the result
+
+### 🧠 Expert Analysis
+Explain the structure-property relationship:
+*   **Likely drivers**: connect metal nodes, linker features, and adsorption/toxicity/price results.
+*   **Potential mechanism**: for example, pore confinement, aromatic interactions, metal leaching risk, or linker-driven toxicity.
+*   **Uncertainty**: state model/domain limitations and failed or missing tool outputs.
+
+### 💡 Decision Advice
+Give concrete next-step guidance:
+*   Whether the MOF looks suitable for the requested use case.
+*   Whether modification, coating, or linker/metal substitution should be considered.
+*   Which experiment or manual check should be prioritized next.
+
+Use a professional, objective, advisory tone.
+"""
 
 
-def _run_safety_tools(metals: list[str], smiles: str | None) -> dict:
-    from tools.safety_rules import check_metal_safety, check_pmt_pre_filter
-
-    metal = check_metal_safety(metals) if metals else {
-        "tier": "unknown",
-        "flagged_metals": [],
-        "details": "No metals found.",
-    }
-    pmt = check_pmt_pre_filter(smiles)
-    result = {
-        "metal_tier": metal["tier"],
-        "metal_flagged": metal["flagged_metals"],
-        "metal_details": metal["details"],
-        "pmt_pass": pmt["pmt_pass"],
-        "pmt_flags": pmt["flags"],
-        "pmt_descriptors": pmt["descriptors"],
-    }
-    return {
-        "result": result,
-        "warnings": [],
-        "trace": [{
-            "agent": "Tool Execution Agent",
-            "action": "ran safety rules because safety was requested",
-            "output": result,
-        }],
-    }
+def _try_llm_result_response(user_request: str, result: dict, provider: str, api_key: str | None) -> str | None:
+    if provider == "rule_based" or not api_key:
+        return None
+    try:
+        return _llm(provider, api_key, temperature=0.2, max_tokens=1600).invoke(_result_prompt(user_request, result)).content
+    except Exception:
+        return None
 
 
-def _missing_file_message(intent: dict) -> str:
-    tools = ", ".join(intent.get("tool_plan", []))
-    return (
-        "我已经识别到你的任务，但这个预测需要 CIF 结构文件。"
-        f"请先上传一个 .cif 文件，然后我会只调用这些工具：{tools}。"
-    )
+def stream_llm_result_response(user_request: str, result: dict, provider: str, api_key: str | None):
+    if result.get("assistant_message") and not (result.get("tool_results") or result.get("results")):
+        yield str(result["assistant_message"])
+        return
+    if provider == "rule_based" or not api_key:
+        yield _build_response(result)
+        return
+    try:
+        for chunk in _llm(provider, api_key, temperature=0.2, max_tokens=1600).stream(_result_prompt(user_request, result)):
+            text = getattr(chunk, "content", "")
+            if text:
+                yield text
+    except Exception as exc:
+        yield f"LLM call failed; using the rule-based summary instead. Error: {_safe_error(exc)}\n\n"
+        yield _build_response(result)
 
 
-def _build_selective_response(intent: dict, results: dict, warnings: list[str], errors: list[str]) -> str:
-    if errors:
-        return "任务执行失败：" + "; ".join(errors)
+def _build_response(result: dict) -> str:
+    if result.get("errors"):
+        return "Task failed: " + "; ".join(result["errors"])
+    tools = result.get("tool_results") or result.get("results") or {}
+    if not tools:
+        return result.get("assistant_message") or "No screening tools were called."
 
-    lines = ["Intent Agent 已识别你的需求，并只调用了必要工具。"]
-    adsorption = results.get("adsorption") or {}
+    lines = ["### 📊 Conclusion", "Online tool calculations were run for the uploaded CIF. This fallback summary is based only on deterministic tool outputs because no usable LLM API key was available.", "", "### 🔬 Key Evidence"]
+    adsorption = tools.get("predict_adsorption") or {}
     if adsorption:
-        for target, payload in adsorption.items():
-            label = "苯" if target == "benzene" else "甲苯"
-            lines.append(
-                f"{label}吸附量预测值：{payload['uptake_mg_g']} mg/g "
-                f"({payload['model']}; SCM raw dim={payload['descriptor_raw_dim']})."
-            )
-
-    if "toxicity" in results:
-        tox = results["toxicity"]
-        lines.append(
-            "毒性预测已完成："
-            f"LC50 fish={tox.get('LC50_Pimephales')}, "
-            f"LC50 daphnia={tox.get('LC50_Daphnia')}, "
-            f"IGC50={tox.get('IGC50_Tetrahymena')}, "
-            f"IBC50={tox.get('IBC50_Vibrio')}."
+        lines.append(f"- Adsorption prediction: benzene {adsorption.get('benzene_uptake_mg_g')} mg/g; toluene {adsorption.get('toluene_uptake_mg_g')} mg/g.")
+    metals = tools.get("check_metals") or {}
+    if metals:
+        lines.append(f"- Metal check: {metals.get('summary')}")
+    linker = tools.get("identify_linker") or {}
+    if linker:
+        lines.append(f"- Linker: {linker.get('linker_name') or linker.get('linker_formula') or 'unidentified'}; SMILES={linker.get('linker_smiles') or 'N/A'}.")
+    sa = tools.get("predict_sa_score") or {}
+    if sa:
+        lines.append(f"- SA score：{sa.get('sa_score')}；{sa.get('scale', '')}。")
+    tox = tools.get("predict_aquatic_toxicity") or {}
+    if tox:
+        lines.append(f"- Aquatic toxicity: mean={tox.get('mean_toxicity')}; worst={tox.get('worst_toxicity')}.")
+    price = tools.get("predict_price") or {}
+    if price:
+        lines.append(f"- Price prediction: {price.get('usd_per_mmol')} USD/mmol; {price.get('usd_per_g')} USD/g.")
+    failed = [name for name, payload in tools.items() if payload.get("status") in {"error", "unavailable"}]
+    lines.extend(["", "### 🧠 Expert Analysis"])
+    if linker:
+        lines.append(f"- **Structural clue**: the identified linker is {linker.get('linker_name') or linker.get('linker_formula') or 'unidentified'}, SMILES={linker.get('linker_smiles') or 'N/A'}.")
+    if metals:
+        lines.append(f"- **Metal node**: {metals.get('summary')}.")
+    if adsorption:
+        lines.append("- **Adsorption interpretation**: benzene/toluene uptake values come from a structure-descriptor model and should be checked against the model applicability domain and experiments.")
+    if tox:
+        lines.append("- **Toxicity interpretation**: aquatic toxicity endpoints are predicted from linker SMILES and should be treated as an environmental risk screen, not a substitute for full ecotoxicology testing.")
+    if failed:
+        lines.append("- **Uncertainty**: " + "; ".join(f"{name}: {tools[name].get('summary')}" for name in failed))
+    lines.extend(["", "### 💡 Decision Advice", "- Prioritize manual review of the linker identification and model applicability domain.", "- For formal screening, add experimental validation matched to the target application."])
+    return "\n".join(lines)
+
+
+def _llm(provider: str, api_key: str, temperature: float, max_tokens: int):
+    if provider == "openai":
+        return _OpenAICompatibleLLM("https://api.openai.com/v1", "gpt-4o-mini", api_key, temperature, max_tokens)
+    elif provider == "deepseek":
+        return _OpenAICompatibleLLM("https://api.deepseek.com/v1", "deepseek-chat", api_key, temperature, max_tokens)
+    elif provider == "qwen":
+        return _OpenAICompatibleLLM("https://dashscope.aliyuncs.com/compatible-mode/v1", "qwen-turbo", api_key, temperature, max_tokens)
+    else:
+        raise ValueError(f"Unsupported LLM provider: {provider}")
+
+
+class _OpenAICompatibleLLM:
+    def __init__(self, base_url: str, model: str, api_key: str, temperature: float, max_tokens: int):
+        self.base_url = base_url.rstrip("/")
+        self.model = model
+        self.api_key = api_key
+        self.temperature = temperature
+        self.max_tokens = max_tokens
+
+    def invoke(self, prompt: str):
+        return SimpleNamespace(content=self._complete(prompt))
+
+    def stream(self, prompt: str):
+        # ponytail: non-stream HTTP keeps provider support tiny; swap to SSE parsing if token-level streaming matters.
+        yield SimpleNamespace(content=self._complete(prompt))
+
+    def _complete(self, prompt: str) -> str:
+        payload = json.dumps({
+            "model": self.model,
+            "messages": [{"role": "user", "content": prompt}],
+            "temperature": self.temperature,
+            "max_tokens": self.max_tokens,
+        }).encode("utf-8")
+        req = urllib.request.Request(
+            f"{self.base_url}/chat/completions",
+            data=payload,
+            headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"},
+            method="POST",
         )
+        try:
+            with urllib.request.urlopen(req, timeout=60) as resp:
+                data = json.loads(resp.read().decode("utf-8"))
+        except urllib.error.HTTPError as exc:
+            detail = exc.read().decode("utf-8", "replace")
+            raise RuntimeError(f"LLM HTTP {exc.code}: {_safe_error_text(detail)}") from exc
+        return data.get("choices", [{}])[0].get("message", {}).get("content", "")
 
-    if "safety" in results:
-        safety = results["safety"]
-        lines.append(
-            f"安全规则结果：metal tier={safety.get('metal_tier')}, "
-            f"PMT={'pass' if safety.get('pmt_pass') else 'fail'}."
-        )
 
-    if warnings:
-        lines.append("注意：" + "；".join(warnings[:2]))
+def _safe_error(exc: Exception) -> str:
+    return _safe_error_text(str(exc))[:500]
+
+
+def _safe_error_text(text: str) -> str:
+    return re.sub(r"sk-[A-Za-z0-9_-]+", "sk-***", text)
+
 
-    return "\n\n".join(lines)
+def _extract_json(text: str) -> dict | None:
+    try:
+        return json.loads(text)
+    except Exception:
+        match = re.search(r"\{.*\}", text, flags=re.S)
+        if not match:
+            return None
+        try:
+            return json.loads(match.group(0))
+        except Exception:
+            return None

agent/graph.py

@@ -1,45 +1,32 @@
-"""LangGraph state machine assembly for MOF screening pipeline."""
-from langgraph.graph import StateGraph, END
-
-from agent.state import ScreeningState
+"""Screening workflow assembly for the MOF six-step pipeline."""
 from agent import nodes
+from agent.state import ScreeningState
+
+
+class SimpleScreeningApp:
+    """Small invoke-compatible workflow used when LangGraph is unavailable."""
+
+    def invoke(self, state: ScreeningState) -> ScreeningState:
+        current: ScreeningState = dict(state)
+        for node in (nodes.ingest_cif, nodes.six_step_screening_agent):
+            update = node(current)
+            current.update(update)
+        return current
 
 
 def build_graph():
-    g = StateGraph(ScreeningState)
-
-    g.add_node("ingest_cif", nodes.ingest_cif)
-    g.add_node("planning_agent", nodes.planning_agent)
-    g.add_node("tool_execution_agent", nodes.tool_execution_agent)
-    g.add_node("compute_descriptor", nodes.compute_descriptor)
-    g.add_node("predict_adsorption", nodes.predict_adsorption)
-    g.add_node("extract_linker", nodes.extract_linker)
-    g.add_node("predict_toxicity", nodes.predict_toxicity)
-    g.add_node("apply_safety_rules", nodes.apply_safety_rules)
-    g.add_node("evidence_audit_agent", nodes.evidence_audit_agent)
-    g.add_node("repair_agent", nodes.repair_agent)
-    g.add_node("safety_review_agent", nodes.safety_review_agent)
-    g.add_node("reviewer_panel_agent", nodes.reviewer_panel_agent)
-    g.add_node("decision_agent", nodes.decision_agent)
-    g.add_node("report_agent", nodes.report_agent)
-
-    g.set_entry_point("ingest_cif")
-    g.add_edge("ingest_cif", "planning_agent")
-    g.add_edge("planning_agent", "tool_execution_agent")
-    g.add_edge("tool_execution_agent", "compute_descriptor")
-    g.add_edge("compute_descriptor", "predict_adsorption")
-    g.add_edge("predict_adsorption", "extract_linker")
-    g.add_edge("extract_linker", "predict_toxicity")
-    g.add_edge("predict_toxicity", "apply_safety_rules")
-    g.add_edge("apply_safety_rules", "evidence_audit_agent")
-    g.add_edge("evidence_audit_agent", "repair_agent")
-    g.add_edge("repair_agent", "safety_review_agent")
-    g.add_edge("safety_review_agent", "reviewer_panel_agent")
-    g.add_edge("reviewer_panel_agent", "decision_agent")
-    g.add_edge("decision_agent", "report_agent")
-    g.add_edge("report_agent", END)
-
-    return g.compile()
+    try:
+        from langgraph.graph import END, StateGraph
+
+        g = StateGraph(ScreeningState)
+        g.add_node("ingest_cif", nodes.ingest_cif)
+        g.add_node("six_step_screening", nodes.six_step_screening_agent)
+        g.set_entry_point("ingest_cif")
+        g.add_edge("ingest_cif", "six_step_screening")
+        g.add_edge("six_step_screening", END)
+        return g.compile()
+    except Exception:
+        return SimpleScreeningApp()
 
 
 screening_app = build_graph()

agent/nodes.py

@@ -101,6 +101,48 @@ def ingest_cif(state: ScreeningState) -> dict:
     return {"mof_id": mof_id, "warnings": warnings, "errors": errors}
 
 
+def six_step_screening_agent(state: ScreeningState) -> dict:
+    """Run online screening tools for an uploaded CIF."""
+    warnings = list(state.get("warnings", []))
+    errors = list(state.get("errors", []))
+    if errors:
+        return {"warnings": warnings, "errors": errors}
+
+    try:
+        from tools.online_screening import ACTIVE_TOOLS, run_online_tools
+
+        result = run_online_tools(
+            state.get("cif_path"),
+            list(ACTIVE_TOOLS),
+        )
+        return {
+            **result,
+            "warnings": warnings + list(result.get("warnings", [])),
+            "errors": errors + list(result.get("errors", [])),
+            "agent_trace": _append_trace(
+                state,
+                "Online Screening Agent",
+                "ran online CIF screening tools",
+                {
+                    "gate_status": result.get("gate_status"),
+                    "recommendation": result.get("recommendation"),
+                },
+            ) + list(result.get("agent_trace", [])),
+        }
+    except Exception as e:
+        errors.append(f"online_screening failed: {e}")
+        return {
+            "warnings": warnings,
+            "errors": errors,
+            "agent_trace": _append_trace(
+                state,
+                "Online Screening Agent",
+                "online screening failed",
+                {"error": str(e)},
+            ),
+        }
+
+
 def planning_agent(state: ScreeningState) -> dict:
     """Create a task card before deterministic tools are executed."""
     warnings = list(state.get("warnings", []))

agent/state.py

@@ -4,6 +4,7 @@ from typing import TypedDict
 
 class ScreeningState(TypedDict, total=False):
     cif_path: str
+    user_request: str
     mof_id: str
     task_card: dict | None
     execution_plan: dict | None
@@ -13,6 +14,10 @@ class ScreeningState(TypedDict, total=False):
     linker: dict | None
     toxicity: dict | None
     safety: dict | None
+    six_step: dict | None
+    gate_status: str | None
+    failed_at_step: int | None
+    row: dict | None
     evidence_ledger: list[dict]
     audit_report: dict | None
     repair_report: dict | None

backend/main.py

@@ -9,6 +9,7 @@ import math
 import os
 import sys
 import tempfile
+import json
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
@@ -16,13 +17,14 @@ from typing import Any
 import numpy as np
 from fastapi import FastAPI, File, Form, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 
 ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(ROOT))
 
-from agent.conversation import run_conversational_screening
-from agent.graph import screening_app
+from agent.conversation import run_conversational_screening, stream_llm_result_response
+from tools.six_step_screening import build_result_row
 
 
 app = FastAPI(title="MOFScreen-Agent API", version="1.0.0")
@@ -82,6 +84,17 @@ def _tox_display(value: Any, confidence: str | None) -> str:
 
 
 def _result_row(result: dict[str, Any], mof_id: str | None = None) -> dict[str, Any]:
+    if isinstance(result.get("row"), dict):
+        row = dict(result["row"])
+        if mof_id:
+            row["MOF ID"] = row.get("MOF ID") or mof_id
+        return row
+    if isinstance(result.get("six_step"), dict):
+        row = build_result_row(result)
+        if mof_id:
+            row["MOF ID"] = row.get("MOF ID") or mof_id
+        return row
+
     selective = result.get("results") or {}
     adsorption = selective.get("adsorption") or result.get("adsorption") or {}
     toxicity = selective.get("toxicity") or result.get("toxicity") or {}
@@ -117,7 +130,6 @@ def _result_row(result: dict[str, Any], mof_id: str | None = None) -> dict[str,
             toxicity.get("IBC50_Vibrio_confidence"),
         ),
         "Safety": safety.get("metal_tier", "N/A") if safety else "N/A",
-        "PMT": "Pass" if safety and safety.get("pmt_pass", True) else "N/A",
         "Score": result.get("final_score"),
         "Decision Class": (result.get("decision_record") or {}).get("decision_class", "N/A"),
         "Recommendation": result.get("recommendation", "N/A"),
@@ -137,6 +149,25 @@ def _frontend_result(result: dict[str, Any]) -> dict[str, Any]:
     return compact
 
 
+def _sse(event: str, data: Any) -> str:
+    return f"event: {event}\ndata: {json.dumps(_clean_json(data), ensure_ascii=False)}\n\n"
+
+
+def _tool_summary(name: str, payload: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "tool": name,
+        "status": payload.get("status"),
+        "step": payload.get("step"),
+        "summary": payload.get("summary")
+        or payload.get("reason")
+        or payload.get("price_status")
+        or payload.get("mof")
+        or payload.get("linker_smiles")
+        or payload.get("status"),
+        "result": payload,
+    }
+
+
 def _run_files(
     file_specs: list[tuple[str, bytes]],
     user_request: str,
@@ -156,23 +187,13 @@ def _run_files(
             tmp_path = str(Path(tmp_dir) / safe_name)
             Path(tmp_path).write_bytes(content)
             try:
-                if mode == "selective_tools":
-                    result = run_conversational_screening(
-                        user_request=user_request,
-                        cif_path=tmp_path,
-                        llm_provider=llm_provider,
-                        llm_api_key=llm_api_key,
-                    )
-                    mof_id = Path(filename or tmp_path).stem
-                else:
-                    result = screening_app.invoke({
-                        "cif_path": tmp_path,
-                        "llm_provider": llm_provider,
-                        "llm_api_key": llm_api_key,
-                        "warnings": [],
-                        "errors": [],
-                    })
-                    mof_id = result.get("mof_id") or Path(filename or tmp_path).stem
+                result = run_conversational_screening(
+                    user_request=user_request,
+                    cif_path=tmp_path,
+                    llm_provider=llm_provider,
+                    llm_api_key=llm_api_key,
+                )
+                mof_id = result.get("mof_id") or Path(filename or tmp_path).stem
                 result["uploaded_filename"] = filename
                 result["display_mof_id"] = mof_id
                 rows.append(_result_row(result, mof_id=mof_id))
@@ -187,11 +208,19 @@ def _run_files(
                 results.append(failed)
                 rows.append({
                     "MOF ID": failed["display_mof_id"],
+                    "Benzene (mg/g)": None,
+                    "Toluene (mg/g)": None,
+                    "Metal status": "N/A",
+                    "Metals": "N/A",
+                    "Linker": "N/A",
+                    "Linker SMILES": "N/A",
+                    "SA score": None,
+                    "Mean toxicity": None,
+                    "Price USD/g": None,
+                    "Price USD/mmol": None,
+                    "Gate status": "error",
                     "Recommendation": "error",
-                    "Decision Class": "N/A",
                     "Score": None,
-                    "Safety": "N/A",
-                    "PMT": "N/A",
                 })
 
     return _clean_json({"rows": rows, "results": results, "count": len(results)})
@@ -222,23 +251,13 @@ async def screen(
             tmp_path = str(Path(tmp_dir) / safe_name)
             Path(tmp_path).write_bytes(await uploaded.read())
             try:
-                if mode == "selective_tools":
-                    result = run_conversational_screening(
-                        user_request=user_request,
-                        cif_path=tmp_path,
-                        llm_provider=llm_provider,
-                        llm_api_key=llm_api_key,
-                    )
-                    mof_id = Path(uploaded.filename or tmp_path).stem
-                else:
-                    result = screening_app.invoke({
-                        "cif_path": tmp_path,
-                        "llm_provider": llm_provider,
-                        "llm_api_key": llm_api_key,
-                        "warnings": [],
-                        "errors": [],
-                    })
-                    mof_id = result.get("mof_id") or Path(uploaded.filename or tmp_path).stem
+                result = run_conversational_screening(
+                    user_request=user_request,
+                    cif_path=tmp_path,
+                    llm_provider=llm_provider,
+                    llm_api_key=llm_api_key,
+                )
+                mof_id = result.get("mof_id") or Path(uploaded.filename or tmp_path).stem
                 result["uploaded_filename"] = uploaded.filename
                 result["display_mof_id"] = mof_id
                 rows.append(_result_row(result, mof_id=mof_id))
@@ -253,11 +272,19 @@ async def screen(
                 results.append(failed)
                 rows.append({
                     "MOF ID": failed["display_mof_id"],
+                    "Benzene (mg/g)": None,
+                    "Toluene (mg/g)": None,
+                    "Metal status": "N/A",
+                    "Metals": "N/A",
+                    "Linker": "N/A",
+                    "Linker SMILES": "N/A",
+                    "SA score": None,
+                    "Mean toxicity": None,
+                    "Price USD/g": None,
+                    "Price USD/mmol": None,
+                    "Gate status": "error",
                     "Recommendation": "error",
-                    "Decision Class": "N/A",
                     "Score": None,
-                    "Safety": "N/A",
-                    "PMT": "N/A",
                 })
 
     return _clean_json({
@@ -267,6 +294,66 @@ async def screen(
     })
 
 
+@app.post("/chat/stream")
+async def chat_stream(
+    files: list[UploadFile] | None = File(None),
+    user_request: str = Form("Run full MOF screening."),
+    mode: str = Form("full_screening"),
+    llm_provider: str = Form("rule_based"),
+    llm_api_key: str | None = Form(None),
+):
+    file_specs: list[tuple[str, bytes]] = []
+    for uploaded in files or []:
+        file_specs.append((uploaded.filename or "candidate.cif", await uploaded.read()))
+
+    async def events():
+        if not file_specs:
+            result = run_conversational_screening(
+                user_request=user_request,
+                cif_path=None,
+                llm_provider=llm_provider,
+                llm_api_key=llm_api_key,
+                final_llm=False,
+            )
+            for chunk in stream_llm_result_response(user_request, result, llm_provider, llm_api_key):
+                yield _sse("assistant_delta", chunk)
+            yield _sse("final_result", {"rows": [], "results": [_frontend_result(result)], "count": 1})
+            return
+
+        rows: list[dict[str, Any]] = []
+        results: list[dict[str, Any]] = []
+        for filename, content in file_specs:
+            safe_name = Path(filename or "candidate.cif").name
+            if not safe_name.lower().endswith(".cif"):
+                safe_name = f"{Path(safe_name).stem or 'candidate'}.cif"
+            with tempfile.TemporaryDirectory() as tmp_dir:
+                tmp_path = str(Path(tmp_dir) / safe_name)
+                Path(tmp_path).write_bytes(content)
+                result = run_conversational_screening(
+                    user_request=user_request,
+                    cif_path=tmp_path,
+                    llm_provider=llm_provider,
+                    llm_api_key=llm_api_key,
+                    final_llm=False,
+                )
+                mof_id = result.get("mof_id") or Path(filename or tmp_path).stem
+                result["uploaded_filename"] = filename
+                result["display_mof_id"] = mof_id
+                for name, payload in (result.get("six_step") or result.get("results") or {}).items():
+                    if isinstance(payload, dict):
+                        yield _sse("tool_start", {"tool": name})
+                        yield _sse("tool_result", _tool_summary(name, payload))
+                text_result = dict(result)
+                text_result["assistant_message"] = None
+                for chunk in stream_llm_result_response(user_request, text_result, llm_provider, llm_api_key):
+                    yield _sse("assistant_delta", chunk)
+                rows.append(_result_row(result, mof_id=mof_id))
+                results.append(_clean_json(_frontend_result(result)))
+        yield _sse("final_result", {"rows": rows, "results": results, "count": len(results)})
+
+    return StreamingResponse(events(), media_type="text/event-stream")
+
+
 @app.post("/jobs/{job_id}/run")
 async def run_job(job_id: str, payload: RunJobRequest | None = None) -> dict[str, Any]:
     supabase = _supabase_admin()

tools/descriptors.py

@@ -2,9 +2,13 @@
 from pathlib import Path
 
 import numpy as np
-from matminer.featurizers.structure.matrix import SineCoulombMatrix
 from pymatgen.core import Structure
 
+try:
+    from matminer.featurizers.structure.matrix import SineCoulombMatrix
+except Exception:
+    SineCoulombMatrix = None
+
 
 def compute_scm_eigenvalues(
     cif_path: str,
@@ -37,9 +41,12 @@ def compute_scm_eigenvalues(
     if structure.num_sites == 0:
         raise ValueError(f"CIF parsed but contains 0 sites: {cif_path}")
 
-    scm = SineCoulombMatrix(flatten=False)
-    scm.fit([structure])
-    matrix = scm.featurize(structure)[0]
+    if SineCoulombMatrix is not None:
+        scm = SineCoulombMatrix(flatten=False)
+        scm.fit([structure])
+        matrix = scm.featurize(structure)[0]
+    else:
+        matrix = _fallback_scm_matrix(structure)
 
     eigenvalues = np.sort(np.linalg.eigvalsh(matrix))[::-1]
     raw_dim = len(eigenvalues)
@@ -72,3 +79,20 @@ def compute_scm_eigenvalues(
         "truncated": truncated,
         "applicability_warning": warning,
     }
+
+
+def _fallback_scm_matrix(structure: Structure) -> np.ndarray:
+    """Small deterministic SCM-like matrix used when matminer is unavailable."""
+    size = structure.num_sites
+    matrix = np.zeros((size, size), dtype=float)
+    for i, site_i in enumerate(structure):
+        z_i = float(site_i.specie.Z)
+        matrix[i, i] = 0.5 * z_i ** 2.4
+        for j in range(i + 1, size):
+            site_j = structure[j]
+            z_j = float(site_j.specie.Z)
+            distance = max(float(structure.get_distance(i, j)), 1e-6)
+            value = z_i * z_j / distance
+            matrix[i, j] = value
+            matrix[j, i] = value
+    return matrix

tools/online_screening.py

@@ -0,0 +1,364 @@
+"""Online tools for evaluating an arbitrary uploaded CIF."""
+from __future__ import annotations
+
+import importlib.util
+import math
+import sys
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+from pymatgen.core import Structure
+from rdkit import Chem
+
+from tools.adsorption import predict_benzene, predict_toluene
+from tools.descriptors import compute_scm_eigenvalues
+from tools.linker_extraction import extract_linker
+from tools.toxicity import predict_toxicity
+
+
+ROOT = Path(__file__).resolve().parent.parent
+ALL_DES_ROOT = ROOT.parent
+DOC_ROOT = ALL_DES_ROOT.parent
+SA_DIR = ALL_DES_ROOT / "0629_revise" / "Screening" / "筛选顺序" / "筛选顺序" / "2 配体可合成性"
+PRICE_ROOT = DOC_ROOT / "price_filter"
+COPRINET_ROOT = PRICE_ROOT / "CoPriNet-main"
+PYG201 = PRICE_ROOT / ".deps" / "pyg201_nodeps"
+
+ACTIVE_TOOLS = [
+    "parse_cif",
+    "predict_adsorption",
+    "identify_linker",
+    "check_metals",
+    "predict_sa_score",
+    "predict_aquatic_toxicity",
+    "predict_price",
+]
+
+ALLOWED_METALS = {"Mg", "Al", "Ca", "Ti", "Mn", "Fe", "Cu", "Zn", "Zr", "Ag"}
+ALL_METALS = {
+    "Li", "Be", "Na", "Mg", "Al", "K", "Ca", "Sc", "Ti", "V", "Cr", "Mn", "Fe", "Co", "Ni",
+    "Cu", "Zn", "Ga", "Rb", "Sr", "Y", "Zr", "Nb", "Mo", "Tc", "Ru", "Rh", "Pd", "Ag", "Cd",
+    "In", "Sn", "Cs", "Ba", "La", "Ce", "Pr", "Nd", "Pm", "Sm", "Eu", "Gd", "Tb", "Dy", "Ho",
+    "Er", "Tm", "Yb", "Lu", "Hf", "Ta", "W", "Re", "Os", "Ir", "Pt", "Au", "Hg", "Tl", "Pb",
+    "Bi", "Po", "Fr", "Ra", "Ac", "Th", "Pa", "U", "Np", "Pu",
+}
+
+
+def _jsonable(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {str(k): _jsonable(v) for k, v in value.items()}
+    if isinstance(value, list | tuple):
+        return [_jsonable(v) for v in value]
+    if isinstance(value, np.ndarray):
+        return _jsonable(value.tolist())
+    if isinstance(value, np.generic):
+        return _jsonable(value.item())
+    if isinstance(value, float) and (math.isnan(value) or math.isinf(value)):
+        return None
+    if isinstance(value, Path):
+        return str(value)
+    return value
+
+
+def _error(tool: str, message: str) -> dict[str, Any]:
+    return {"tool": tool, "status": "error", "summary": message, "error": message}
+
+
+def _needs_linker(tool: str) -> dict[str, Any]:
+    return {"tool": tool, "status": "unavailable", "summary": "A linker SMILES must be identified first.", "pass": None}
+
+
+def parse_cif_tool(cif_path: str) -> dict[str, Any]:
+    try:
+        structure = Structure.from_file(cif_path)
+        elements = sorted({site.specie.symbol for site in structure})
+        metals = [element for element in elements if element in ALL_METALS]
+        return {
+            "tool": "parse_cif",
+            "status": "success",
+            "formula": structure.composition.reduced_formula,
+            "num_sites": structure.num_sites,
+            "elements": elements,
+            "metals": metals,
+            "summary": f"Parsed {structure.num_sites} atomic sites; metal elements: {', '.join(metals) or 'none'}",
+        }
+    except Exception as exc:
+        return _error("parse_cif", f"CIF parsing failed: {exc}")
+
+
+def predict_adsorption_tool(cif_path: str) -> dict[str, Any]:
+    try:
+        b_desc = compute_scm_eigenvalues(cif_path, 520)
+        t_desc = compute_scm_eigenvalues(cif_path, 584)
+        benzene = predict_benzene(b_desc["eigenvalues"])
+        toluene = predict_toluene(t_desc["eigenvalues"])
+        warnings = [w for w in [b_desc.get("applicability_warning"), t_desc.get("applicability_warning")] if w]
+        return {
+            "tool": "predict_adsorption",
+            "status": "success",
+            "benzene_uptake_mg_g": benzene["uptake_mg_g"],
+            "toluene_uptake_mg_g": toluene["uptake_mg_g"],
+            "benzene_model": benzene["model_version"],
+            "toluene_model": toluene["model_version"],
+            "descriptor_raw_dim_benzene": b_desc["raw_dim"],
+            "descriptor_raw_dim_toluene": t_desc["raw_dim"],
+            "warnings": warnings,
+            "summary": f"Benzene {benzene['uptake_mg_g']} mg/g; toluene {toluene['uptake_mg_g']} mg/g",
+        }
+    except Exception as exc:
+        return _error("predict_adsorption", f"Adsorption prediction failed: {exc}")
+
+
+def identify_linker_tool(cif_path: str) -> dict[str, Any]:
+    try:
+        result = extract_linker(cif_path)
+        status = "success" if result.get("linker_smiles") else "unavailable"
+        return {
+            "tool": "identify_linker",
+            "status": status,
+            "metals": result.get("metals") or [],
+            "linker_smiles": result.get("linker_smiles"),
+            "linker_name": result.get("linker_name"),
+            "linker_formula": result.get("linker_formula"),
+            "extraction_level": result.get("extraction_level"),
+            "extraction_note": result.get("extraction_note"),
+            "summary": result.get("extraction_note"),
+        }
+    except Exception as exc:
+        return _error("identify_linker", f"Linker identification failed: {exc}")
+
+
+def check_metals_tool(context: dict[str, Any], cif_path: str) -> dict[str, Any]:
+    parse = context.get("parse_cif")
+    if not parse or parse.get("status") == "error":
+        parse = parse_cif_tool(cif_path)
+    metals = parse.get("metals") or []
+    illegal = [metal for metal in metals if metal not in ALLOWED_METALS]
+    passed = len(illegal) == 0
+    return {
+        "tool": "check_metals",
+        "status": "pass" if passed else "fail",
+        "pass": passed,
+        "detected_metals": metals,
+        "illegal_metals": illegal,
+        "allowed_metals": sorted(ALLOWED_METALS),
+        "summary": "Metal check passed" if passed else f"Disallowed metals detected: {', '.join(illegal)}",
+    }
+
+
+@lru_cache(maxsize=1)
+def _sa_module():
+    path = SA_DIR / "sascore" / "sascorer.py"
+    spec = importlib.util.spec_from_file_location("mofscreen_sascorer", path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"cannot load sascorer: {path}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def predict_sa_score_tool(context: dict[str, Any]) -> dict[str, Any]:
+    smiles = _linker_smiles(context)
+    if not smiles:
+        return _needs_linker("predict_sa_score")
+    mol = Chem.MolFromSmiles(smiles)
+    if mol is None:
+        return _error("predict_sa_score", f"Invalid linker SMILES: {smiles}")
+    try:
+        score = float(_sa_module().calculateScore(mol))
+        return {
+            "tool": "predict_sa_score",
+            "status": "success",
+            "linker_smiles": smiles,
+            "sa_score": round(score, 3),
+            "scale": "1 easy, 10 hard",
+            "easy_synthesis": score <= 6.0,
+            "summary": f"SA score {score:.3f} (1 easy to synthesize, 10 hard to synthesize)",
+        }
+    except Exception as exc:
+        return _error("predict_sa_score", f"SA score calculation failed: {exc}")
+
+
+def predict_aquatic_toxicity_tool(context: dict[str, Any]) -> dict[str, Any]:
+    smiles = _linker_smiles(context)
+    if not smiles:
+        return _needs_linker("predict_aquatic_toxicity")
+    try:
+        tox = predict_toxicity(smiles)
+        values = [
+            tox.get("IBC50_Vibrio"),
+            tox.get("IGC50_Tetrahymena"),
+            tox.get("LC50_Pimephales"),
+            tox.get("LC50_Daphnia"),
+        ]
+        present = [float(v) for v in values if v is not None]
+        status = "success" if present else "unavailable"
+        tox.update({
+            "tool": "predict_aquatic_toxicity",
+            "status": status,
+            "linker_smiles": smiles,
+            "mean_toxicity": round(sum(present) / len(present), 4) if present else None,
+            "worst_toxicity": max(present) if present else None,
+            "summary": "Aquatic toxicity prediction completed" if present else "The toxicity model did not return valid endpoints",
+        })
+        return tox
+    except Exception as exc:
+        return _error("predict_aquatic_toxicity", f"Aquatic toxicity prediction failed: {exc}")
+
+
+@lru_cache(maxsize=1)
+def _price_predictor():
+    if str(PYG201) not in sys.path:
+        sys.path.insert(0, str(PYG201))
+    if str(COPRINET_ROOT) not in sys.path:
+        sys.path.insert(0, str(COPRINET_ROOT))
+
+    import torch
+
+    if not getattr(torch.load, "_mofscreen_weights_patch", False):
+        original_load = torch.load
+
+        def load_with_trusted_checkpoint(*args, **kwargs):
+            # ponytail: local trusted CoPriNet checkpoint; remove when upstream supports torch>=2.6 defaults.
+            kwargs.setdefault("weights_only", False)
+            return original_load(*args, **kwargs)
+
+        load_with_trusted_checkpoint._mofscreen_weights_patch = True
+        torch.load = load_with_trusted_checkpoint
+
+    from pricePrediction.predict.predict import GraphPricePredictor
+
+    return GraphPricePredictor(n_gpus=0, n_cpus=0, batch_size=1)
+
+
+def predict_price_tool(context: dict[str, Any]) -> dict[str, Any]:
+    smiles = _linker_smiles(context)
+    if not smiles:
+        return _needs_linker("predict_price")
+    try:
+        pred = float(next(_price_predictor().yieldPredictions([smiles])))
+        mol = Chem.MolFromSmiles(smiles)
+        mw = Chem.Descriptors.ExactMolWt(mol) if mol is not None else None
+        usd_per_mmol = math.exp(pred)
+        usd_per_g = usd_per_mmol * 1000 / mw if mw else None
+        return {
+            "tool": "predict_price",
+            "status": "success",
+            "linker_smiles": smiles,
+            "coprinet_log_usd_per_mmol": round(pred, 6),
+            "usd_per_mmol": round(usd_per_mmol, 6),
+            "usd_per_g": round(usd_per_g, 6) if usd_per_g is not None else None,
+            "exact_mol_wt": round(mw, 6) if mw else None,
+            "summary": f"CoPriNet predicted price {usd_per_mmol:.3f} USD/mmol",
+        }
+    except Exception as exc:
+        return {"tool": "predict_price", "status": "unavailable", "summary": f"CoPriNet price prediction unavailable: {exc}", "pass": None}
+
+
+def run_online_tools(cif_path: str, requested_tools: list[str]) -> dict[str, Any]:
+    context: dict[str, Any] = {}
+    trace: list[dict[str, Any]] = []
+    tools = _with_dependencies(requested_tools)
+    for name in tools:
+        result = _run_one(name, context, cif_path)
+        context[name] = _jsonable(result)
+        trace.append({"agent": "Tool Execution Agent", "action": f"ran {name}", "output": context[name]})
+    row = _row(context, Path(cif_path).stem)
+    return _jsonable({
+        "mof_id": Path(cif_path).stem,
+        "results": context,
+        "tool_results": context,
+        "row": row,
+        "agent_trace": trace,
+        "gate_status": _gate_status(context),
+        "recommendation": _recommendation(context),
+        "final_score": _score(_gate_status(context)),
+        "explanation": _explanation(context),
+        "warnings": [],
+        "errors": [],
+    })
+
+
+def _run_one(name: str, context: dict[str, Any], cif_path: str) -> dict[str, Any]:
+    if name == "parse_cif":
+        return parse_cif_tool(cif_path)
+    if name == "predict_adsorption":
+        return predict_adsorption_tool(cif_path)
+    if name == "identify_linker":
+        return identify_linker_tool(cif_path)
+    if name == "check_metals":
+        return check_metals_tool(context, cif_path)
+    if name == "predict_sa_score":
+        return predict_sa_score_tool(context)
+    if name == "predict_aquatic_toxicity":
+        return predict_aquatic_toxicity_tool(context)
+    if name == "predict_price":
+        return predict_price_tool(context)
+    return _error(name, f"Unknown tool: {name}")
+
+
+def _with_dependencies(tools: list[str]) -> list[str]:
+    out: list[str] = []
+    for tool in tools:
+        if tool in {"check_metals"}:
+            out.append("parse_cif")
+        if tool in {"predict_sa_score", "predict_aquatic_toxicity", "predict_price"}:
+            out.append("identify_linker")
+        out.append(tool)
+    return list(dict.fromkeys(out))
+
+
+def _linker_smiles(context: dict[str, Any]) -> str | None:
+    linker = context.get("identify_linker") or {}
+    smiles = linker.get("linker_smiles")
+    return str(smiles) if smiles else None
+
+
+def _gate_status(results: dict[str, Any]) -> str:
+    if any(r.get("status") == "fail" for r in results.values()):
+        return "failed"
+    if any(r.get("status") in {"error", "unavailable"} for r in results.values()):
+        return "incomplete"
+    return "completed"
+
+
+def _recommendation(results: dict[str, Any]) -> str:
+    if any(r.get("status") == "fail" for r in results.values()):
+        return "not_recommended"
+    if any(r.get("status") in {"error", "unavailable"} for r in results.values()):
+        return "needs_manual_review"
+    return "continue_evaluation"
+
+
+def _score(status: str) -> float:
+    return {"completed": 10.0, "incomplete": 5.0, "failed": 0.0}.get(status, 5.0)
+
+
+def _explanation(results: dict[str, Any]) -> str:
+    return "; ".join(f"{name}: {payload.get('status')}" for name, payload in results.items())
+
+
+def _row(results: dict[str, Any], mof_id: str) -> dict[str, Any]:
+    adsorption = results.get("predict_adsorption") or {}
+    metals = results.get("check_metals") or {}
+    linker = results.get("identify_linker") or {}
+    sa = results.get("predict_sa_score") or {}
+    tox = results.get("predict_aquatic_toxicity") or {}
+    price = results.get("predict_price") or {}
+    return {
+        "MOF ID": mof_id,
+        "Benzene (mg/g)": adsorption.get("benzene_uptake_mg_g"),
+        "Toluene (mg/g)": adsorption.get("toluene_uptake_mg_g"),
+        "Metal status": metals.get("status"),
+        "Metals": ", ".join(metals.get("detected_metals") or []),
+        "Linker": linker.get("linker_name") or linker.get("linker_formula"),
+        "Linker SMILES": linker.get("linker_smiles"),
+        "SA score": sa.get("sa_score"),
+        "Mean toxicity": tox.get("mean_toxicity"),
+        "Price USD/g": price.get("usd_per_g"),
+        "Price USD/mmol": price.get("usd_per_mmol"),
+        "Gate status": _gate_status(results),
+        "Recommendation": _recommendation(results),
+    }

tools/six_step_screening.py

@@ -0,0 +1,550 @@
+"""Deterministic six-step MOF screening tools backed by precomputed tables."""
+from __future__ import annotations
+
+import math
+import re
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+import joblib
+import numpy as np
+import pandas as pd
+
+
+ROOT = Path(__file__).resolve().parent.parent
+ALL_DES_ROOT = ROOT.parent
+DOC_ROOT = ALL_DES_ROOT.parent
+REVISE_ROOT = ALL_DES_ROOT / "0629_revise"
+ORDER_ROOT = REVISE_ROOT / "Screening" / "筛选顺序" / "筛选顺序"
+PRICE_ROOT = DOC_ROOT / "price_filter" / "mof_price_0707"
+
+STEP1_FULL = REVISE_ROOT / "screen_result" / "all_mofs_sorted_by_balanced_score_with_YEYVOO_clean5b_only.csv"
+STEP1_TOP20 = REVISE_ROOT / "screen_result" / "top20_percent_mofs_by_balanced_score_with_YEYVOO_clean5b_only.csv"
+STEP2_DIR = ORDER_ROOT / "1重金属结果"
+STEP3_SA = ORDER_ROOT / "2 配体可合成性" / "All_MOF_SA_Ranking.csv"
+STEP3_TOP6000 = ORDER_ROOT / "3 水生物毒性" / "Top6000_MOF_MaxSA_Linker.csv"
+STEP4_TOX = ORDER_ROOT / "3 水生物毒性" / "Strict_EasySynth_Linkers_Optimized_Predictions.csv"
+STEP5_DIR = ORDER_ROOT / "4 PMT"
+STEP5_DESC = STEP5_DIR / "PMT描述符.csv"
+STEP6_PRICE = PRICE_ROOT / "Strict_EasySynth_Linkers_Toxicity_Ranking0705_CoPriNet_price_ranked.csv"
+
+SIX_STEP_TOOLS = [
+    "adsorption_screen",
+    "heavy_metal",
+    "ligand_sa",
+    "aquatic_toxicity",
+    "pmt",
+    "price",
+]
+
+ALLOWED_METALS = {"Mg", "Al", "Ca", "Ti", "Mn", "Fe", "Cu", "Zn", "Zr", "Ag"}
+ALL_METALS = {
+    "Li", "Be", "Na", "Mg", "Al", "K", "Ca", "Sc", "Ti", "V", "Cr", "Mn", "Fe", "Co", "Ni",
+    "Cu", "Zn", "Ga", "Rb", "Sr", "Y", "Zr", "Nb", "Mo", "Tc", "Ru", "Rh", "Pd", "Ag", "Cd",
+    "In", "Sn", "Cs", "Ba", "La", "Ce", "Pr", "Nd", "Pm", "Sm", "Eu", "Gd", "Tb", "Dy", "Ho",
+    "Er", "Tm", "Yb", "Lu", "Hf", "Ta", "W", "Re", "Os", "Ir", "Pt", "Au", "Hg", "Tl", "Pb",
+    "Bi", "Po", "Fr", "Ra", "Ac", "Th", "Pa", "U", "Np", "Pu",
+}
+
+
+def normalize_mof_name(value: Any) -> str:
+    """Normalize MOF ids for table and filename matching."""
+    if value is None:
+        return ""
+    return re.sub(r"[\s\-_()]+", "", str(value)).lower()
+
+
+def _smiles_key(value: Any) -> str:
+    return "" if value is None else re.sub(r"\s+", "", str(value))
+
+
+@lru_cache(maxsize=16)
+def _csv(path: str) -> pd.DataFrame:
+    return pd.read_csv(path)
+
+
+def _read(path: Path) -> pd.DataFrame:
+    return _csv(str(path))
+
+
+def _find_by_name(df: pd.DataFrame, name: str | None, columns: tuple[str, ...] = ("MOF_Name", "MOF")) -> pd.Series | None:
+    if not name:
+        return None
+    key = normalize_mof_name(name)
+    for column in columns:
+        if column not in df.columns:
+            continue
+        mask = df[column].map(normalize_mof_name) == key
+        if mask.any():
+            return df.loc[mask].iloc[0]
+    return None
+
+
+def _find_by_smiles(df: pd.DataFrame, smiles: str | None, column: str = "SMILES") -> pd.Series | None:
+    if not smiles or column not in df.columns:
+        return None
+    key = _smiles_key(smiles)
+    mask = df[column].map(_smiles_key) == key
+    if mask.any():
+        return df.loc[mask].iloc[0]
+    return None
+
+
+def _num(value: Any) -> float | None:
+    try:
+        if pd.isna(value):
+            return None
+        value = float(value)
+        if math.isnan(value) or math.isinf(value):
+            return None
+        return value
+    except Exception:
+        return None
+
+
+def _int(value: Any) -> int | None:
+    number = _num(value)
+    return None if number is None else int(number)
+
+
+def _jsonable(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {str(k): _jsonable(v) for k, v in value.items()}
+    if isinstance(value, list | tuple):
+        return [_jsonable(v) for v in value]
+    if isinstance(value, np.ndarray):
+        return _jsonable(value.tolist())
+    if isinstance(value, np.generic):
+        return _jsonable(value.item())
+    if isinstance(value, float) and (math.isnan(value) or math.isinf(value)):
+        return None
+    return value
+
+
+def _unknown(tool: str, reason: str, step: int) -> dict[str, Any]:
+    return {"tool": tool, "step": step, "status": "unknown", "reason": reason, "pass": None}
+
+
+def resolve_candidate(cif_path: str | None = None, user_text: str = "") -> dict[str, Any]:
+    """Resolve uploaded filename or user text to a known MOF candidate."""
+    query_names: list[str] = []
+    if cif_path:
+        query_names.append(Path(cif_path).stem)
+    query_names.extend(re.findall(r"[A-Za-z0-9]+(?:[-_][A-Za-z0-9]+)*(?:-\(id[:_]\d+\))?", user_text or ""))
+
+    candidate: dict[str, Any] = {
+        "cif_path": cif_path,
+        "query_names": list(dict.fromkeys([q for q in query_names if q])),
+        "matched_mof": None,
+        "linker_smiles": None,
+        "match_source": None,
+    }
+
+    for path, columns, source in [
+        (STEP1_FULL, ("MOF_Name", "MOF"), "adsorption_screen"),
+        (STEP3_SA, ("MOF", "MOF_Name"), "ligand_sa"),
+        (STEP3_TOP6000, ("MOF", "MOF_Name"), "top6000_linker"),
+        (STEP6_PRICE, ("MOF", "MOF_Name"), "price"),
+    ]:
+        if not path.exists():
+            continue
+        df = _read(path)
+        for name in candidate["query_names"]:
+            row = _find_by_name(df, name, columns)
+            if row is not None:
+                matched = row.get("MOF") if "MOF" in row.index else row.get("MOF_Name")
+                candidate["matched_mof"] = str(matched)
+                candidate["match_source"] = source
+                if "SMILES" in row.index and pd.notna(row.get("SMILES")):
+                    candidate["linker_smiles"] = str(row.get("SMILES"))
+                return candidate
+
+    return candidate
+
+
+def adsorption_screen_tool(candidate: dict[str, Any]) -> dict[str, Any]:
+    """Step 1: precomputed top-20% adsorption screen."""
+    if not STEP1_FULL.exists() or not STEP1_TOP20.exists():
+        return _unknown("adsorption_screen", "adsorption ranking table is missing", 1)
+
+    full = _read(STEP1_FULL)
+    top = _read(STEP1_TOP20)
+    name = candidate.get("matched_mof") or next(iter(candidate.get("query_names", [])), None)
+    row = _find_by_name(full, name)
+    if row is None:
+        return _unknown("adsorption_screen", "MOF was not found in the precomputed adsorption ranking", 1)
+
+    mof = str(row.get("MOF_Name"))
+    top_row = _find_by_name(top, mof)
+    rank = _int(row.get("rank"))
+    cutoff = len(top)
+    passed = top_row is not None
+    candidate["matched_mof"] = mof
+    return {
+        "tool": "adsorption_screen",
+        "step": 1,
+        "status": "pass" if passed else "fail",
+        "pass": passed,
+        "mof": mof,
+        "rank": rank,
+        "top20_cutoff_rank": cutoff,
+        "predicted_benzene_adsorption": _num(row.get("predicted_benzene_adsorption")),
+        "predicted_toluene_adsorption": _num(row.get("predicted_toluene_adsorption")),
+        "balanced_score": _num(row.get("balanced_score")),
+        "model": "precomputed benzene/toluene adsorption ranking from 0629_revise/screen_result",
+    }
+
+
+def heavy_metal_tool(candidate: dict[str, Any]) -> dict[str, Any]:
+    """Step 2: allowed-metal filter from uploaded CIF."""
+    cif_path = candidate.get("cif_path")
+    if not cif_path:
+        return _unknown("heavy_metal", "no CIF file was provided for metal parsing", 2)
+
+    try:
+        from pymatgen.core import Structure
+
+        structure = Structure.from_file(cif_path)
+        elements = sorted({str(site.specie.symbol) for site in structure})
+    except Exception as exc:
+        return _unknown("heavy_metal", f"failed to parse CIF metals: {exc}", 2)
+
+    metals = [element for element in elements if element in ALL_METALS]
+    illegal = [element for element in metals if element not in ALLOWED_METALS]
+    passed = len(illegal) == 0
+    return {
+        "tool": "heavy_metal",
+        "step": 2,
+        "status": "pass" if passed else "fail",
+        "pass": passed,
+        "detected_metals": metals,
+        "illegal_metals": illegal,
+        "allowed_metals": sorted(ALLOWED_METALS),
+        "model": "pymatgen CIF parser + fixed allowed-metal list",
+    }
+
+
+def ligand_sa_tool(candidate: dict[str, Any]) -> dict[str, Any]:
+    """Step 3: ligand synthesizability lookup."""
+    if not STEP3_SA.exists() or not STEP3_TOP6000.exists():
+        return _unknown("ligand_sa", "ligand SA tables are missing", 3)
+
+    name = candidate.get("matched_mof") or next(iter(candidate.get("query_names", [])), None)
+    sa = _read(STEP3_SA)
+    top = _read(STEP3_TOP6000)
+    row = _find_by_name(sa, name)
+    if row is None:
+        return _unknown("ligand_sa", "MOF was not found in All_MOF_SA_Ranking.csv", 3)
+
+    mof = str(row.get("MOF"))
+    top_row = _find_by_name(top, mof)
+    passed = top_row is not None
+    smiles = str(top_row.get("SMILES")) if top_row is not None and pd.notna(top_row.get("SMILES")) else None
+    candidate["matched_mof"] = mof
+    if smiles:
+        candidate["linker_smiles"] = smiles
+
+    return {
+        "tool": "ligand_sa",
+        "step": 3,
+        "status": "pass" if passed else "fail",
+        "pass": passed,
+        "mof": mof,
+        "sa_rank": _int(row.get("SA_Rank")),
+        "sa_score": _num(row.get("Max_SA")),
+        "mean_sa": _num(row.get("Mean_SA")),
+        "n_linker": _int(row.get("NLinker")),
+        "in_top6000": passed,
+        "linker_id": _int(top_row.get("Linker_ID")) if top_row is not None else None,
+        "linker_smiles": smiles,
+        "model": "All_MOF_SA_Ranking.csv + Top6000_MOF_MaxSA_Linker.csv lookup",
+    }
+
+
+def aquatic_toxicity_tool(candidate: dict[str, Any]) -> dict[str, Any]:
+    """Step 4: aquatic toxicity lookup by linker SMILES."""
+    if not STEP4_TOX.exists():
+        return _unknown("aquatic_toxicity", "aquatic toxicity prediction table is missing", 4)
+    if not candidate.get("linker_smiles"):
+        ligand_sa_tool(candidate)
+    smiles = candidate.get("linker_smiles")
+    if not smiles:
+        return _unknown("aquatic_toxicity", "no linker SMILES was available for toxicity lookup", 4)
+
+    row = _find_by_smiles(_read(STEP4_TOX), smiles)
+    if row is None:
+        return _unknown("aquatic_toxicity", "linker SMILES was not found in toxicity predictions", 4)
+
+    values = [
+        _num(row.get("Predicted_Tox_IBC50")),
+        _num(row.get("Predicted_Tox_IGC50")),
+        _num(row.get("Predicted_Tox_LC50")),
+        _num(row.get("Predicted_Tox_LC50DM")),
+    ]
+    present = [v for v in values if v is not None]
+    return {
+        "tool": "aquatic_toxicity",
+        "step": 4,
+        "status": "pass" if len(present) == 4 else "unknown",
+        "pass": True if len(present) == 4 else None,
+        "linker_smiles": smiles,
+        "linker_id": _int(row.get("Linker_ID")),
+        "IBC50": values[0],
+        "IGC50": values[1],
+        "LC50": values[2],
+        "LC50DM": values[3],
+        "mean_toxicity": round(sum(present) / len(present), 6) if present else None,
+        "worst_toxicity": min(present) if present else None,
+        "model": "Strict_EasySynth_Linkers_Optimized_Predictions.csv lookup",
+    }
+
+
+def pmt_tool(candidate: dict[str, Any]) -> dict[str, Any]:
+    """Step 5: PMT classifier lookup + model inference from precomputed descriptors."""
+    if not STEP5_DESC.exists():
+        return _unknown("pmt", "PMT descriptor table is missing", 5)
+    if not candidate.get("linker_smiles"):
+        ligand_sa_tool(candidate)
+    smiles = candidate.get("linker_smiles")
+    if not smiles:
+        return _unknown("pmt", "no linker SMILES was available for PMT descriptor lookup", 5)
+
+    desc = _read(STEP5_DESC)
+    row = _find_by_smiles(desc, smiles)
+    if row is None:
+        return _unknown("pmt", "linker SMILES was not found in PMT描述符.csv", 5)
+
+    try:
+        imputer = joblib.load(STEP5_DIR / "PMT_imputer.pkl")
+        scaler = joblib.load(STEP5_DIR / "PMT_scaler.pkl")
+        selector = joblib.load(STEP5_DIR / "PMT_selector.pkl")
+        model = joblib.load(STEP5_DIR / "PMT_xgb_model.pkl")
+        feature_row = row.drop(labels=["SMILES"], errors="ignore").to_frame().T
+        feature_row = feature_row.apply(pd.to_numeric, errors="coerce")
+        x = imputer.transform(feature_row)
+        x = scaler.transform(x)
+        x = selector.transform(x)
+        proba = model.predict_proba(x)[0]
+        classes = list(getattr(model, "classes_", [0, 1]))
+        positive_index = classes.index(1) if 1 in classes else len(proba) - 1
+        probability = float(proba[positive_index])
+    except Exception as exc:
+        return _unknown("pmt", f"PMT model inference failed: {exc}", 5)
+
+    threshold = 0.4
+    passed = probability >= threshold
+    return {
+        "tool": "pmt",
+        "step": 5,
+        "status": "pass" if passed else "fail",
+        "pass": passed,
+        "linker_smiles": smiles,
+        "pmt_probability": round(probability, 6),
+        "pmt_class": "non_PMT" if passed else "PMT_risk",
+        "threshold": threshold,
+        "threshold_note": "Class 1 is treated as non-PMT/pass; probability >= 0.4 passes.",
+        "model": "PMT_xgb_model.pkl with PMT_imputer/scaler/selector",
+    }
+
+
+def price_tool(candidate: dict[str, Any]) -> dict[str, Any]:
+    """Step 6: CoPriNet price lookup."""
+    if not STEP6_PRICE.exists():
+        return _unknown("price", "price ranking table is missing", 6)
+
+    price = _read(STEP6_PRICE)
+    row = _find_by_name(price, candidate.get("matched_mof"))
+    if row is None and candidate.get("linker_smiles"):
+        row = _find_by_smiles(price, candidate.get("linker_smiles"))
+    if row is None:
+        for name in candidate.get("query_names", []):
+            row = _find_by_name(price, name)
+            if row is not None:
+                break
+    if row is None:
+        return _unknown("price", "MOF/linker was not found in the CoPriNet price ranking", 6)
+
+    status = str(row.get("Price_Prediction_Status", "unknown"))
+    passed = status.upper() == "OK"
+    if pd.notna(row.get("SMILES")):
+        candidate["linker_smiles"] = str(row.get("SMILES"))
+    if pd.notna(row.get("MOF")):
+        candidate["matched_mof"] = str(row.get("MOF"))
+    return {
+        "tool": "price",
+        "step": 6,
+        "status": "pass" if passed else "fail",
+        "pass": passed,
+        "mof": str(row.get("MOF")),
+        "linker_smiles": str(row.get("SMILES")) if pd.notna(row.get("SMILES")) else None,
+        "coprinet_price_rank": _int(row.get("CoPriNet_Price_Rank")),
+        "usd_per_g": _num(row.get("CoPriNet_USD_per_g")),
+        "usd_per_mmol": _num(row.get("CoPriNet_USD_per_mmol")),
+        "price_status": status,
+        "model": "CoPriNet price ranking CSV lookup; independent of PMT",
+    }
+
+
+def _run_tool(name: str, candidate: dict[str, Any]) -> dict[str, Any]:
+    tools = {
+        "adsorption_screen": adsorption_screen_tool,
+        "heavy_metal": heavy_metal_tool,
+        "ligand_sa": ligand_sa_tool,
+        "aquatic_toxicity": aquatic_toxicity_tool,
+        "pmt": pmt_tool,
+        "price": price_tool,
+    }
+    if name not in tools:
+        return _unknown(name, "unknown six-step tool name", 0)
+    return _jsonable(tools[name](candidate))
+
+
+def run_selected_six_step_tools(cif_path: str | None, user_text: str, tools: list[str]) -> dict[str, Any]:
+    candidate = resolve_candidate(cif_path, user_text)
+    results: dict[str, Any] = {}
+    trace: list[dict[str, Any]] = []
+    for name in tools:
+        if name in {"aquatic_toxicity", "pmt", "price"} and not candidate.get("linker_smiles"):
+            ligand = results.get("ligand_sa") or _run_tool("ligand_sa", candidate)
+            results.setdefault("ligand_sa", ligand)
+            trace.append({"agent": "Tool Execution Agent", "action": "resolved linker SMILES via ligand_sa", "output": ligand})
+        result = _run_tool(name, candidate)
+        results[name] = result
+        trace.append({"agent": "Tool Execution Agent", "action": f"ran {name}", "output": result})
+
+    payload = _final_payload(candidate, results, full=False)
+    payload["agent_trace"] = trace + payload["agent_trace"]
+    return payload
+
+
+def run_six_step_screening(cif_path: str | None, user_text: str = "") -> dict[str, Any]:
+    candidate = resolve_candidate(cif_path, user_text)
+    results: dict[str, Any] = {}
+    trace: list[dict[str, Any]] = []
+    for name in SIX_STEP_TOOLS:
+        result = _run_tool(name, candidate)
+        results[name] = result
+        trace.append({"agent": "Tool Execution Agent", "action": f"ran step {result.get('step')}: {name}", "output": result})
+    payload = _final_payload(candidate, results, full=True)
+    payload["agent_trace"] = trace + payload["agent_trace"]
+    return payload
+
+
+def _final_payload(candidate: dict[str, Any], results: dict[str, Any], full: bool) -> dict[str, Any]:
+    failed = [r for r in results.values() if r.get("status") == "fail"]
+    unknown = [r for r in results.values() if r.get("status") == "unknown"]
+    if failed:
+        failed_first = sorted(failed, key=lambda r: r.get("step", 999))[0]
+        gate_status = "failed"
+        failed_at_step = failed_first.get("step")
+        recommendation = f"failed_at_step_{failed_at_step}_{failed_first.get('tool')}"
+    elif unknown:
+        gate_status = "incomplete"
+        failed_at_step = None
+        recommendation = "incomplete_due_to_unknown_evidence"
+    else:
+        gate_status = "pass_full_screening" if full else "pass_selected_tools"
+        failed_at_step = None
+        recommendation = gate_status
+
+    decision_record = {
+        "decision_class": gate_status,
+        "recommendation": recommendation,
+        "failed_at_step": failed_at_step,
+        "unknown_steps": [r.get("step") for r in unknown],
+        "blocking_tools": [r.get("tool") for r in failed],
+        "full_screening": full,
+    }
+    payload = {
+        "mof_id": candidate.get("matched_mof") or next(iter(candidate.get("query_names", [])), "unknown"),
+        "candidate": candidate,
+        "six_step": results,
+        "results": results,
+        "gate_status": gate_status,
+        "failed_at_step": failed_at_step,
+        "recommendation": recommendation,
+        "decision_record": decision_record,
+        "final_score": _score_from_status(gate_status),
+        "explanation": _summary(results, gate_status, failed_at_step),
+        "warnings": [],
+        "errors": [],
+        "agent_trace": [{
+            "agent": "Decision Agent",
+            "action": "computed deterministic six-step gate status",
+            "output": decision_record,
+        }],
+    }
+    payload["row"] = build_result_row(payload)
+    payload["evidence_ledger"] = [
+        {"tool": name, "status": result.get("status"), "outputs": result, "confidence": "precomputed_or_deterministic"}
+        for name, result in results.items()
+    ]
+    return _jsonable(payload)
+
+
+def _score_from_status(status: str) -> float | None:
+    if status.startswith("pass"):
+        return 10.0
+    if status == "incomplete":
+        return 5.0
+    if status == "failed":
+        return 0.0
+    return None
+
+
+def _summary(results: dict[str, Any], gate_status: str, failed_at_step: int | None) -> str:
+    parts = []
+    for name in SIX_STEP_TOOLS:
+        result = results.get(name)
+        if not result:
+            continue
+        parts.append(f"step {result.get('step')} {name}: {result.get('status')}")
+    if failed_at_step:
+        tail = f"Final gate status is failed at step {failed_at_step}."
+    elif gate_status == "incomplete":
+        tail = "Final gate status is incomplete because at least one required tool returned unknown."
+    else:
+        tail = f"Final gate status is {gate_status}."
+    return "; ".join(parts + [tail])
+
+
+def _fmt_status(result: dict[str, Any] | None) -> str:
+    if not result:
+        return "N/A"
+    return str(result.get("status", "unknown"))
+
+
+def build_result_row(result: dict[str, Any]) -> dict[str, Any]:
+    steps = result.get("six_step") or result.get("results") or {}
+    adsorption = steps.get("adsorption_screen") or {}
+    metal = steps.get("heavy_metal") or {}
+    sa = steps.get("ligand_sa") or {}
+    tox = steps.get("aquatic_toxicity") or {}
+    pmt = steps.get("pmt") or {}
+    price = steps.get("price") or {}
+
+    return {
+        "MOF ID": result.get("mof_id", "unknown"),
+        "Step 1 adsorption": (
+            f"{_fmt_status(adsorption)}; rank={adsorption.get('rank')}; "
+            f"B={adsorption.get('predicted_benzene_adsorption')}; T={adsorption.get('predicted_toluene_adsorption')}"
+        ),
+        "Step 2 metal": (
+            f"{_fmt_status(metal)}; metals={', '.join(metal.get('detected_metals', []) or [])}; "
+            f"illegal={', '.join(metal.get('illegal_metals', []) or [])}"
+        ),
+        "Step 3 SA": f"{_fmt_status(sa)}; rank={sa.get('sa_rank')}; score={sa.get('sa_score')}",
+        "Step 4 toxicity": (
+            f"{_fmt_status(tox)}; mean={tox.get('mean_toxicity')}; worst={tox.get('worst_toxicity')}"
+        ),
+        "Step 5 PMT": (
+            f"{_fmt_status(pmt)}; class={pmt.get('pmt_class')}; prob={pmt.get('pmt_probability')}"
+        ),
+        "Step 6 price": (
+            f"{_fmt_status(price)}; rank={price.get('coprinet_price_rank')}; USD/g={price.get('usd_per_g')}"
+        ),
+        "Final gate status": result.get("gate_status", "unknown"),
+        "Recommendation": result.get("recommendation", "N/A"),
+        "Score": result.get("final_score"),
+    }