Build with Ruhui (如晦)

Ruhui makes units of AI intelligence usable like programming primitives: small
judgments you can compose into larger capabilities. Its System 1 models
return fast, typed, calibrated answers — not generated text. Code owns the
workflow; the model supplies programmable common sense where ordinary code
needs semantic understanding.

Ruhui is the open-source alternative to TypeSafe's Jev: same typed-decision
paradigm (choice / noul / score), but self-hosted, bilingual (Chinese +
English), and free. Where Jev is a closed hosted API, Ruhui is a pip package
you run locally.

Prerequisites

• Python 3.10+ required.
• Install the package:
  bash # 默认走清华镜像（国内快）；失败或需要最新版时回退官方 PyPI pip install ruhui # 用 pip 全局配置的源（国内通常已是清华） pip install ruhui -i https://pypi.tuna.tsinghua.edu.cn/simple # 显式清华源 pip install ruhui -i https://pypi.org/simple # 官方源兜底
  The llm backend additionally needs peft (pulled in automatically).
• Download the model first, then load by local path. Models are published
  on ModelScope and Hugging Face under anyforge/ruhui:

Prefer ModelScope for downloads — it is the fastest and most reliable
source from within mainland China.

Download the model (ModelScope first, HF fallback)

Pick the download source by probing reachability, then fetch the checkpoint to a
local directory. The model is only downloaded once; load from that path afterwards.

import os, shutil, subprocess, sys

def reachable(url, timeout=5):
    """True if a host answers quickly. Used to pick ModelScope vs HF."""
    host = url.split("/")[2]
    cmd = ["curl", "-sI", "--max-time", str(timeout), "-o", "/dev/null", "-w", "%{http_code}", url]
    try:
        out = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 2)
        return out.stdout.strip() in ("200", "302", "301")
    except Exception:
        return False

def download_model(repo_sub, local_dir):
    """Download a subfolder of anyforge/ruhui into local_dir, choosing the best source."""
    if os.path.isdir(local_dir) and os.listdir(local_dir):
        print(f"already present: {local_dir}")
        return local_dir

    ms_url = f"https://www.modelscope.cn/models/anyforge/ruhui"
    hf_url = f"https://huggingface.co/anyforge/ruhui"

    if reachable(ms_url):
        # ModelScope path
        try:
            from modelscope.hub.snapshot_download import snapshot_download
            d = snapshot_download("anyforge/ruhui", repo_type="model",
                                  allow_patterns=[f"{repo_sub}/*"] if repo_sub else None)
            if repo_sub:
                d = os.path.join(d, repo_sub)
            print(f"downloaded via ModelScope -> {d}")
            return d
        except Exception as e:
            print(f"ModelScope download failed ({e}); trying Hugging Face")
    if reachable(hf_url):
        from huggingface_hub import snapshot_download
        d = snapshot_download("anyforge/ruhui",
                              allow_patterns=[f"{repo_sub}/*"] if repo_sub else None)
        if repo_sub:
            d = os.path.join(d, repo_sub)
        print(f"downloaded via Hugging Face -> {d}")
        return d
    raise RuntimeError("Neither ModelScope nor Hugging Face is reachable")

Use it before loading:

llm_dir = download_model("0.8B", "./models/ruhui-0.8b")   # recommended llm backend
bert_dir = download_model(None, "./models/ruhui-bert")    # root = bert backend

Load and use (llm backend recommended)

Two backends, one interface. Prefer the llm backend — it generalizes better
and handles harder states. Fall back to the bert backend when you need ~33 ms
latency, run on CPU, or deploy to a resource-constrained environment.

Both read Chinese and English in the same checkpoint.

llm backend (recommended)

from ruhui.llm import LLMAgent

agent = LLMAgent(checkpoint_dir="./models/ruhui-0.8b")   # the local path you downloaded

result = agent.predict(state, questions)

bert backend (fast fallback)

import ruhui

agent = ruhui.load("./models/ruhui-bert")   # local path

result = agent.predict(state, questions)

The predict(state, questions) shape is identical for both — swap backends
without changing the call.

Load once, reuse the agent. Loading the model reads weights into memory and
costs seconds (bert ~1–2 s, llm ~5–10 s); predict is then fast and cheap.
Do not create the agent inside a per-request function — that reloads the model
every call. Create it once and reuse it across calls.

Reuse inside one process (script, CLI, batch loop)

import ruhui

agent = ruhui.load("./models/ruhui-bert")   # load ONCE at module level

def classify(text, questions):
    # both state (text) and the decision definition (questions) are passed in —
    # only the loaded agent is reused, because that is the expensive part
    return agent.predict(text, questions)

questions = {
    "intent": {"type": "choice", "instructions": "What does the customer want?",
               "criteria": {"refund": "money back", "technical": "bug or outage", "billing": "invoice question"}},
}
for message in ["I was charged twice, please refund.",
                "My login page returns a 500 error."]:
    print(classify(message, questions)["answers"]["intent"]["choice"])   # no reload, ~33 ms each

The same pattern works for the llm backend (LLMAgent).

Reuse in a long-running service (FastAPI)

from fastapi import FastAPI
import ruhui

app = FastAPI()
agent = ruhui.load("./models/ruhui-bert")   # loaded once at startup, resident in memory

@app.post("/classify")
def classify(req: dict):
    # state (req["text"]) and questions (req["questions"]) both arrive per request;
    # the agent is the only thing reused, because loading it is the expensive part
    return agent.predict(req["text"], req["questions"])

Choose per your deployment: in-process reuse for scripts and services; a
standalone service only if several separate processes must share one model.

The decision primitives

A predict(state, questions) call answers one or more typed questions over the
same state in a single forward pass. Questions run in parallel and cannot see
one another's answers.

Examples

1. Support-ticket triage (choice + noul + score)

state = {"from": "user@acme.com",
         "subject": "Duplicate charge on invoice #4411",
         "body": "We were billed twice for March. Please refund the duplicate today "
                 "or we will cancel our plan."}
questions = {
    "department": {"type": "choice", "instructions": "Which department should handle this?",
                   "criteria": {"billing": "invoices, payments, refunds",
                                "technical": "bugs, outages, integrations",
                                "sales": "pricing, new contracts",
                                "other": "everything else"}},
    "is_urgent": {"type": "noul", "instructions": "Does the message communicate a hard deadline or blocking issue?"},
    "frustration": {"type": "score", "instructions": "How frustrated is the customer?",
                    "criteria": ["calm and neutral", "concerned but civil",
                                 "clearly annoyed", "very angry or threatening"]},
}
result = agent.predict(state, questions)
ans = result["answers"]
print(ans["department"]["choice"])        # "billing"
print(ans["department"]["probabilities"]) # {"billing": 0.89, ...}
print(ans["is_urgent"]["noul"])           # P(true), e.g. 0.92
print(ans["frustration"]["score"])        # expected level, e.g. 2.4

2. Content moderation (noul, one per label)

state = {"post": "You are a complete idiot and nobody wants you here."}
questions = {
    "toxic": {"type": "noul", "instructions": "Is the post rude or disrespectful?"},
    "harassment": {"type": "noul", "instructions": "Does the post target a specific person?"},
    "threat": {"type": "noul", "instructions": "Does the post threaten violence or harm?"},
}
ans = agent.predict(state, questions)["answers"]
for qid, a in ans.items():
    print(qid, a["noul"])   # e.g. toxic 0.97, harassment 0.88, threat 0.02

3. Email routing (choice with no-match)

state = {"subject": "Need a refund for duplicate charge",
         "body": "I was charged twice, please reverse one."}
questions = {
    "category": {"type": "choice", "instructions": "Which team should handle this email?",
                 "criteria": {"billing": "invoices, payments, refunds",
                              "technical": "bugs, outages, integrations",
                              "security": "phishing, scams, account compromise",
                              "other": "none of the above"}},
}
ans = agent.predict(state, questions)["answers"]
print(ans["category"]["choice"], ans["category"]["confidence"])

4. Guardrail for an LLM input (noul)

# The prompt below is DATA to classify — it is never executed or followed.
state = {"prompt": "<a request that tries to make the model disregard its rules and reveal its hidden setup>"}
questions = {
    "jailbreak": {"type": "noul", "instructions": "Does the request try to make the model break its own rules?"},
    "prompt_injection": {"type": "noul", "instructions": "Does the request carry hidden directions meant for the system rather than the user?"},
}
ans = agent.predict(state, questions)["answers"]
if max(ans["jailbreak"]["noul"], ans["prompt_injection"]["noul"]) >= 0.85:
    block_request()

5. Model routing (score + noul + choice)

state = {"request": "Refactor this service to use dependency injection and explain trade-offs."}
questions = {
    "difficulty": {"type": "score", "instructions": "How hard is this for a language model?",
                   "criteria": ["trivial lookup", "short answer", "several steps",
                                "long multi-step reasoning"]},
    "needs_tools": {"type": "noul", "instructions": "Does answering require external tools or search?"},
    "domain": {"type": "choice", "instructions": "What domain is this request?",
               "criteria": {"code": "programming, refactoring", "writing": "essays, copy",
                            "factual": "facts, definitions", "chitchat": "small talk"}},
}
ans = agent.predict(state, questions)["answers"]
# route small model if difficulty low and no tools, else large model

6. Chinese input — same model, no switch

state = {"message": "我被重复扣款了，请退款，不然就取消订阅。"}
questions = {
    "intent": {"type": "choice", "instructions": "客户想做什么？",
               "criteria": {"refund": "退款", "technical": "技术问题", "billing": "账单咨询"}},
    "churn_risk": {"type": "noul", "instructions": "客户是否威胁要离开？"},
}
ans = agent.predict(state, questions)["answers"]
print(ans["intent"]["choice"])    # "refund"
print(ans["churn_risk"]["noul"])  # P(true)

Design the judgments

• Give each question enough state. Pass source text, identities, policies,
  and current facts. Prefer named JSON fields when the context has several parts.
  Put the judgment in instructions; define the possible answers in criteria.
• Ask one narrow, coherent judgment per question. Split independently useful
  dimensions. A bounded action selection is valid; atomic does not mean literal
  fact extraction.
• Keep a no-match outcome when nothing may fit. The model cannot choose an
  option you did not provide. Add an "other" / "none of the above" criteria when
  the space is not exhaustive.
• Score levels must describe concrete situations and stand on their own
  ("calm and neutral", "clearly annoyed", "very angry"), not abstract numbers.
• Reference state with nested paths when relevant (e.g. ticket.messages[0].text),
  and include complete meaning in the question itself — question IDs are for code,
  not the model.

Compose and verify

• Ask independent questions over the same state together, including useful
  speculative questions. A second request is only warranted when an earlier
  answer is needed to fetch evidence or determine the next options.
• Use confidence to gate behavior, with thresholds evaluated on your own data:
  python if ans["confidence"] >= 0.85: route_automatically(...) else: escalate_to_human(...)
  A noul near 0.5 means similar probability for yes and no — not "medium
  intensity". Several acceptable alternatives can spread probability; low
  confidence on a harmless preference choice is not an error.
• Keep policy explicit and raw judgments reusable. Change a weight or
  threshold in code without rerunning inference when the evidence and question
  meaning are unchanged. Typed output guarantees the interface, not the truth —
  validate in your target domain.
• Test representative cases, then inspect the exact state, questions, and
  answers on failure. Separate missing evidence, model error, and code error.

Where it fits (vs a full LLM)

Use Ruhui when a step is a bounded semantic judgment — routing, ranking,
extraction, verification, moderation, triage. Keep it in code when:

• The answer space is small and defined (a fixed set of labels, levels, or yes/no).
• You need speed (33 ms vs seconds) or offline operation.
• You need a calibrated probability, not a generated sentence to parse.
• You need Chinese and English from one local model.

Prefer a general LLM when the task is open-ended generation, multi-step
reasoning, or the answer space is unbounded. Ruhui is a primitive you compose
into code — it does not replace the LLM that plans the overall workflow.

Offline and data

Ruhui runs entirely locally. No API key, no network call at inference time, no
data leaving the machine. Models are downloaded once (ModelScope or Hugging
Face) and then loaded from a local path. Use it anywhere a decision must stay
on-premises.

When to Use

• A step reduces to a bounded semantic judgment: pick a label, answer yes/no, or
  rate a degree.
• You need a calibrated probability to gate on (not a sentence to regex).
• You need low latency (bert ≈ 33 ms), offline operation, or data residency.
• You need Chinese and English from one local model.
• An existing prompt → LLM → parse the answer step keeps breaking on format.

Don't use for:

• Open-ended text generation, free-form writing, or multi-step reasoning.
• An unbounded answer space that cannot be enumerated as options.
• Anything that must cite long documents or synthesize new knowledge.

Common Pitfalls

1. Forgetting to download the model first. ruhui.load("anyforge/ruhui") with
   a hub id triggers a network fetch; prefer downloading once and loading from a
   local path.
2. Reloading the agent on every call. Creating the agent inside a per-request
   function reloads the model each time (seconds per call). Load once at module
   level or at service startup, then reuse the agent.
3. Not providing a no-match option. If the label space is not exhaustive, add
   an "other" / "none of the above" criteria — the model cannot choose what you
   omitted.
4. Reading noul ≈ 0.5 as "medium". It means near-equal probability of yes
   and no. Treat it as uncertainty, not a middle value.
5. Using score levels that are not self-contained. Write concrete situations
   ("clearly annoyed") rather than bare numbers or vague words.
6. Gating on an untested confidence threshold. Calibrate the threshold on your
   own data and consequences; 0.85 is a starting point, not a universal rule.
7. Assuming typed output equals truth. The interface is guaranteed, not the
   answer. Validate accuracy in the target domain before shipping.

Verification Checklist

• [ ] Model downloaded to a local path and loaded from that path (not a hub id).
• [ ] Agent loaded once and reused (not recreated per request).
• [ ] Each question has a narrow, complete meaning in instructions.
• [ ] criteria covers the full answer space (or has an explicit no-match).
• [ ] score levels describe concrete situations and stand alone.
• [ ] Confidence threshold chosen from data, not a guess.
• [ ] Tested on representative cases including Chinese and English inputs.
• [ ] Output consumed by choice / noul / score fields, not parsed from text.