yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	import argparse
#4	import asyncio
#5	import json
#6	from typing import Any
#7
#8	from agent_libos import Runtime
#9	from agent_libos.config import DEFAULT_CONFIG
#10	from agent_libos.llm.client import LLMCompletion
#11	from agent_libos.models import ProcessStatus
#12	from agent_libos.utils.serde import to_jsonable
#13	from scripts.llm_context_probe import last_tool_result, recent_events
#14
#15	_RUNTIME_DEFAULTS = DEFAULT_CONFIG.runtime
#16	_SCRIPT_DEFAULTS = DEFAULT_CONFIG.scripts
#17
#18
#19	def main() -> None:
#20	parser = argparse.ArgumentParser(
#21	description="Ask the human which workspace file to view, then show that file's content through HumanObject output."
#22	)
#23	parser.add_argument(
#24	"--db",
#25	default=_RUNTIME_DEFAULTS.local_store_target,
#26	help=f"Runtime SQLite database path, or '{_RUNTIME_DEFAULTS.local_store_target}' for in-memory.",
#27	)
#28	parser.add_argument(
#29	"--max-bytes",
#30	type=int,
#31	default=_SCRIPT_DEFAULTS.ask_file_max_bytes,
#32	help="Maximum bytes to read from the selected file.",
#33	)
#34	parser.add_argument(
#35	"--max-quanta",
#36	type=int,
#37	default=_SCRIPT_DEFAULTS.ask_file_max_quanta,
#38	help="Maximum Agent execution quanta to run.",
#39	)
#40	parser.add_argument(
#41	"--auto-answer",
#42	default=None,
#43	help="Non-interactive answer to the file-name question, for example README.md.",
#44	)
#45	args = parser.parse_args()
#46	report = asyncio.run(
#47	run_file_viewer(
#48	db=args.db,
#49	max_bytes=args.max_bytes,
#50	max_quanta=args.max_quanta,
#51	auto_answer=args.auto_answer,
#52	)
#53	)
#54	print(json.dumps(report, indent=2, ensure_ascii=False, default=str))
#55
#56
#57	async def run_file_viewer(
#58	*,
#59	db: str = _RUNTIME_DEFAULTS.local_store_target,
#60	max_bytes: int = _SCRIPT_DEFAULTS.ask_file_max_bytes,
#61	max_quanta: int = _SCRIPT_DEFAULTS.ask_file_max_quanta,
#62	auto_answer: str \| None = None,
#63	echo: bool = True,
#64	) -> dict[str, Any]:
#65	if max_bytes < 1:
#66	raise ValueError("max_bytes must be positive")
#67	runtime = Runtime.open(db)
#68	outputs: list[str] = []
#69	client = AskFileViewerClient(max_bytes=max_bytes)
#70	runtime.llm.client = client
#71
#72	def output_sink(message: str) -> None:
#73	outputs.append(message)
#74	if echo:
#75	print(message, flush=True)
#76
#77	runtime.substrate.human.output_sink = output_sink
#78	try:
#79	pid = runtime.process.spawn(
#80	image=_RUNTIME_DEFAULTS.coding_image_id,
#81	goal=(
#82	"Ask the human which workspace file they want to view. Read that file and show its content "
#83	"to the human. If reading fails, show the failure reason to the human. Then exit."
#84	),
#85	)
#86	results = await runtime.arun_until_idle(
#87	max_quanta=max_quanta,
#88	human_auto_answer=auto_answer,
#89	)
#90	process = runtime.process.get(pid)
#91	report = {
#92	"pid": pid,
#93	"selected_path": client.selected_path,
#94	"displayed": client.displayed,
#95	"error": client.error,
#96	"process_status": process.status.value,
#97	"actions": [_action_name(result) for result in results],
#98	"outputs": outputs,
#99	"model_calls": client.calls,
#100	"results": to_jsonable(results),
#101	}
#102	if process.status != ProcessStatus.EXITED:
#103	raise RuntimeError(f"process did not exit after {max_quanta} quanta; status={process.status.value}")
#104	return report
#105	finally:
#106	runtime.close()
#107
#108
#109	class AskFileViewerClient:
#110	def __init__(self, *, max_bytes: int):
#111	self.max_bytes = max_bytes
#112	self.calls = 0
#113	self.step = 0
#114	self.selected_path: str \| None = None
#115	self.displayed = False
#116	self.error: str \| None = None
#117
#118	def complete_action(self, messages: list[dict[str, str]], tools: list[dict[str, object]]) -> LLMCompletion:
#119	self.calls += 1
#120	if self.step == 0:
#121	# Drive the process through real human/file primitives while keeping
#122	# this script deterministic and testable without a model call.
#123	self.step = 1
#124	return self._completion(
#125	"ask_human",
#126	{
#127	"question": "Which workspace file do you want to view?",
#128	"context": {"path_rule": "Use a path under the runtime workspace root."},
#129	},
#130	)
#131	if self.step == 1:
#132	answer = self._last_tool_result(messages, "ask_human").get("answer")
#133	if not isinstance(answer, str) or not answer.strip():
#134	raise AssertionError("ask_human result did not include a non-empty answer")
#135	self.selected_path = answer.strip()
#136	self.step = 2
#137	return self._completion(
#138	"read_text_file",
#139	{"path": self.selected_path, "max_bytes": self.max_bytes},
#140	)
#141	if self.step == 2:
#142	read_result = self._last_tool_result(messages, "read_text_file", required=False)
#143	if read_result is None:
#144	self.error = self._last_tool_error(messages) or "read_text_file failed without a visible error"
#145	message = f"Could not read {self.selected_path!r}: {self.error}"
#146	else:
#147	content = str(read_result.get("content", ""))
#148	truncated = bool(read_result.get("truncated", False))
#149	suffix = "\n\n[content truncated]" if truncated else ""
#150	message = f"----- {self.selected_path} -----\n{content}{suffix}"
#151	self.displayed = True
#152	self.step = 3
#153	return self._completion("human_output", {"message": message})
#154	if self.step == 3:
#155	self.step = 4
#156	return self._completion(
#157	"process_exit",
#158	{
#159	"payload": {
#160	"selected_path": self.selected_path,
#161	"displayed": self.displayed,
#162	"error": self.error,
#163	}
#164	},
#165	)
#166	raise AssertionError("file viewer action plan is already complete")
#167
#168	def _completion(self, name: str, args: dict[str, Any]) -> LLMCompletion:
#169	return LLMCompletion(
#170	content="",
#171	tool_calls=[{"id": f"file_viewer_{self.calls}", "name": name, "arguments": json.dumps(args)}],
#172	)
#173
#174	def _last_tool_result(
#175	self,
#176	messages: list[dict[str, str]],
#177	tool_name: str,
#178	*,
#179	required: bool = True,
#180	) -> dict[str, Any] \| None:
#181	result = last_tool_result(messages, tool_name)
#182	if result is not None:
#183	return result
#184	if required:
#185	raise AssertionError(f"no visible result for {tool_name}")
#186	return None
#187
#188	def _last_tool_error(self, messages: list[dict[str, str]]) -> str \| None:
#189	for event in reversed(recent_events(messages)):
#190	if event.get("type") != "tool_failed":
#191	continue
#192	payload = event.get("payload")
#193	if isinstance(payload, dict) and isinstance(payload.get("error"), str):
#194	return payload["error"]
#195	return None
#196
#197
#198	def _action_name(result: object) -> str \| None:
#199	if not isinstance(result, dict):
#200	return None
#201	action = result.get("action")
#202	if isinstance(action, dict):
#203	return action.get("action")
#204	return None
#205
#206
#207	if __name__ == "__main__":
#208	main()
#209

z6MkqRzA/yingqi-z20-Agent-libOS