DemonGatanjieuandClaude Fable 5.2
375b347eb6
A run's, a job sheet's and a blueprint's progress are shown by Anomalous itself with real numbers; the AI's own bar could be stale, and the divider was decoration. Both are refused now and gone from the guide and the page, with the "passed" greying that only served the bar. A panel kept with such a block is still added to; the page leaves a block it no longer draws out. Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>
485 lines
31 KiB
Python
485 lines
31 KiB
Python
"""MCP tools that shape the AI workbench's record (ai_tasks.py): begin_task names a piece of
|
||
work, report leaves a note for the user (with pictures made outside Anomalous's tools),
|
||
show_panel / update_panel draw a panel from Anomalous's blocks (ai_panels.py), get_answers
|
||
reads what the user picked or typed on it, log_outside keeps what the AI did with other tools
|
||
(ai_outside.py)."""
|
||
|
||
from . import ai_answer_wait, ai_decisions, ai_marks, ai_message_refs, ai_outside, ai_panels, ai_repaint, ai_run_control, ai_run_tracking, ai_task_links, ai_task_runs, ai_tasks, ai_toolbox, ai_privacy, ai_skills, mcp_routes, reference_blocks
|
||
from .mcp_actions import downloads as model_downloads
|
||
from .mcp_tools import MAX_LIMIT, ToolError, _schema
|
||
|
||
PACK_TYPES = 150 # node types listed per picked pack
|
||
ASK_WAIT_MIN = 30 # ask_user waits this long even when told not to: the user, reading the work, often answers at once
|
||
ASK_NEXT = ("Before you end your turn, ask_user what to do next or whether to go ahead: a note or a panel alone leaves "
|
||
"the user waiting. Only a run, a download or a scan going on lets you end it without asking.")
|
||
|
||
|
||
def begin_task(ctx, title, new_task=False):
|
||
title = str(title or "").strip()
|
||
if not title:
|
||
raise ToolError("title is empty: a few words on what the user asked for.")
|
||
task = ai_tasks.begin_task(
|
||
ctx.get("client") or "AI", title, new_task=bool(new_task),
|
||
session=ctx.get("session") or "", task_id=ctx.get("task"),
|
||
)
|
||
answer = {"task": task["id"], "title": task["title"], "note": (
|
||
"Continue this task; do not begin another for its steps. Every AMB call is a step the user sees, opens and "
|
||
"undoes in Anomalous's AI workbench. Ask them with ask_user (they answer there, not in your chat); report "
|
||
"explains a choice or hands in pictures, done=true at the end. Run with run_workflow. In the task's own tab, "
|
||
"do not switch tabs yourself. "
|
||
"Everything is in AMB's tools (models, pictures, node types, the canvas, runs and their previews): do not "
|
||
"read ComfyUI's folders or source, run preprocessors or call its routes yourself. Change pictures only by "
|
||
"running workflows, never by editing files with code. If you use another tool anyway, log_outside right "
|
||
"after. The directory below lists the tools by what you want to do.")}
|
||
answer["next"] = ("get_situation shows where things stand; a daily run is one call: run_workflow(workflow: a saved "
|
||
"one, with the prompt, seed or size to change).")
|
||
answer["blocks"] = _blocks(ctx)
|
||
if ai_privacy.is_private(ai_tasks.get_task(task["id"])):
|
||
answer["privacy"] = ai_privacy.NOTE # the user writes the prompts and looks at the pictures: no skill for those
|
||
elif ai_skills.directory():
|
||
answer["skills"] = ai_skills.directory()
|
||
box = ai_toolbox.effective(ai_tasks.get_task(task["id"]))
|
||
if box:
|
||
answer["toolbox"] = ai_toolbox.words(box)
|
||
if _routes(ctx, task["id"]):
|
||
answer["next"] = mcp_routes.TOOLBOX_NOTE
|
||
goes_on = ai_task_links.carried(task)
|
||
if goes_on:
|
||
answer["continues"] = goes_on # the task this one goes on from, and what was decided there
|
||
decided = ai_decisions.lines(ai_tasks.get_task(task["id"]))
|
||
if decided:
|
||
answer["decisions"] = decided # what was settled so far, the user's own choices in Anomalous too
|
||
if task.get("about"):
|
||
# Handed over from a model page, an image or a combo in Anomalous: the ids the tools take.
|
||
answer["about"] = task["about"]
|
||
answer["about_note"] = ("The user started this task from these in Anomalous; work with them: model_id goes to "
|
||
"set_model / add_lora / get_model, image to get_image_info, a combo's name to place_combo, "
|
||
"a workflow's name to open_workflow; a canvas is the user's tab your work starts from "
|
||
"(describe_canvas reads it).")
|
||
return answer
|
||
|
||
|
||
def _node_list(ctx):
|
||
"""ComfyUI's node list, to say which blocks can run here; None when it cannot be read."""
|
||
from .mcp_canvas_edit import object_info
|
||
try:
|
||
return object_info(ctx) if ctx.get("base_url") else None
|
||
except ToolError:
|
||
return None
|
||
|
||
|
||
def _blocks(ctx):
|
||
"""The ready-made blocks (reference_blocks.py), each a line, for begin_task's answer: what to reach for first."""
|
||
info = _node_list(ctx)
|
||
return {"note": "add_nodes(use=…, want=…) builds the best one this ComfyUI can run, wired in; prefer it. A picture in several "
|
||
"of these steps (the picture, a redraw on another model, details, an enlargement): add_nodes(plan=[{use: "
|
||
"'picture', model}, {use: …}, …]) shows the steps to the user and builds them all on a new tab.",
|
||
**reference_blocks.directory(info)}
|
||
|
||
|
||
def report(ctx, text="", images=None, done=False, reply_to=None, task=None):
|
||
try:
|
||
step, missing = ai_tasks.report(
|
||
ctx.get("client") or "AI", text, images or [], bool(done),
|
||
session=ctx.get("session") or "", task_id=task or ctx.get("task"), reply_to=reply_to,
|
||
)
|
||
except ValueError as error:
|
||
raise ToolError(str(error))
|
||
answer = {"noted": True, "pictures": len(step["results"])}
|
||
if not done:
|
||
answer["next"] = ASK_NEXT
|
||
if any(mark in str(text or "") for mark in ("?", "？")):
|
||
answer["next"] = "Your note asks the user something: ask it with ask_user (options to press) instead. " + ASK_NEXT
|
||
if missing:
|
||
answer["not_found"] = missing
|
||
answer["note"] = "Pictures are paths inside ComfyUI's output folder (subfolder/name.png), as search_images gives them."
|
||
return answer
|
||
|
||
|
||
def log_outside(ctx, what, why="", files=None, task=None):
|
||
try:
|
||
step = ai_outside.log(ctx.get("client") or "AI", what, why, files, session=ctx.get("session") or "",
|
||
task_id=task or ctx.get("task"))
|
||
except ValueError as error:
|
||
raise ToolError(str(error))
|
||
return {"logged": True, "step": step["id"]}
|
||
|
||
|
||
def log_decision(ctx, title, decision, reason="", rejected=None, by="user", related=None, task=None):
|
||
try:
|
||
step = ai_decisions.log(ctx.get("client") or "AI", title, decision, reason, rejected, by, related,
|
||
session=ctx.get("session") or "", task_id=task or ctx.get("task"))
|
||
except ValueError as error:
|
||
raise ToolError(str(error))
|
||
return {"recorded": True, "step": step["id"], "related": step["decision"]["related"]}
|
||
|
||
|
||
def show_panel(ctx, title, blocks, task=None):
|
||
# A prompt box tells the user what suits the canvas's model (the prompt_pose skill's words for them).
|
||
boxed = isinstance(blocks, list) and any(isinstance(block, dict) and block.get("type") == "prompts" for block in blocks)
|
||
advice = ai_skills.for_user(_canvas_model()) if boxed else None
|
||
try:
|
||
panel = ai_panels.show(
|
||
ctx.get("client") or "AI", title, blocks,
|
||
session=ctx.get("session") or "", task_id=task or ctx.get("task"), extra={"prompt_advice": advice} if advice else None,
|
||
)
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(f"{error} Every block's fields: read_guide(topic='panel').")
|
||
asks = [block["id"] for block in panel["blocks"] if block["type"] in ai_panels.INTERACTIVE]
|
||
answer = {"panel": panel["id"], "shown": True,
|
||
"blocks": [{"id": block["id"], "type": block["type"]} for block in panel["blocks"]]}
|
||
if asks:
|
||
answer["questions"] = asks
|
||
answer["note"] = ("The user answers in Anomalous when they get to it; call get_answers with this panel id "
|
||
"(wait_seconds up to 55 waits for it) before going on with what depends on it.")
|
||
else:
|
||
answer["next"] = ASK_NEXT
|
||
return answer
|
||
|
||
|
||
def _routes(ctx, task):
|
||
"""What the user can do with the two models they picked (mcp_routes.py), or None."""
|
||
from .mcp_canvas_edit import object_info
|
||
|
||
task_id = task or ctx.get("task") or ai_tasks.active_task_id(ctx.get("client") or "AI", ctx.get("session") or "")
|
||
box = ai_toolbox.effective(ai_tasks.get_task(task_id)) if task_id else None
|
||
return mcp_routes.offer(box, object_info(ctx)) if len(mcp_routes._main(box)) >= 2 else None
|
||
|
||
|
||
TRIAL_QUESTIONS = {"test", "testing", "test question", "hello", "hi", "ping", "测试", "试试", "试一下"}
|
||
|
||
|
||
def ask_user(ctx, question="", options=None, multiple=False, other=True, wait_seconds=ai_panels.MAX_WAIT, task=None, routes=False):
|
||
"""A question in the AI workbench, where the user watches the work: options to press (one may be
|
||
recommended), a line of their own; waits for the answer. `routes`: Anomalous's own question and options."""
|
||
extra, words = None, {}
|
||
if routes:
|
||
offered = _routes(ctx, task)
|
||
if not offered:
|
||
raise ToolError("No routes to ask about: they need two main models of one family picked for the task. Go on with what the user asked.")
|
||
question, options, multiple, other = mcp_routes.QUESTION, offered["options"], False, True
|
||
extra, words = {"routes": offered["calls"]}, {"title_key": "aiRouteTitle"}
|
||
if not str(question or "").strip():
|
||
raise ToolError("question is empty.")
|
||
if str(question).strip().strip("?？!！.。 ").lower() in TRIAL_QUESTIONS: # with made-up options too ("方案1", "方案2")
|
||
raise ToolError("Not shown to the user: a trial question would wait for an answer nobody gives. Ask when you need "
|
||
"their answer: question (needed), options (the choices to press), multiple, other, wait_seconds.")
|
||
block = ({"type": "choices", "id": "answer", "options": options, "multiple": bool(multiple), "other": bool(other)}
|
||
if options else {"type": "input", "id": "answer", "question": str(question or "")[:200]})
|
||
try:
|
||
panel = ai_panels.show(ctx.get("client") or "AI", question, [block], session=ctx.get("session") or "",
|
||
task_id=task or ctx.get("task"), tool="ask_user", extra=extra, **words)
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(str(error))
|
||
return _answer_of(ctx, panel["id"], wait_seconds)
|
||
|
||
|
||
def propose_repaint_area(ctx, title, marks, areas=None, grow=12, feather=8, note="", wait_seconds=ai_panels.MAX_WAIT):
|
||
"""The area to repaint for the user's marks, theirs beside the AI's for them to pick (ai_repaint.py); waits like ask_user."""
|
||
try:
|
||
panel = ai_repaint.propose(ctx.get("client") or "AI", title, str(marks or ""), areas, grow, feather,
|
||
str(note or "")[:400], session=ctx.get("session") or "")
|
||
except ai_repaint.RepaintError as error:
|
||
raise ToolError(str(error))
|
||
return _answer_of(ctx, panel["id"], wait_seconds)
|
||
|
||
|
||
def _answer_of(ctx, panel_id, wait_seconds):
|
||
"""The answer to the question of panel `panel_id` once the user gives it, waiting at least ASK_WAIT_MIN."""
|
||
client = ctx.get("client") or "AI"
|
||
try:
|
||
answer = ai_answer_wait.collect(client, panel_id, max(ASK_WAIT_MIN, float(wait_seconds or 0)))
|
||
given = next((item for item in answer["answers"] if item["question"] == "answer"), None)
|
||
if not given:
|
||
if answer.get("messages") and answer.get("still_waiting"):
|
||
# Read the user's new instruction before considering an automatic choice.
|
||
answer.update(ai_answer_wait.wait_on(panel_id, allow_recommended=False))
|
||
return {"answered": False, "panel": panel_id, **answer}
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(str(error))
|
||
out = {"answered": True, "panel": panel_id, "next": "Go on with the user's answer; log_decision it when it settles a choice."}
|
||
if answer.get("messages"):
|
||
out["messages"] = answer["messages"]
|
||
if given.get("inpaint_image"):
|
||
out.update(inpaint_image=given["inpaint_image"], inpaint_note=given["note"])
|
||
elif given.get("note"):
|
||
out["next"] = given["note"]
|
||
if given.get("skipped"):
|
||
out.update(skipped=True, note="The user left this to you: decide it yourself, log_decision what you chose and "
|
||
"go on. Do not ask the same again.")
|
||
elif given.get("text"):
|
||
out["text"] = given["text"]
|
||
if given.get("choice") == ["other"]:
|
||
out["note"] = "None of your options: the user wrote their own answer."
|
||
else:
|
||
out.update(choice=given.get("choice"), labels=given.get("labels"))
|
||
if given.get("recommended_on_timeout"):
|
||
out.update(recommended_on_timeout=True, next=ai_answer_wait.TOOK)
|
||
return out
|
||
|
||
|
||
def update_panel(ctx, panel, title=None, blocks=None, closed=None, task=None):
|
||
del ctx, task
|
||
try:
|
||
updated = ai_panels.update(panel, title, blocks, closed)
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(f"{error} Every block's fields: read_guide(topic='panel').")
|
||
return {"updated": True, "blocks": [{"id": block.get("id"), "type": block["type"]}
|
||
for block in updated["blocks"]]}
|
||
|
||
|
||
def run_status(ctx, wait_seconds=0, task=None, downloads=False):
|
||
if downloads:
|
||
return model_downloads(ctx)
|
||
try:
|
||
return ai_task_runs.status(ctx.get("client") or "AI", ctx.get("session") or "", task or ctx.get("task"), wait_seconds)
|
||
except ValueError as error:
|
||
raise ToolError(str(error))
|
||
|
||
|
||
def show_run_map(ctx, title, run=None, task=None):
|
||
client, session = ctx.get("client") or "AI", ctx.get("session") or ""
|
||
task_id = ai_tasks.active_task_id(client, session, task or ctx.get("task"))
|
||
runs = [step for step in (ai_tasks.get_task(task_id) or {}).get("steps", []) if step.get("tool") == "run_workflow"] if task_id else []
|
||
found = next((step for step in reversed(runs) if step.get("map") and (not run or step["id"] == run)), None)
|
||
if not found:
|
||
raise ToolError("No run of this task has its shape yet: it is read once ComfyUI has queued the run (run_workflow).")
|
||
try:
|
||
panel = ai_panels.show(client, title, [{"type": "run_map", "step": found["id"]}], session=session, task_id=task_id,
|
||
tool="show_run_map")
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(str(error))
|
||
boxes = found["map"]["boxes"]
|
||
return {"panel": panel["id"], "shown": True, "run": found["id"], "changes": found.get("changes") or [],
|
||
"boxes": [{"node_id": box["id"], "title": box["title"], "role": box["role"]} for box in boxes]}
|
||
|
||
|
||
def stop_run(ctx, task=None):
|
||
client, session = ctx.get("client") or "AI", ctx.get("session") or ""
|
||
task_id = ai_tasks.active_task_id(client, session, task or ctx.get("task"))
|
||
run = next((step for step in reversed((ai_tasks.get_task(task_id) or {}).get("steps", []))
|
||
if step.get("tool") == "run_workflow"), None) if task_id else None
|
||
if not run:
|
||
raise ToolError("No run in this task yet.")
|
||
try:
|
||
done = ai_run_control.stop(run["id"], by="ai", task_id=task_id)
|
||
except ai_run_control.RunControlError as error:
|
||
raise ToolError(str(error))
|
||
return {**done, "stopped": True, "note": "Stopping: run_status(wait_seconds=10) confirms it ended."}
|
||
|
||
|
||
def get_toolbox(ctx, task=None, skill=None):
|
||
from .mcp_canvas_edit import find_nodes
|
||
|
||
task_id = task or ctx.get("task") or ai_tasks.active_task_id(ctx.get("client") or "AI", ctx.get("session") or "")
|
||
if skill:
|
||
private = ai_privacy.applies(task_id)
|
||
try:
|
||
answer = ai_skills.get(str(skill), _canvas_model() if ai_skills.by_model(str(skill)) else None, private, _node_list(ctx))
|
||
except ai_skills.SkillError as error:
|
||
raise ToolError(str(error))
|
||
ai_skills.note_use(ctx.get("client") or "AI", answer, session=ctx.get("session") or "", task_id=task or ctx.get("task"))
|
||
return answer
|
||
box = ai_toolbox.effective(ai_tasks.get_task(task_id)) if task_id else None
|
||
if not box:
|
||
return {"picked": False, "note": "The user picked nothing for this task; use what fits."}
|
||
answer = {"picked": True, "mode": box["mode"], "note": ai_toolbox.words(box)}
|
||
if box.get("nodes"):
|
||
found = find_nodes(ctx, types=box["nodes"])
|
||
answer["nodes"] = found["nodes"]
|
||
if found.get("not_found"):
|
||
answer["page_only_nodes"] = found["not_found"] # helpers that live only in the page: add_node takes them
|
||
if box.get("packs"):
|
||
from .mcp_canvas_edit import object_info
|
||
|
||
info = object_info(ctx)
|
||
answer["packs"] = []
|
||
for pack in box["packs"]:
|
||
types = sorted(name for name, item in info.items() if ai_toolbox.source_of(item) == pack)
|
||
answer["packs"].append({"pack": pack, "count": len(types), "node_types": [
|
||
{"type": name, "display_name": info[name].get("display_name") or name, "category": info[name].get("category") or ""}
|
||
for name in types[:PACK_TYPES]]})
|
||
answer["packs_note"] = "Every node type of a pack is the user's pick; find_nodes(types=[...]) gives one in full."
|
||
for kind in ("combos", "workflows", "prompts"):
|
||
if box.get(kind):
|
||
answer[kind] = [found for name in box[kind] if (found := ai_message_refs.lookup(kind, name)) is not None]
|
||
if box.get("ways"):
|
||
answer["ways"] = [reference_blocks.call(name) or name for name in box["ways"]]
|
||
if box.get("models"):
|
||
answer["models"] = [ai_message_refs.lookup("models", model_id) or {"model_id": model_id, "missing": True} for model_id in box["models"]]
|
||
for model in answer["models"]:
|
||
if not model.get("missing") and model["model_id"].split(":", 1)[0] != "loras":
|
||
model["in_cards"] = mcp_routes.cards_of(model["model_id"]) # its examples: how the user's own routes use it
|
||
if _routes(ctx, task_id):
|
||
answer["routes"] = mcp_routes.TOOLBOX_NOTE
|
||
return answer
|
||
|
||
|
||
def _canvas_model():
|
||
"""The main model the canvas's sampler uses {name, params}, for advice that depends on its size; None when
|
||
the page cannot tell (no page, several samplers, no loader)."""
|
||
from .mcp_actions import _page
|
||
from .model_advice import main_params
|
||
|
||
try:
|
||
name = (_page("block_model", {"near_node_id": None}) or {}).get("model") or ""
|
||
except ToolError:
|
||
return None
|
||
return {"name": name, "params": main_params(name)} if name else None
|
||
|
||
|
||
def get_answers(ctx, panel=None, wait_seconds=0, task=None):
|
||
try:
|
||
client = ctx.get("client") or "AI"
|
||
task_id = task or ctx.get("task") or None
|
||
if not task_id and ctx.get("session"):
|
||
task_id = ai_tasks.active_task_id(client, ctx["session"]) or ""
|
||
answer = ai_answer_wait.collect(client, panel, wait_seconds, task_id=task_id)
|
||
return answer, ai_marks.image_items(answer.get("messages") or []) # marked pictures, as pictures
|
||
except ai_panels.PanelError as error:
|
||
raise ToolError(str(error))
|
||
|
||
|
||
_TASK_PROP = {"type": "string", "description": "begin_task's id (optional)."}
|
||
|
||
|
||
def _task_tool(name, title, description, run, properties, required=()):
|
||
return name, {"run": run, "context": True, "spec": {
|
||
"name": name, "title": title, "description": description,
|
||
"inputSchema": _schema(properties, required),
|
||
"annotations": {"title": title, "readOnlyHint": name in ("run_status", "get_toolbox"), "destructiveHint": False,
|
||
"idempotentHint": False, "openWorldHint": False},
|
||
}}
|
||
|
||
|
||
def _question_with_images(run):
|
||
"""The question receipt also carries pictures the user sent while it was waiting."""
|
||
def call(ctx, **arguments):
|
||
result = run(ctx, **arguments)
|
||
return result, ai_marks.image_items(result.get("messages") or [])
|
||
return call
|
||
|
||
|
||
TASK_TOOLS = dict([
|
||
_task_tool("track_run", "Track an externally submitted run",
|
||
"Only right after YOU queued a prompt another way (/prompt, a script): attaches the prompt_ids it "
|
||
"returned to your task, so the workbench shows its progress and pictures. Runs nothing; only your own "
|
||
"ids, never guessed; repeating them is safe. Waits up to wait_seconds, then run_status(wait_seconds=55).",
|
||
ai_run_tracking.track_run, {"prompt_ids": {"type": "array", "items": {"type": "string"}, "minItems": 1, "maxItems": 8},
|
||
"task": _TASK_PROP, "wait_seconds": {"type": "number", "minimum": 0, "maximum": 55, "default": 15}}, ["prompt_ids"]),
|
||
_task_tool("begin_task", "Begin a task",
|
||
"Call first: names what the user asked for in Anomalous's AI workbench, where they see, open and undo "
|
||
"each of your steps. Continues the waiting or active task; new_task=true only for a separate request.",
|
||
begin_task, {"title": {"type": "string", "description": "A few words, in the user's language."},
|
||
"new_task": {"type": "boolean", "default": False}},
|
||
["title"]),
|
||
_task_tool("report", "Tell the user",
|
||
"A note in the AI workbench, where the user reads your work: your choices, what to look at, what is "
|
||
"left, your answers. Markdown headings, lists, bold and code are laid out. reply_to: the id of the "
|
||
"user's message you answer. images: output pictures (subfolder/name.png) to keep with the task. "
|
||
"done=true ends it.",
|
||
report, {"text": {"type": "string"},
|
||
"images": {"type": "array", "items": {"type": "string"}, "maxItems": MAX_LIMIT},
|
||
"done": {"type": "boolean", "default": False},
|
||
"reply_to": {"type": "string", "description": "The message id you answer."},
|
||
"task": _TASK_PROP}),
|
||
_task_tool("log_outside", "Say what you did outside Anomalous",
|
||
"Right after using a tool that is not Anomalous's (a command, a script, files, another server): what "
|
||
"you ran, why, and the files it touched, for the task's timeline. Prefer Anomalous's tools.",
|
||
log_outside, {"what": {"type": "string"}, "why": {"type": "string"},
|
||
"files": {"type": "array", "items": {"type": "string"}, "maxItems": ai_outside.MAX_FILES},
|
||
"task": _TASK_PROP},
|
||
["what"]),
|
||
_task_tool("log_decision", "Record a decision",
|
||
"Records a settled decision of the task (a style, a model, a fix, what to drop) where the user looks "
|
||
"back on it: title, decision, reason, rejected options, by (user, ai or together), related step ids. "
|
||
"The user's own choices in Anomalous are kept already.",
|
||
log_decision, {"title": {"type": "string"}, "decision": {"type": "string"}, "reason": {"type": "string"},
|
||
"rejected": {"type": "array", "items": {"type": "string"}, "maxItems": ai_decisions.MAX_REJECTED},
|
||
"by": {"type": "string", "enum": list(ai_decisions.BY), "default": "user"},
|
||
"related": {"type": "array", "items": {"type": "string"}, "maxItems": ai_decisions.MAX_RELATED},
|
||
"task": _TASK_PROP},
|
||
["title", "decision"]),
|
||
_task_tool("show_panel", "Show the user a panel",
|
||
"Draws a panel in the AI workbench, in Anomalous's own style, from blocks: to compare pictures side by "
|
||
"side, present models with their covers, lay out settings, a table, and to ask (buttons or a line of "
|
||
"text). Use a new panel for each new phase (plan, workflow, then results), in the order they happen. "
|
||
"Use it when a choice is the user's, or to show results better than words do. Blocks: heading, text, "
|
||
"images, slider, picks, diagram, models, table, list, params, choices, input; "
|
||
"read_guide(topic='panel') gives their fields before your first panel.",
|
||
show_panel, {"title": {"type": "string"},
|
||
"blocks": {"type": "array", "items": {"type": "object"}, "maxItems": ai_panels.MAX_BLOCKS},
|
||
"task": _TASK_PROP},
|
||
["title", "blocks"]),
|
||
_task_tool("ask_user", "Ask the user",
|
||
"Asks the user in the AI workbench and waits: a question with 2-8 options ({label, detail, "
|
||
"recommended}; mark your suggestion recommended=true), and a line for their own answer unless "
|
||
"other=false; no options: a free answer. routes=true: Anomalous's own question on the two models picked. "
|
||
"Use it for every choice that is the user's, not your chat's "
|
||
f"question form. Waits at least {ASK_WAIT_MIN} s, up to wait_seconds; unanswered and keep_waiting: wait on "
|
||
"with get_answers. End every round of work with it.",
|
||
_question_with_images(ask_user), {"question": {"type": "string"},
|
||
"options": {"type": "array", "maxItems": 8, "items": {"type": "object", "properties": {
|
||
"label": {"type": "string"}, "detail": {"type": "string"}, "recommended": {"type": "boolean"}}}},
|
||
"multiple": {"type": "boolean", "default": False}, "other": {"type": "boolean", "default": True},
|
||
"wait_seconds": {"type": "number", "minimum": 0, "maximum": ai_panels.MAX_WAIT, "default": ai_panels.MAX_WAIT},
|
||
"routes": {"type": "boolean"}, "task": _TASK_PROP}),
|
||
_task_tool("propose_repaint_area", "Agree on the area to repaint",
|
||
"Before repainting only the marked areas: the user's marks are rough, so show them the area you would "
|
||
"repaint. marks: the marks' message id. areas: boxes or ellipses {shape, x, y, w, h} as fractions of the "
|
||
"picture (x, y its top left), or none to keep the marks; grow: pixels added around; feather: pixels of "
|
||
"soft edge. The user sees theirs and yours side by side and picks one, or marks again; with an area "
|
||
"picked the answer has inpaint_image (the picture with that area as its alpha, in ComfyUI's input "
|
||
"folder) and inpaint_note: add_nodes(use=\"repaint\", image=inpaint_image) builds the repaint. "
|
||
"Waits like ask_user.",
|
||
_question_with_images(propose_repaint_area), {"title": {"type": "string", "description": "A few words, in the user's language."},
|
||
"marks": {"type": "string"},
|
||
"areas": {"type": "array", "maxItems": ai_repaint.MAX_AREAS, "items": {"type": "object", "properties": {
|
||
"shape": {"type": "string", "enum": list(ai_repaint.SHAPES)}, "x": {"type": "number"},
|
||
"y": {"type": "number"}, "w": {"type": "number"}, "h": {"type": "number"}}}},
|
||
"grow": {"type": "integer", "minimum": 0, "maximum": ai_repaint.MAX_GROW, "default": 12},
|
||
"feather": {"type": "integer", "minimum": 0, "maximum": ai_repaint.MAX_FEATHER, "default": 8},
|
||
"note": {"type": "string", "description": "Why you moved it, in a line."},
|
||
"wait_seconds": {"type": "number", "minimum": 0, "maximum": ai_panels.MAX_WAIT, "default": ai_panels.MAX_WAIT}},
|
||
["title", "marks"]),
|
||
_task_tool("update_panel", "Change a panel",
|
||
"Adds blocks at the BOTTOM of a panel, keeping what the user saw: send only the new ones. A block "
|
||
"sent with its returned id and type is corrected in place (a table, a question). title changes "
|
||
"it; closed=true stops answers. read_guide(topic='panel') has the rest.",
|
||
update_panel, {"panel": {"type": "string"}, "title": {"type": "string"},
|
||
"blocks": {"type": "array", "items": {"type": "object"}, "maxItems": ai_panels.MAX_BLOCKS},
|
||
"closed": {"type": "boolean"}, "task": _TASK_PROP}, ["panel"]),
|
||
_task_tool("run_status", "How the run is going",
|
||
"Your task's latest run: none, queued (runs ahead), running (node, steps, percent), done (its "
|
||
"pictures; the first answer telling you it ended attaches up to 4) or failed; summary says it in a "
|
||
"sentence. wait_seconds (up to 55) waits for the end. Call it instead of guessing or looking in "
|
||
"folders. downloads=true: how the model downloads are going instead.",
|
||
run_status, {"wait_seconds": {"type": "number", "minimum": 0, "maximum": ai_task_runs.MAX_STATUS_WAIT, "default": 0},
|
||
"task": _TASK_PROP, "downloads": {"type": "boolean"}}),
|
||
_task_tool("show_run_map", "Show what the run really was",
|
||
"Puts the shape of your task's latest run (or `run`, a run step id) on the user's stage: the nodes it "
|
||
"really ran as boxes (models, LoRAs with weights, prompts, size, sampler settings, saving), linked, "
|
||
"each one locates its node on the canvas, with what changed since the run before. Anomalous reads it "
|
||
"from the prompt ComfyUI queued, so use this instead of drawing the workflow yourself; add your "
|
||
"explanation with report. title: a few words in the user's language.",
|
||
show_run_map, {"title": {"type": "string"}, "run": {"type": "string"}, "task": _TASK_PROP}, ["title"]),
|
||
_task_tool("stop_run", "Stop the run",
|
||
"Stops your task's run: its queued prompts leave the queue, the running one is interrupted (never "
|
||
"anyone else's). When run_status warns it is far too big and the user agrees, or when the user asks.",
|
||
stop_run, {"task": _TASK_PROP}),
|
||
_task_tool("get_toolbox", "What the user picked for you",
|
||
"The node types, node packs, combos, workflows, prompts and models the user picked for this task, in "
|
||
"full, and how strictly: prefer (use them first) or only (add_node and new_workflow accept only those "
|
||
"node types, the packs' and ComfyUI's built-in ones). begin_task says when there are any. "
|
||
"skill: a skill begin_task lists, instead.",
|
||
get_toolbox, {"task": _TASK_PROP, "skill": {"type": "string"}}),
|
||
_task_tool("get_answers", "Read the user's answers and messages",
|
||
"What the user picked or typed on your panels and wrote to you in the AI workbench since you last "
|
||
"asked, each once (kind=change: something they changed themselves; keep it). panel: one panel. "
|
||
"wait_seconds (up to 55) waits when nothing is there. still_waiting: questions not answered, and "
|
||
"keep_waiting=true: call it again; skipped=true: they left it to you, decide yourself.",
|
||
get_answers, {"panel": {"type": "string"},
|
||
"wait_seconds": {"type": "number", "minimum": 0, "maximum": ai_panels.MAX_WAIT, "default": 0},
|
||
"task": _TASK_PROP}),
|
||
])
|