Emil_Shanaty/aicc-capsule
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chat: interactive mode (natural language -> tool calls), shared LLM driver refactor, run_chat.sh; verified with gemma4:e2b and gemma4:12b

Browse Source
This commit is contained in:
opencode
2026-08-08 12:27:05 +03:00
parent 9f877339f8
commit 346d918118
70 changed files with 950 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
testbed/demo.py
+17 -295
View File
@@ -33,8 +33,7 @@ from pathlib import Path
from typing import Any
from aicc.client import AICCClient
from aicc.errors import ToolError
from aicc.protocol import SessionInit, ToolSpec
from aicc.protocol import SessionInit
from aicc.transport.websocket import WebSocketClientTransport
from PIL import Image
@@ -42,51 +41,6 @@ DEFAULT_URL = "ws://127.0.0.1:8765"
DEFAULT_BASE_URL = "http://localhost:11434/v1"
DEFAULT_MODEL = "gemma4:e2b"
NUDGE = (
"You have not called a tool. Continue the mission: check the CURRENT STATE note, "
"then call a tool (turn or move, or interact if within 1.6 m)."
)
MISSION = (
"You are a capsule AI inside a 16x16 m room. Your mission: reach the glowing "
"beacon pillar and activate it with the interact tool. "
"World: yaw 0 faces +z, 90 faces +x, coordinates in meters; three wooden "
"crates block parts of the room. "
"Perception: proprioception gives position/heading; world_query gives the room "
"layout and the beacon's exact coordinates (call it once at the start); "
"vision gives the first-person frame; hear tells you how close you are to the beacon. "
"Every tool result is prefixed with a CURRENT STATE note: read it — it tells you "
"your position, heading, distance to the beacon, and the bearing to turn toward it. "
"Standard loop, one action per turn: "
"1. call proprioception (and hear every few steps); "
"2. compute the bearing to the beacon and turn toward it (turn yaw_deg = that bearing, max 90 per call); "
"3. move forward 1.0 m; if the result says collision, turn 60-90 degrees and go around, "
"then re-aim. "
"Never walk into a crate or wall twice in a row. "
"Call interact ONLY when proprioception puts you within 1.6 m of the beacon. "
"Keep tool calls short and decisive. Stop once the beacon is active."
)
COLOR_NAMES = [
("beacon_cyan", (120, 210, 235)),
("beacon_yellow", (255, 190, 90)),
("crate_red", (178, 64, 54)),
("crate_blue", (64, 96, 178)),
("crate_olive", (128, 128, 60)),
("floor_gray", (60, 60, 66)),
("wall_gray", (96, 96, 106)),
("dark", (30, 30, 36)),
]
def _nearest_color(r: float, g: float, b: float) -> str:
best, best_d = "dark", 1e9
for name, (cr, cg, cb) in COLOR_NAMES:
d = (r - cr) ** 2 + (g - cg) ** 2 + (b - cb) ** 2
if d < best_d:
best, best_d = name, d
return best
class FrameRecorder:
"""Visual mode: saves first-person frames and a sensor-built top-down map
@@ -171,63 +125,11 @@ class FrameRecorder:
return summary
def vision_digest(png_b64: str, cols: int = 12, rows: int = 9) -> str:
"""Coarse color-grid summary of a vision frame (agent-side preprocessing)."""
try:
from PIL import Image
img = Image.open(io.BytesIO(base64.b64decode(png_b64))).convert("RGB")
except Exception: # noqa: BLE001 - digest is best-effort
return "(no digest: unreadable frame)"
small = img.resize((cols, rows))
grid: list[list[tuple[int, int, int]]] = []
for r in range(rows):
row: list[tuple[int, int, int]] = []
for c in range(cols):
px = small.getpixel((c, r))
if isinstance(px, (tuple, list)) and len(px) >= 3:
row.append((int(px[0]), int(px[1]), int(px[2])))
elif isinstance(px, (int, float)):
v = int(px)
row.append((v, v, v))
else:
row.append((0, 0, 0))
grid.append(row)
lines = []
for r in range(rows):
lines.append(" ".join(_nearest_color(*grid[r][c]) for c in range(cols)))
return f"frame digest ({cols}x{rows}, center = where you face):\n" + "\n".join(
lines
)
# ---------------------------------------------------------------------------
# LLM agent (OpenAI-compatible)
# ---------------------------------------------------------------------------
def to_openai_tools(tools: list[ToolSpec]) -> list[dict[str, Any]]:
out = []
for t in tools:
if t.id in (
"echo",
"boom",
"bump",
): # conformance-only tools: not for the agent
continue
out.append(
{
"type": "function",
"function": {
"name": t.id,
"description": t.description,
"parameters": t.input_schema,
},
}
)
return out
async def run_llm_agent(
client: AICCClient,
manifest: SessionInit,
@@ -238,205 +140,25 @@ async def run_llm_agent(
max_steps: int,
recorder: FrameRecorder | None = None,
) -> dict[str, Any]:
from openai import AsyncOpenAI
"""Autonomous LLM run: controller + nudge/correct loop (see llm_agent)."""
from testbed.llm_agent import LLMController, run_llm_agent_loop
ac = AsyncOpenAI(base_url=base_url, api_key=api_key)
tools = to_openai_tools(manifest.tools)
messages: list[dict[str, Any]] = [
{"role": "system", "content": MISSION},
{
"role": "user",
"content": "Begin your mission. Use the tools to reach and activate the beacon.",
},
]
summary: dict[str, Any] = {
"steps": 0,
"tool_calls": 0,
"result": None,
"interacted": False,
}
controller = LLMController(
client,
manifest,
base_url=base_url,
api_key=api_key,
model=model,
log=lambda role, msg: print(f"[{role}] {msg}"),
)
try:
return await run_llm_agent_loop(controller, max_steps, log=print_log, recorder=recorder)
except RuntimeError as exc:
return {"steps": 0, "tool_calls": 0, "result": str(exc), "interacted": False}
# Agent-side working memory: a compact summary of state the model has
# already observed via sensors, prefixed to tool results so small models
# don't drift (protocol §9: agent memory is the agent's concern).
beacon_pos: tuple[float, float] | None = None
pos: tuple[float, float] | None = None
yaw: float = 0.0
def state_hint() -> str:
if pos is None:
return ""
if beacon_pos is not None:
dist = math.hypot(beacon_pos[0] - pos[0], beacon_pos[1] - pos[1])
bearing = (
math.degrees(math.atan2(beacon_pos[0] - pos[0], beacon_pos[1] - pos[1]))
- yaw
) % 360.0
return (
f"CURRENT STATE: you are at ({pos[0]:.2f}, {pos[1]:.2f}), heading {yaw:.1f} deg; "
f"the beacon is {dist:.2f} m away at relative bearing {bearing:.0f} deg. "
f"Action: turn yaw_deg={bearing:.0f} to face it, then move forward."
)
return f"CURRENT STATE: you are at ({pos[0]:.2f}, {pos[1]:.2f}), heading {yaw:.1f} deg."
def observe(output: dict[str, Any]) -> None:
nonlocal pos, yaw
if isinstance(output.get("position"), dict):
pos = (output["position"]["x"], output["position"]["z"])
if isinstance(output.get("rotation"), dict):
yaw = output["rotation"]["yaw_deg"] % 360.0
def log(role: str, msg: str) -> None:
print(f"[{role}] {msg}")
# Progress monitor: if the capsule keeps moving away from the beacon,
# reflect that back to the model as a correction (framework-style loop).
dist_history: list[float] = []
corrections = 0
last_correct_step = -99
last_collision_step = -99
def distance_to_beacon() -> float:
if pos is None or beacon_pos is None:
return float("nan")
return math.hypot(beacon_pos[0] - pos[0], beacon_pos[1] - pos[1])
def maybe_correct(step: int) -> None:
nonlocal corrections, last_correct_step
if pos is None or beacon_pos is None:
return
dist = distance_to_beacon()
if not math.isfinite(dist) or dist < 2.0:
return
dist_history.append(dist)
recent = dist_history[-4:]
moving_away = len(recent) == 4 and all(
recent[i] > recent[i + 1] for i in range(3)
)
bearing = (
math.degrees(math.atan2(beacon_pos[0] - pos[0], beacon_pos[1] - pos[1]))
- yaw
) % 360.0
deviation = min(bearing, 360.0 - bearing)
facing_wrong = dist > 4.0 and deviation > 40.0 and step - last_correct_step > 3
detouring = step - last_collision_step < 3
if (moving_away or (facing_wrong and not detouring)) and corrections < 5:
corrections += 1
last_correct_step = step
hint = state_hint()
msg = (
"You are not making progress toward the beacon. Stop and follow the CURRENT "
f"STATE note exactly: {hint} Call turn with that yaw_deg, then move forward."
)
messages.append({"role": "user", "content": msg})
log("agent", "(correction: re-aim at the beacon)")
dist_history.clear()
for step in range(max_steps):
summary["steps"] = step + 1
if recorder is not None and pos is not None:
await recorder.snap(client, pos, yaw)
try:
resp = await ac.chat.completions.create(
model=model, messages=messages, tools=tools, tool_choice="auto"
)
except Exception as exc: # noqa: BLE001 - provider errors end the demo
log("agent", f"LLM error: {exc}")
summary["result"] = f"LLM error: {exc}"
return summary
choice = resp.choices[0]
if choice.message.content:
log("agent", choice.message.content[:400])
messages.append({"role": "assistant", "content": choice.message.content})
calls = choice.message.tool_calls
if not calls:
# Nudge a model that drifted into plain text back to acting.
silent = (not choice.message.content) or len(choice.message.content) < 8
already_nudged = bool(messages) and messages[-1].get("content") == NUDGE
if silent or already_nudged:
summary["result"] = "model produced no tool call"
return summary
messages.append({"role": "user", "content": NUDGE})
log("agent", "(nudge: no tool call; continue the mission)")
continue
messages.append(
{
"role": "assistant",
"content": choice.message.content or "",
"tool_calls": [
{
"id": c.id,
"type": "function",
"function": {
"name": c.function.name,
"arguments": c.function.arguments,
},
}
for c in calls
],
}
)
for tc in calls:
name = tc.function.name
try:
args = json.loads(tc.function.arguments or "{}")
except json.JSONDecodeError:
args = {}
log("tool", f"{name}({json.dumps(args)})")
try:
result = await client.call_tool(name, args)
output = result.output
extra = ""
if name == "vision" and isinstance(output.get("png_b64"), str):
extra = "\n" + vision_digest(output["png_b64"])
output = dict(output, png_b64="<binary, decoded for digest>")
log("bridge", f"ok {json.dumps(output)[:500]}{extra}")
messages.append(
{
"role": "tool",
"tool_call_id": tc.id,
"content": json.dumps(result.output) + extra,
}
)
summary["tool_calls"] += 1
if name == "world_query":
b = output.get("beacon")
if isinstance(b, dict):
beacon_pos = (b["x"], b["z"])
observe(output)
hint = state_hint()
if hint:
messages[-1] = dict(
messages[-1], content=hint + "\n" + messages[-1]["content"]
)
maybe_correct(step)
if name == "move" and output.get("collision"):
last_collision_step = step
obj = output.get("collision_normal")
hint_msg = (
f"You collided with an obstacle (normal {obj}). Turn yaw_deg=90 and move "
"forward twice to get around it, then follow the CURRENT STATE note to re-aim."
)
if not (messages and messages[-1].get("content") == hint_msg):
messages.append({"role": "user", "content": hint_msg})
log("agent", "(collision: go around)")
if name == "interact" and output.get("success"):
summary["interacted"] = True
summary["result"] = output.get("message", "beacon activated")
return summary
except ToolError as e:
log("bridge", f"error {e.code}: {e.message}")
messages.append(
{
"role": "tool",
"tool_call_id": tc.id,
"content": json.dumps({"error": e.code, "message": e.message}),
}
)
summary["tool_calls"] += 1
summary["result"] = f"exceeded {max_steps} steps"
return summary
def print_log(role: str, msg: str) -> None:
print(f"[{role}] {msg}")
# ---------------------------------------------------------------------------