Spaces:
Running
Running
File size: 2,589 Bytes
105f2e1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | from agent_core.config import DEFAULT_CADQUERY_OUTPUT_PATH, get_client, get_model_id
from agent_core.prompts import SYSTEM
from agent_core.tools.cadquery_tool import TOOL_SCHEMA as CADQUERY_TOOL_SCHEMA
from agent_core.tools.cadquery_tool import run_execute_cadquery
from agent_core.tools.lux3d_tool import TOOL_SCHEMA as LUX3D_TOOL_SCHEMA
from agent_core.tools.lux3d_tool import run_generate_3d_model
TOOL_HANDLERS = {
"execute_cadquery": lambda **kw: run_execute_cadquery(
kw["code"],
kw.get("output_path", DEFAULT_CADQUERY_OUTPUT_PATH),
),
"generate_3d_model": lambda **kw: run_generate_3d_model(
kw["image_path"],
kw.get("output_path"),
),
}
TOOLS = [
CADQUERY_TOOL_SCHEMA,
LUX3D_TOOL_SCHEMA,
]
def latest_user_prompt(messages: list) -> str | None:
for message in reversed(messages):
if message.get("role") == "user" and isinstance(message.get("content"), str):
return message["content"]
return None
def agent_loop(messages: list):
client = get_client()
model = get_model_id()
while True:
response = client.messages.create(
model=model,
system=SYSTEM,
messages=messages,
tools=TOOLS,
max_tokens=8000,
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
return
results = []
prompt = latest_user_prompt(messages)
for block in response.content:
if block.type == "tool_use":
if block.name == "execute_cadquery":
output = run_execute_cadquery(
code=block.input["code"],
output_path=block.input.get("output_path", DEFAULT_CADQUERY_OUTPUT_PATH),
prompt=prompt,
)
elif block.name == "generate_3d_model":
output = run_generate_3d_model(
image_path=block.input["image_path"],
output_path=block.input.get("output_path"),
prompt=prompt,
)
else:
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
print(f"> {block.name}: {output[:200]}")
results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})
messages.append({"role": "user", "content": results})
|