# Project export: ScalePilot This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: UC Berkeley AI Hackathon 2026 - Tagline: ScalePilot turns lab-scale process briefs into transparent equipment recommendations, CAPEX estimates, and RFQ drafts for faster early-stage chemical engineering process scale-up. - Devpost: https://devpost.com/software/scalepilot - GitHub: https://github.com/gjh2025-bot/process-equipment-agent - Demo: https://agentverse.ai/agents/details/agent1qfc9fmma5nwq7qgvkpex60qwnmvk8wtkaenvnt3ctt60jk2lq358wzn73zs/profile - Video: https://www.youtube.com/embed/nbt6fLmxiO4?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 0 GitHub contributor(s) — ## Devpost submission (written by the team) ### Inspiration In many chemical engineering and materials research projects, there is a difficult gap between a successful lab-scale recipe and a practical pilot-scale process. Researchers may know the chemistry, reaction conditions, product requirements, and major concerns, but turning that information into equipment choices is still slow and messy. Early process scale-up requires engineers to think about reactors, separation methods, drying equipment, material compatibility, pressure and temperature limits, contamination control, supplier communication, and rough cost ranges. For students, researchers, and early-stage technical teams, this information is often scattered across textbooks, supplier websites, engineering experience, and informal notes. ScalePilot was inspired by this gap. I wanted to build a transparent AI-assisted workflow that helps users move from a lab-scale process brief toward a structured preliminary equipment recommendation. The goal is not to replace qualified process engineers, but to make early process scale-up easier to understand, organize, and communicate. ### What it does ScalePilot is an ASI process equipment selection agent for early-stage chemical process scale-up. Given a structured lab-scale process brief, ScalePilot runs a modular engineering workflow that: Builds a preliminary process basis Identifies the target product, operating conditions, scale-up factor, and missing data Decomposes the process into major unit operations Searches a mock equipment database for possible equipment candidates Screens candidates based on temperature, pressure, volume, material compatibility, and contamination concerns Explains why each equipment candidate is accepted or rejected Matches recommended equipment with mock suppliers Estimates a rough CAPEX range using mock data Generates a supplier inquiry draft that users could later turn into an RFQ-style message The demo case focuses on scaling a lab-scale hydrothermal synthesis process for MnO₂ nanoparticle powder used in aqueous zinc-ion battery cathodes. The system produces both a structured JSON output and a human-readable Markdown report. ScalePilot is also wrapped as a Fetch.ai uAgent and can be reached through ASI using the chat protocol. Users can interact with it conversationally by asking prompts like “analyze,” “why the Teflon-lined autoclave?”, or “rfq.” The deterministic engineering workflow remains the source of truth, while the optional LLM layer only makes the response more conversational. How I built it I built ScalePilot with Claude Code and Codex as a local-first Python MVP, then wrapped the working workflow into an ASI Fetch.ai uAgent. This helped me keep the core engineering logic simple, testable, and transparent while also making the project usable as a conversational agent. The workflow is organized into modular components: process_basis.py extracts the product, scale, reaction conditions, priorities, and missing data. unit_decomposer.py breaks the process into unit operations such as precursor preparation, hydrothermal reaction, cooling, separation, washing, drying, and powder collection. equipment_selector.py searches a mock equipment database for candidates that match each unit operation. engineering_reviewer.py screens and ranks equipment candidates using engineering constraints. supplier_matcher.py connects recommended equipment to mock supplier options and rough cost ranges. rfq_generator.py creates a supplier inquiry draft based on the selected equipment and operating requirements. main.py orchestrates the local workflow and generates the final output files. agent.py wraps the workflow as a uAgent so users can interact with ScalePilot through ASI. The MVP uses structured JSON input files and mock equipment and supplier databases. I intentionally avoided hidden APIs, real procurement actions, and external dependencies in the core workflow so that the project could run locally and be easy to explain during a hackathon demo. Challenges I ran into One major challenge was scope control. Process scale-up is a huge engineering problem, and it would be easy to overbuild the project by adding real supplier search, ASPEN integration, web apps, complex multi-agent orchestration, or detailed process simulation too early. I had to keep reminding myself that the first goal was a clean, runnable MVP. Another challenge was translating engineering reasoning into modular software logic. Equipment selection is not just keyword matching. The system needs to consider temperature, pressure, working volume, material compatibility, contamination risk, phase suitability, and unresolved assumptions. I had to design the workflow so that the reasoning was clear enough for users to inspect. A later challenge was adapting the local workflow into an agent that could be used through ASI. I had to make sure the conversational layer did not replace the engineering logic or invent unsupported data. The final design keeps the deterministic workflow as the source of truth and uses the LLM only to explain the results conversationally. I also had to be careful about ethical and safety boundaries. ScalePilot should not pretend to produce a certified process design. The report includes limitations and clearly states that the output is preliminary and requires review by a qualified process engineer before procurement or implementation. Finally, I had to work in an unfamiliar software development environment under a very limited hackathon timeline and limited API credit. This included learning how to use the terminal, structure a Python project, debug the workflow, publish an actual project on GitHub, and connect an agent to Agentverse and ASI. Accomplishments that I'm proud of I am proud that ScalePilot runs as a complete local workflow from a sample process input to final output files. The system generates both a structured JSON result and a readable Markdown report that explains the process summary, unit operations, equipment recommendations, screening reasoning, supplier matches, rough CAPEX, and next steps. I am also proud that ScalePilot is now wrapped as a conversational uAgent and can be used through ASI. This turns the project from a local command-line MVP into an interactive agent demo where users can ask follow-up questions, request reasoning, and generate a supplier inquiry draft. I am especially proud of the transparency of the recommendation logic. Instead of only giving a final answer, ScalePilot shows accepted candidates, rejected candidates, reasoning, missing data, and unresolved risks. This makes the system more useful as an educational and decision-support tool. On a personal level, I am proud of what I accomplished in the past 24 hours. This was my first time seriously using the terminal, my first time publishing an actual project on GitHub, and my first time pushing myself to build a software project from 0 to 1 in an unfamiliar field. As someone with a chemical engineering background, this hackathon challenged me to step outside my comfort zone and turn an engineering idea into a runnable AI agent prototype. Most importantly, I built something that connects real chemical engineering thinking with an AI agent workflow in a practical and explainable way. What I learned I learned that building an AI engineering assistant starts with designing a good workflow, not just adding an LLM or agent layer. Before a system can become a useful agent, it needs clear inputs, outputs, responsibilities, assumptions, and boundaries. I also learned how important transparency is in technical AI tools. In engineering contexts, users need to know why a recommendation was made, what assumptions were used, and what risks remain. A confident but unexplained answer is not enough. From the software side, I learned how to structure a local MVP so that it can later be adapted into a multi-agent system. Keeping the logic modular made the project easier to debug, explain, and extend. From the engineering side, I learned how many decisions are involved even in a “simple” early scale-up problem: equipment type, operating limits, contamination control, product handling, supplier questions, cost ranges, and missing process data all matter. More broadly, I learned that AI can be a strong boost for engineering fields beyond computer science. As engineers increasingly collaborate with software engineers and AI systems, I believe we should also learn to adapt AI tools ourselves. This helps us better understand how this technology can change future engineering work, and this hackathon was my way of pushing myself to start that process. ### What's next ScalePilot is now wrapped as a Fetch.ai uAgent and can be reached through ASI as a conversational manager agent. The next step is to move from a single manager agent to a true multi-agent workflow, where specialist agents handle reactor selection, solid-liquid separation, drying, supplier matching, and supplier inquiry drafting. Future versions could also add human-in-the-loop equipment selection, more process examples beyond MnO₂ nanoparticle synthesis, support for less-structured user inputs, curated real equipment and supplier databases, improved cost estimation, and stronger privacy controls for proprietary process data. Long term, ScalePilot could become a transparent AI co-pilot for early process scale-up, helping researchers and engineers move from lab chemistry to practical process design faster, while keeping human engineering judgment at the center. ## README (from the GitHub repository) # ScalePilot Process Equipment Selection Assistant for Early Process Scale-Up A local-first engineering prototype that turns a lab-scale process brief into a **preliminary equipment recommendation report**. The demo case is the scale-up of a lab-scale hydrothermal synthesis of **MnO₂ nanoparticle powder** for aqueous zinc-ion battery cathodes. It is an early-stage **decision-support** tool, not a certified process design tool. All equipment and supplier data is mock data for demonstration. ## What it does Given a structured process brief, it runs a simple, deterministic workflow: ``` data/sample_input.json -> process_basis (product, scale, scale-up factor, missing data) -> unit_decomposer (break process into unit operations) -> equipment_selector (find candidate equipment from a mock database) -> engineering_reviewer (screen on temperature / pressure / volume, recommend) -> supplier_matcher (match mock suppliers, rough CAPEX range) -> rfq_generator (draft an RFQ-style supplier inquiry) -> outputs/sample_output.json (structured result) -> outputs/sample_report.md (human-readable report) ``` Every step is rule-based and transparent: the report shows **why** each equipment candidate was accepted or rejected. ## Requirements - Python 3.9+ (developed on 3.11). - The **local workflow** (`main.py` / `src/`) uses the **standard library only**. - The **ASI:One agent** (`agent.py`) additionally needs `uagents` (and `openai` for the optional LLM layer): `pip install -r requirements.txt openai`. ## How to run From the project root: ```bash python main.py --input data/sample_input.json ``` On Windows, if `python` is not found (the Microsoft Store stub), use the Python launcher instead: ```powershell py main.py --input data/sample_input.json ``` This writes two files into `outputs/`: - `outputs/sample_output.json` — full structured result - `outputs/sample_report.md` — readable report for demo ## ASI:One agent The same workflow is wrapped as a conversational [Fetch.ai uAgent](https://fetch.ai/) (`agent.py`) using the chat protocol, so it can be reached from [ASI:One](https://asi1.ai). The deterministic workflow is the source of truth; an optional ASI:One LLM layer only makes the replies conversational and never invents equipment, prices, or suppliers. ```bash # Standard-library workflow only needs Python; the agent needs uagents: pip install -r requirements.txt # Optional: enable the conversational LLM layer (else it uses rule-based replies) # PowerShell: $env:ASI_ONE_API_KEY = "your-key" # bash: export ASI_ONE_API_KEY=your-key python agent.py # on Windows: py agent.py ``` On startup the agent prints an **Agent Inspector** link; open it (while the agent is running) to connect an Agentverse mailbox, after which the agent is reachable from ASI:One. Sample prompts: `analyze`, `rfq`, or `what equipment do you recommend for my MnO₂ process?`. See `AGENT_README.md` for the Agentverse profile text. ## Project structure ``` hackathon-project/ main.py # workflow entry point / orchestration agent.py # ASI:One-compatible uAgent wrapping the workflow AGENT_README.md # Agentverse profile text for the agent requirements.txt # uagents (workflow itself is standard-library only) data/ sample_input.json # the demo process brief equipment_database.json # mock equipment knowledge base supplier_database.json # mock supplier + CAPEX data src/ process_basis.py unit_decomposer.py equipment_selector.py engineering_reviewer.py supplier_matcher.py rfq_generator.py outputs/ sample_output.json # generated sample_report.md # generated docs/ project_guide.md architecture.md ``` ## Example result (MnO₂ demo) For the 2 L bench-pilot batch, the workflow recommends, for example: | Unit operation | Recommended equipment | | --------------------------- | --------------------------------------------- | | Hydrothermal batch reaction | Teflon-lined stainless steel batch autoclave | | Solid-liquid separation | Lab or pilot vacuum filtration setup | | Vacuum drying | Vacuum oven | - Rough total CAPEX (mock, deduplicated): **$11,550 – $278,000 USD** - Equipment shared across steps (e.g. the reactor used for both reaction and cooling) is counted once in the CAPEX total. ## Scope and limitations - Preliminary and illustrative only — **requires review by a qualified process engineer before any procurement**. - No process simulation, heat/mass balance, or pressure-vessel design. - No real web search, supplier APIs, or ASPEN integration. The optional ASI:One LLM layer only phrases the deterministic results conversationally — it does not select equipment or generate data. - Prices, lead times, and suppliers are mock placeholders. ## Possible future work - Split the single manager agent into specialist sub-agents (reactor, separation, drying, supplier matching) communicating via uAgents messages. - Add human-in-the-loop "top-2" selection per unit operation. - Replace the mock databases with real, curated equipment/supplier data. See `docs/architecture.md` for the intended agent architecture. ## Detected evidence (automated analysis) Indexed codebase: 15 recognized source files, 104 KB. - Python (language) — detected in the code ## Codebase structure (from repository index) ### Files (21 of 21) ``` .gitignore ~$ENT_README.md AGENT_README.md agent.py data/equipment_database.json data/sample_input.json data/supplier_database.json DEMO.md docs/architecture.md docs/project_guide.md main.py outputs/sample_output.json outputs/sample_report.md README.md requirements.txt src/engineering_reviewer.py src/equipment_selector.py src/process_basis.py src/rfq_generator.py src/supplier_matcher.py src/unit_decomposer.py ``` ### Dependencies - requirements.txt: uagents@>=0.25 ### Recent commits (newest first) - Update README.md - Add demo script - Update README to document the ASI:One agent and current state - Add Agentverse agent profile README - Harden ASI:One LLM layer against agentic garbage output - Add optional ASI:One LLM layer and conversational routing to the agent - Add ASI:One-compatible manager uAgent wrapping the workflow - Initial commit: local MVP process equipment selection assistant ## Key source files (fetched from GitHub, selected and truncated for size) ### ~$ENT_README.md ```markdown Jinghan Gao Jinghan Gao��u@z��u� � �F^���?ؖ�a���dW��P���u��M ``` ### DEMO.md ```markdown # Demo Script (~90 seconds) A tight script for presenting the Process Equipment Selection Assistant on ASI:One. English lines are ready to speak; _italics_ are stage directions. --- ## 1. The pitch (~15s) > "When a chemist has a process that works at lab scale, the first engineering > question is: **what equipment do I need to scale it up, and roughly what will > it cost?** Today that takes an engineer hours of manual screening. Our agent > does a **preliminary pass in seconds** — and you just talk to it on ASI:One." ## 2. Live demo (~50s) _Open your agent in ASI:One. Type:_ **`analyze`** > "I gave it a real case — scaling a lab-scale **hydrothermal synthesis of > MnO₂** for battery cathodes from 40 mL up to a 2-liter pilot batch. It > recognized the reaction, scaled it 50×, broke it into unit operations, and > recommended equipment for each — a **pressure-rated Teflon-lined autoclave** > for the 180 °C reaction, vacuum filtration for the fine particles, a vacuum > oven for drying — with a rough CAPEX range." _Then ask a follow-up to show it reasons, not just dumps:_ **`why the Teflon-lined autoclave?`** > "Notice it explains the *why* — pressure rating, contamination control for > battery-grade purity. And this is the key part: **the equipment, prices, and > suppliers all come from a deterministic engineering workflow** — the language > model only phrases it conversationally, it never makes up data." _Then:_ **`rfq`** > "And it drafts a ready-to-send supplier inquiry, stating our operating point > and what the equipment must be rated to cover." ## 3. The close (~20s) > "Under the hood it's a **Fetch.ai uAgent** on Agentverse, ASI:One-compatible > via the chat protocol, wrapping a transparent rule-based workflow. It's > positioned honestly as **preliminary decision-support** — not a certified > design — so an engineer reviews the output before procurement. Next steps are > splitting it into specialist sub-agents and adding human-in-the-loop selection." --- ## Backup / Q&A one-liners - **"Is it making the numbers up?"** — No. A deterministic Python workflow selects equipment and computes CAPEX from a database; the LLM only explains. - **"What if the LLM is down?"** — It falls back to clean rule-based replies, so the demo never breaks. - **"How is it scoped?"** — Mock data, preliminary only, requires engineer review. No process simulation or vessel design. - **"Why Fetch.ai?"** — Modular agent design; the manager can later coordinate specialist sub-agents, all discoverable and chat-accessible via ASI:One. ``` ### requirements.txt ``` # Local workflow (main.py / src/) uses only the Python standard library. # The ASI:One-compatible agent (agent.py) needs uAgents: uagents>=0.25 ``` ### main.py ```python """ main.py Entry point for the Process Equipment Selection Assistant local MVP. """ import argparse import json import os import sys sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "src")) from process_basis import build_process_basis from unit_decomposer import decompose_process from equipment_selector import select_equipment from engineering_reviewer import review_candidates from supplier_matcher import match_suppliers from rfq_generator import generate_rfq # Anchor data/output paths to this file's directory so the workflow works no # matter which directory it is launched from (CLI or agent). PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) EQUIPMENT_DB_PATH = os.path.join(PROJECT_ROOT, "data", "equipment_database.json") SUPPLIER_DB_PATH = os.path.join(PROJECT_ROOT, "data", "supplier_database.json") OUTPUT_DIR = os.path.join(PROJECT_ROOT, "outputs") OUTPUT_JSON_PATH = os.path.join(OUTPUT_DIR, "sample_output.json") OUTPUT_REPORT_PATH = os.path.join(OUTPUT_DIR, "sample_report.md") def load_json(path): """Load JSON from a UTF-8 file.""" with open(path, "r", encoding="utf-8") as handle: return json.load(handle) def run_workflow(input_path): """Run the full local MVP workflow from an input file path.""" process_input = load_json(input_path) return run_workflow_from_dict(process_input) def run_workflow_from_dict(process_input): """Run the full local MVP workflow from an already-parsed input dict. Exposed separately so an agent wrapper can call the workflow in-memory without writing the input to a file first. """ equipment_database = load_json(EQUIPMENT_DB_PATH) supplier_database = load_json(SUPPLIER_DB_PATH) process_basis = build_process_basis(process_input) units = decompose_process(process_input, process_basis) selection_results = select_equipment(units, equipment_database) reviews = review_candidates(selection_results, process_basis) supplier_result = match_suppliers(reviews, supplier_database) rfq = generate_rfq(reviews, supplier_result, process_basis) return { "project_name": process_input.get("project_name"), "process_goal": process_input.get("process_goal"), "process_basis": process_basis, "unit_operations": units, "equipment_review": reviews, "supplier_matching": supplier_result, "rfq_draft": rfq, "disclaimer": ( "Preliminary decision-support output generated with mock data. " "Not a certified engineering design. Requires review by a " "qualified process engineer before any procurement." ), } def save_json_output(result, path): """Write the structured result to a readable JSON file.""" cleaned = _strip_equipment_records(result) with open(path, "w", encoding="utf-8") as handle: json.dump(cleaned, handle, indent=2, ensure_ascii=False) def _strip_equipment_records(result): """Remove bulky carried equipment records from the saved JSON output.""" copied = json.loads(json.dumps(result, ensure_ascii=False)) for review in copied.get("equipment_review", []): for key in ("accepted_candidates", "rejected_candidates"): for candidate in review.get(key, []): candidate.pop("equipment_record", None) recommended = review.get("recommended_equipment") if recommended: recommended.pop("equipment_record", None) return copied def save_markdown_report(result, path): """Write a human-readable Markdown report.""" lines = [] add = lines.append basis = result["process_basis"] product = basis["product_summary"] add("# Preliminary Equipment Selection Report") add("") add("**Project:** {}".format(result.get("project_name", "N/A"))) add("") add("**Goal:** {}".format(result.get("process_goal", "N/A"))) add("") add("> {}".format(result["disclaimer"])) add("") add("## 1. Process Summary") add("") add("- **Target product:** {}".format(product.get("name"))) add("- **Intended application:** {}".format(product.get("intended_application"))) add("- **Scale-up factor:** {}".format(basis.get("scale_up_factor"))) add( "- **Target batch volume:** {} L".format( basis["target_scale_basis"].get("target_batch_liquid_volume_l") ) ) add("") add("**Critical quality attributes:**") add("") for attribute in product.get("critical_quality_attributes", []): add("- {}".format(attribute)) add("") add("## 2. Key Assumptions and Missing Data") add("") for note in basis.get("missing_or_uncertain_data", []): add("- {}".format(note)) add("") add("## 3. Major Unit Operations") add("") add("| ID | Unit Operation | Type | Temp (C) | Pressure (bar) | Volume (L) |") add("| --- | --- | --- | --- | --- | --- |") for unit in result["unit_operations"]: add( "| {} | {} | {} | {} | {} | {} |".format( unit["unit_id"], unit["unit_name"], unit["unit_type"], unit["temperature_c"], unit["pressure_bar"], unit["required_volume_l"], ) ) add("") add("## 4. Recommended Equipment") add("") add("| Unit | Recommended Equipment | Type | Contamination Control |") add("| --- | --- | --- | --- |") for review in result["equipment_review"]: rec = review["recommended_equipment"] if rec: add( "| {} | {} | {} | {} |".format( review["unit_name"], rec["equipment_name"], rec["equipment_type"], rec.get("contamination_control_level", "n/a"), ) ) else: add("| {} | _No suitable equipment accepted_ | - | - |".format(review["unit_na [truncated — 5121 more characters] ``` ### agent.py ```python """ agent.py ASI:One-compatible manager uAgent for the Process Equipment Selection Assistant. This agent wraps the existing local workflow (see main.py / src/) and exposes it over the uAgents chat protocol, so it can be reached from ASI:One. Design for the hackathon MVP: - The agent's "brain" is the deterministic local workflow (mock data only). - On any chat message it acts as a coordinator ("manager"): it recognizes the process, runs the unit decomposition + equipment screening + supplier matching (the "specialist" steps), and replies with a readable summary. - If the user sends a JSON process brief, it is used as the input. Otherwise the bundled MnO2 sample is used, so a judge can just say "hi" and get a full demo. - Reply "rfq" to get the RFQ-style supplier inquiry draft (human-in-the-loop). Run locally: python agent.py """ import json import os import sys from datetime import datetime from uuid import uuid4 # Make src/ importable and reuse the existing workflow. sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "src")) from main import run_workflow_from_dict, load_json, EQUIPMENT_DB_PATH # noqa: E402 from uagents import Agent, Context, Protocol # noqa: E402 from uagents_core.contrib.protocols.chat import ( # noqa: E402 ChatAcknowledgement, ChatMessage, EndSessionContent, TextContent, chat_protocol_spec, ) PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) SAMPLE_INPUT_PATH = os.path.join(PROJECT_ROOT, "data", "sample_input.json") # Optional ASI:One LLM layer. If an API key is set, the agent uses the LLM to # converse naturally on top of the deterministic analysis. If not, it falls # back to the rule-based replies, so the demo never breaks. ASI_ONE_API_KEY = os.environ.get("ASI_ONE_API_KEY", "").strip() _llm_client = None if ASI_ONE_API_KEY: from openai import OpenAI _llm_client = OpenAI(base_url="https://api.asi1.ai/v1", api_key=ASI_ONE_API_KEY) def llm_reply(grounding, user_text): """Ask the ASI:One LLM to answer naturally, grounded in the analysis. Returns None on any failure so the caller can fall back to a fixed reply. """ if not _llm_client: return None system_prompt = ( "You are a process equipment selection assistant for early-stage " "chemical process scale-up. A deterministic tool has ALREADY produced " "the analysis below. Answer the user using ONLY this analysis - never " "invent equipment, prices, suppliers, or numbers.\n\n" "STRICT OUTPUT RULES:\n" "- Reply in ENGLISH only.\n" "- Reply with a direct, plain-text answer for a human to read.\n" "- Do NOT emit tool calls, function calls, or any XML/JSON/tags such " "as or . Just write the answer text.\n" "- Be concise, friendly, and conversational.\n" "- For a greeting, give a one or two sentence intro and offer to " "analyze or show the demo - do NOT dump the whole report.\n" "- When giving recommendations, remind the user this is preliminary, " "mock-data decision support.\n\n" "=== ANALYSIS (ground truth) ===\n{}".format(grounding) ) try: response = _llm_client.chat.completions.create( model="asi1", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_text or "hello"}, ], temperature=0.3, max_tokens=1024, ) reply = str(response.choices[0].message.content).strip() # Guard against the agentic model emitting tool-call/markup garbage. # If it does, signal failure so the caller falls back to a clean reply. markers = (" target {} L (factor {}).".format( basis["lab_scale_basis"].get("batch_liquid_volume_ml"), basis["target_scale_basis"].get("target_batch_liquid_volume_l"), basis.get("scale_up_factor"), ) ) lines.append("") lines.append("**Recommended equipment per unit:**") for review in result["equipment_review"]: rec = review["recommended_equipment"] if rec: [truncated — 5762 more characters] ``` ### src/equipment_selector.py ```python """ equipment_selector.py Find candidate equipment for each unit operation from the mock equipment database. This step is intentionally a LOOSE, first-pass search. It only checks the two fields that decide whether a piece of equipment is even the right kind of tool for the job: 1. process step -- the unit's process_step must be listed in the equipment's "suitable_process_steps" 2. phase -- the unit's phase must overlap the equipment's "suitable_phase" The stricter engineering checks (temperature, pressure, volume, contamination, materials) are done later in engineering_reviewer.py, so that the "candidate search" and the "accept / reject decision" stay clearly separated. """ def select_equipment(units, equipment_database): """Return candidate equipment for every unit operation. Args: units: list of unit-operation dicts from unit_decomposer. equipment_database: parsed data/equipment_database.json. Returns: list of dicts, one per unit, each containing the unit id/name and a list of candidate equipment entries. """ all_equipment = equipment_database.get("equipment", []) results = [] for unit in units: candidates = [] for equipment in all_equipment: if _is_candidate(unit, equipment): candidates.append(_describe_candidate(unit, equipment)) results.append( { "unit_id": unit["unit_id"], "unit_name": unit["unit_name"], "unit_type": unit["unit_type"], # Carry the unit's operating requirements forward so the # engineering reviewer can apply its strict checks without # re-reading the unit list. "process_step": unit["process_step"], "phase": unit["phase"], "temperature_c": unit["temperature_c"], "pressure_bar": unit["pressure_bar"], "required_volume_l": unit["required_volume_l"], "candidates": candidates, } ) return results def _is_candidate(unit, equipment): """Return True if the equipment is the right kind of tool for the unit.""" step_matches = unit["process_step"] in equipment.get( "suitable_process_steps", [] ) phase_matches = unit["phase"] in equipment.get("suitable_phase", []) return step_matches and phase_matches def _describe_candidate(unit, equipment): """Build a candidate record that records WHY it was matched. The matched_fields / initial_fit_notes make the first-pass selection transparent and easy to read in the final report. """ return { "equipment_id": equipment["equipment_id"], "equipment_name": equipment["name"], "equipment_type": equipment["equipment_type"], "matched_fields": { "process_step": unit["process_step"], "phase": unit["phase"], }, "initial_fit_notes": [ "Process step '{}' is listed in this equipment's suitable steps.".format( unit["process_step"] ), "Phase '{}' is supported by this equipment.".format(unit["phase"]), ], # Carry the full equipment record forward so the engineering reviewer # can apply the strict temperature / pressure / volume / material # checks without re-reading the database. "equipment_record": equipment, } ``` ### src/supplier_matcher.py ```python """ supplier_matcher.py Match each recommended piece of equipment to mock suppliers and build a rough CAPEX estimate. Matching is by equipment_type: a supplier is a match if the recommended equipment's type is listed in the supplier's "equipment_types_supported". All prices and lead times are illustrative mock data from data/supplier_database.json and must not be used for real procurement. """ def match_suppliers(reviews, supplier_database): """Match recommended equipment to suppliers and total a rough CAPEX range. Args: reviews: output of engineering_reviewer.review_candidates. supplier_database: parsed data/supplier_database.json. Returns: dict with per-unit supplier matches and an overall CAPEX summary. """ all_suppliers = supplier_database.get("suppliers", []) matches = [] capex_low_total = 0 capex_high_total = 0 have_any_price = False # Track equipment already counted, so one physical unit shared by several # process steps (e.g. the reactor used for both reaction and cooling) is # not double-counted in the total CAPEX. counted_equipment_ids = set() for review in reviews: recommended = review["recommended_equipment"] if not recommended: matches.append( { "unit_id": review["unit_id"], "unit_name": review["unit_name"], "recommended_equipment": None, "suppliers": [], "unit_capex_range_usd": None, } ) continue equipment_type = recommended["equipment_type"] suppliers = _find_suppliers(equipment_type, all_suppliers) # Use the lowest min price and highest max price across the matched # suppliers as this unit's rough CAPEX range. unit_range = _unit_capex_range(suppliers) # Only add to the total the first time we see a given equipment id, so # equipment shared across steps is counted once. if unit_range and recommended["equipment_id"] not in counted_equipment_ids: counted_equipment_ids.add(recommended["equipment_id"]) capex_low_total += unit_range[0] capex_high_total += unit_range[1] have_any_price = True matches.append( { "unit_id": review["unit_id"], "unit_name": review["unit_name"], "recommended_equipment": { "equipment_id": recommended["equipment_id"], "equipment_name": recommended["equipment_name"], "equipment_type": equipment_type, }, "suppliers": suppliers, "unit_capex_range_usd": unit_range, } ) capex_summary = { "rough_total_capex_usd": [capex_low_total, capex_high_total] if have_any_price else None, "note": ( "Illustrative mock CAPEX only. Each distinct recommended equipment " "item is counted once (equipment shared across steps is not " "double-counted). Totals use the lowest and highest mock supplier " "prices. Not a quote." ), } return {"unit_supplier_matches": matches, "capex_summary": capex_summary} def _find_suppliers(equipment_type, all_suppliers): """Return supplier records that support the given equipment type.""" found = [] for supplier in all_suppliers: if equipment_type in supplier.get("equipment_types_supported", []): found.append( { "supplier_id": supplier["supplier_id"], "supplier_name": supplier["supplier_name"], "example_equipment": supplier.get("example_equipment"), "location": supplier.get("location"), "estimated_price_range_usd": supplier.get( "estimated_price_range_usd" ), "typical_lead_time_weeks": supplier.get( "typical_lead_time_weeks" ), "strengths": supplier.get("strengths", []), "limitations": supplier.get("limitations", []), "rfq_notes": supplier.get("rfq_notes", []), } ) return found def _unit_capex_range(suppliers): """Combine matched supplier prices into one [low, high] range for a unit.""" lows = [] highs = [] for supplier in suppliers: price = supplier.get("estimated_price_range_usd") if price and len(price) == 2: lows.append(price[0]) highs.append(price[1]) if not lows: return None return [min(lows), max(highs)] ``` ### src/process_basis.py ```python """ process_basis.py Build a preliminary "process basis" from the lab-scale process brief. This module does NOT simulate anything. It only reads the structured input JSON and pulls out / lightly infers the engineering basics that the rest of the workflow needs (product, scale, operating conditions, priorities, and the data we are missing). """ def build_process_basis(process_input): """Return a structured process basis dictionary. Args: process_input: the parsed contents of data/sample_input.json Returns: dict describing the preliminary process basis. """ lab = process_input.get("lab_scale_process", {}) target = process_input.get("target_scale", {}) product = process_input.get("target_product", {}) # --- Lab-scale basis ------------------------------------------------- lab_volume_ml = lab.get("batch_liquid_volume_ml") hydrothermal = lab.get("hydrothermal_step", {}) pre_step = lab.get("pre_hydrothermal_step", {}) lab_basis = { "batch_liquid_volume_ml": lab_volume_ml, "synthesis_route": lab.get("synthesis_route"), "pre_reaction_temperature_c_range": pre_step.get("temperature_c_range"), "hydrothermal_temperature_c": hydrothermal.get("temperature_c"), "hydrothermal_duration_hr": hydrothermal.get("duration_hr"), "hydrothermal_pressure_condition": hydrothermal.get("pressure_condition"), } # --- Target-scale basis --------------------------------------------- target_volume_l = target.get("target_batch_liquid_volume_l") target_basis = { "mode": target.get("mode"), "target_batch_liquid_volume_l": target_volume_l, "batches_per_day": target.get("batches_per_day"), "operation_mode": target.get("operation_mode"), } # --- Scale-up factor ------------------------------------------------- # Prefer the value given in the input. If absent, compute it from volumes. scale_up_factor = target.get("scale_up_factor_from_lab_volume") if scale_up_factor is None and lab_volume_ml and target_volume_l: scale_up_factor = round((target_volume_l * 1000.0) / lab_volume_ml, 1) # --- Operating condition summary ------------------------------------ operating_conditions = { "pre_reaction_temperature_c_range": pre_step.get("temperature_c_range"), "reaction_temperature_c": hydrothermal.get("temperature_c"), "reaction_pressure": hydrothermal.get("pressure_condition"), "drying_temperature_c": _find_drying_temperature(lab), } # --- Missing / uncertain data --------------------------------------- # These are simple, transparent checks so a reader can see WHY each item # is flagged. This is intentionally rule-based, not "smart". missing_data = _find_missing_data(process_input, scale_up_factor) return { "product_summary": { "name": product.get("name"), "intended_application": product.get("intended_application"), "critical_quality_attributes": product.get( "critical_quality_attributes", [] ), }, "lab_scale_basis": lab_basis, "target_scale_basis": target_basis, "scale_up_factor": scale_up_factor, "operating_condition_summary": operating_conditions, "quality_and_contamination_priorities": process_input.get( "equipment_selection_priorities", [] ), "preferred_materials": process_input.get("preferred_materials", []), "avoid_materials": process_input.get("avoid_materials", []), "known_risks": process_input.get("known_process_concerns", []), "missing_or_uncertain_data": missing_data, } def _find_drying_temperature(lab): """Look through post-reaction steps for a drying temperature.""" for step in lab.get("post_reaction_steps", []): if step.get("operation") == "drying": return step.get("temperature_c") return None def _find_missing_data(process_input, scale_up_factor): """Return a list of plain-language notes about gaps in the input.""" notes = [] target = process_input.get("target_scale", {}) if target.get("target_batch_liquid_volume_l") is None: notes.append("Target batch volume is not specified.") if scale_up_factor is None: notes.append("Scale-up factor could not be determined.") hydrothermal = process_input.get("lab_scale_process", {}).get( "hydrothermal_step", {} ) # Autogenous pressure is given as a condition, not a number, so the real # design pressure still needs to be confirmed. if not isinstance(hydrothermal.get("pressure_bar"), (int, float)): notes.append( "Hydrothermal operating pressure is autogenous and given as a " "condition, not a confirmed numeric value. Design pressure at " "the hold temperature must be verified." ) notes.append( "Particle size, yield, and morphology after scale-up are not predicted " "by this tool and require experimental confirmation." ) return notes ``` ### src/rfq_generator.py ```python """ rfq_generator.py Generate an RFQ-style (Request For Quote) draft for the recommended equipment. This module only builds text. It never sends emails or contacts suppliers. The draft is meant to be pasted into the final report so a human can review and send it themselves. """ def generate_rfq(reviews, supplier_result, process_basis): """Build an RFQ draft covering every recommended equipment item. Args: reviews: output of engineering_reviewer.review_candidates. supplier_result: output of supplier_matcher.match_suppliers. process_basis: output of process_basis.build_process_basis. Returns: dict with a short intro, one RFQ item per recommended equipment, and a closing note. """ product = process_basis["product_summary"] target = process_basis["target_scale_basis"] preferred_materials = process_basis.get("preferred_materials", []) avoid_materials = process_basis.get("avoid_materials", []) # Index supplier matches by unit id so we can attach supplier names. suppliers_by_unit = { m["unit_id"]: m for m in supplier_result["unit_supplier_matches"] } intro = ( "Request for preliminary quotation for pilot-scale equipment supporting " "the production of {product} for {application}. Target batch size is " "{volume} L in {mode} batch operation. This is an early-stage inquiry " "for budgetary purposes only.".format( product=product.get("name", "the target product"), application=product.get("intended_application", "the intended application"), volume=target.get("target_batch_liquid_volume_l", "TBD"), mode=target.get("mode", "pilot"), ) ) # Build one RFQ item per distinct recommended equipment item, so a shared # unit is not requested twice. items = [] seen_equipment_ids = set() for review in reviews: recommended = review["recommended_equipment"] if not recommended: continue equipment_id = recommended["equipment_id"] if equipment_id in seen_equipment_ids: continue seen_equipment_ids.add(equipment_id) items.append( _build_item( review, recommended, suppliers_by_unit.get(review["unit_id"], {}), preferred_materials, avoid_materials, ) ) closing = ( "Please provide budgetary pricing, lead time, product-contact material " "options, and relevant certifications. All values will be treated as " "preliminary and subject to engineering review." ) return {"intro": intro, "items": items, "closing": closing} def _build_item(review, recommended, supplier_match, preferred_materials, avoid_materials): """Build a single RFQ item for one recommended equipment piece.""" equipment = recommended["equipment_record"] # State the ACTUAL operating point first, then explain that the equipment's # rated range must cover that point (with margin). This reads the way an # engineer would actually write an RFQ. operating_conditions = [ "Operating temperature is {} C, so the equipment must be rated to " "cover this point with margin (recommended example is rated " "{} C).".format( review.get("operating_temperature_c", "TBD"), _range_text(equipment.get("temperature_c_range")), ), "Operating pressure is about {} bar, so the equipment must be rated " "to cover this with adequate relief (recommended example is rated " "{} bar).".format( review.get("operating_pressure_bar", "TBD"), _range_text(equipment.get("pressure_bar_range")), ), "Batch volume is {} L, so the working volume must accommodate this " "(recommended example handles {} L).".format( review.get("required_volume_l", "TBD"), _range_text(equipment.get("working_volume_l_range")), ), ] # Standard questions every supplier should answer, plus any RFQ notes the # mock supplier database already provided. questions = [ "What product-contact materials are available, and can they avoid: " + ", ".join(avoid_materials) + "?", "Can the equipment meet the operating conditions listed above with an " "adequate safety margin?", "What contamination control, cleaning, and validation options are " "supported?", "What is the budgetary price and typical lead time?", ] for supplier in supplier_match.get("suppliers", []): for note in supplier.get("rfq_notes", []): questions.append("({}) {}".format(supplier["supplier_name"], note)) supplier_names = [ s["supplier_name"] for s in supplier_match.get("suppliers", []) ] return { "for_unit": review["unit_name"], "equipment_type": recommended["equipment_type"], "equipment_example": recommended["equipment_name"], "operating_conditions": operating_conditions, "preferred_contact_materials": preferred_materials, "materials_to_avoid": avoid_materials, "questions_for_supplier": questions, "candidate_suppliers": supplier_names, } def _range_text(range_pair): """Format a [low, high] pair, or note that it is unspecified.""" if range_pair and len(range_pair) == 2: return "{} to {}".format(range_pair[0], range_pair[1]) return "to be specified" ``` [6 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]