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 collections.abc import Sequence
#7	from typing import Any, Protocol
#8
#9	from agent_libos import Runtime
#10	from agent_libos.config import DEFAULT_CONFIG
#11	from agent_libos.llm.client import LLMClient, LLMCompletion
#12	from agent_libos.models import AgentImage, LLMCallRecord, ProcessStatus, ResourceBudget
#13	from agent_libos.utils.ids import new_id, utc_now
#14	from agent_libos.utils.serde import to_jsonable
#15	from scripts.llm_context_probe import last_tool_result, recent_events
#16
#17	_RUNTIME_DEFAULTS = DEFAULT_CONFIG.runtime
#18	_SCRIPT_DEFAULTS = DEFAULT_CONFIG.scripts
#19
#20	CHAT_IMAGE_ID = "chat-image:v0"
#21	CHAT_IMAGE_NAME = "ChatImage"
#22	DEFAULT_EXIT_WORDS = ["/exit", "/quit"]
#23	DEFAULT_SYSTEM_PROMPT = ("You are a helpful assistant in a terminal chat. Reply to the user's latest message directly. "
#24	"Keep answers concise unless the user asks for detail.")
#25
#26
#27	class ChatResponder(Protocol):
#28	def reply(self, history: list[dict[str, str]], user_message: str) -> str:
#29	raise NotImplementedError
#30
#31
#32	def main() -> None:
#33	parser = argparse.ArgumentParser(
#34	description=("Run a traditional human/LLM chat loop through Agent libOS HumanObject tools: "
#35	"ask_human for user input and human_output for assistant replies."))
#36	parser.add_argument(
#37	"--db",
#38	default=_RUNTIME_DEFAULTS.local_store_target,
#39	help=f"Runtime SQLite database path, or '{_RUNTIME_DEFAULTS.local_store_target}' for in-memory.",
#40	)
#41	parser.add_argument(
#42	"--max-turns",
#43	type=int,
#44	default=_SCRIPT_DEFAULTS.chat_max_turns,
#45	help="Maximum user turns before the chat process exits.",
#46	)
#47	parser.add_argument("--max-quanta", type=int, default=None, help="Maximum Agent execution quanta.")
#48	parser.add_argument("--system", default=DEFAULT_SYSTEM_PROMPT, help="System prompt for the chat model.")
#49	parser.add_argument("--exit-word", action="append", default=None,
#50	help="Exit word. Can be repeated. Defaults include /exit, quit, 退出, 再见.", )
#51	parser.add_argument("--auto-message", action="append", default=None,
#52	help="Non-interactive human message. Repeat for multiple turns; /exit is used when exhausted.", )
#53	parser.add_argument("--mock", action="store_true",
#54	help="Use a deterministic local echo responder instead of calling the configured LLM provider.", )
#55	args = parser.parse_args()
#56
#57	responder: ChatResponder = EchoResponder() if args.mock else ModelResponder(system_prompt=args.system)
#58	report = asyncio.run(run_chat(db=args.db, responder=responder, max_turns=args.max_turns, max_quanta=args.max_quanta,
#59	exit_words=args.exit_word or DEFAULT_EXIT_WORDS, auto_messages=args.auto_message, ))
#60	print(json.dumps(report, indent=2, ensure_ascii=False, default=str))
#61
#62
#63	async def run_chat(*, db: str = _RUNTIME_DEFAULTS.local_store_target, responder: ChatResponder \| None = None, max_turns: int = _SCRIPT_DEFAULTS.chat_max_turns,
#64	max_quanta: int \| None = None, exit_words: Sequence[str] = DEFAULT_EXIT_WORDS,
#65	auto_messages: Sequence[str] \| None = None, echo: bool = True, ) -> dict[str, Any]:
#66	if max_turns < 1:
#67	raise ValueError("max_turns must be positive")
#68	runtime = Runtime.open(db)
#69	outputs: list[str] = []
#70	client = HumanChatActionClient(responder=responder or ModelResponder(system_prompt=DEFAULT_SYSTEM_PROMPT),
#71	max_turns=max_turns, exit_words=exit_words, )
#72	runtime.llm.client = client
#73	runtime.register_image(chat_image())
#74
#75	def output_sink(message: str) -> None:
#76	outputs.append(message)
#77	if echo:
#78	print(message, flush=True)
#79
#80	input_fn = _auto_input_fn(auto_messages, echo=echo) if auto_messages is not None else None
#81	runtime.substrate.human.output_sink = output_sink
#82	if input_fn is not None:
#83	runtime.substrate.human.input_reader = input_fn
#84	try:
#85	pid = runtime.process.spawn(image=CHAT_IMAGE_ID, goal=(
#86	"You are an AI assistant interacting via a terminal interface. To ensure a smooth and efficient conversation, please adhere to the following rules:"
#87	"1. Do not repeat the same text, greetings, or explanations in one turn. Keep responses concise and strictly relevant to the new context.",
#88	"2. Every turn MUST conclude with a tool call to either `ask_human` (to receive the next user message) or `process_exit` (when the user wants to exit or the task is done). Never end a turn without calling one of these tools."),
#89	resource_budget=ResourceBudget(max_materialized_tokens=_SCRIPT_DEFAULTS.chat_context_tokens), )
#90	client.bind_runtime(runtime, pid)
#91	default_max_quanta = max_turns * _SCRIPT_DEFAULTS.chat_quanta_per_turn + _SCRIPT_DEFAULTS.chat_quanta_overhead
#92	results = await runtime.arun_until_idle(max_quanta=max_quanta or default_max_quanta)
#93	process = runtime.process.get(pid)
#94	report = {"pid": pid, "turns": client.turns, "process_status": process.status.value,
#95	"actions": [_action_name(result) for result in results], "outputs": outputs, "history": client.history,
#96	"model_calls": client.calls, "results": to_jsonable(results), }
#97	if process.status != ProcessStatus.EXITED:
#98	raise RuntimeError(f"chat process did not exit; status={process.status.value}")
#99	return report
#100	finally:
#101	runtime.close()
#102
#103
#104	class HumanChatActionClient:
#105	def __init__(self, *, responder: ChatResponder, max_turns: int, exit_words: Sequence[str], ):
#106	self.responder = responder
#107	self.max_turns = max_turns
#108	self.exit_words = {word.strip().lower() for word in exit_words if word.strip()}
#109	self.calls = 0
#110	self.turns = 0
#111	self.history: list[dict[str, str]] = []
#112	self._waiting_for_answer = False
#113	self._exit_after_output = False
#114
#115	def bind_runtime(self, runtime: Runtime, pid: str) -> None:
#116	binder = getattr(self.responder, "bind_runtime", None)
#117	if callable(binder):
#118	binder(runtime, pid)
#119
#120	def complete_action(self, messages: list[dict[str, str]], tools: list[dict[str, object]]) -> LLMCompletion:
#121	self.calls += 1
#122	if self._exit_after_output:
#123	return self._completion("process_exit", {"payload": {"turns": self.turns, "reason": "chat_finished"}}, )
#124	if not self._waiting_for_answer:
#125	# The script uses ask_human/human_output as the terminal transport.
#126	# This local client only decides which libOS tool action comes next.
#127	if self.turns >= self.max_turns:
#128	return self._completion("process_exit",
#129	{"payload": {"turns": self.turns, "reason": "max_turns_reached"}}, )
#130	self._waiting_for_answer = True
#131	return self._completion("ask_human", {"question": "Human:"}, )
#132
#133	user_message = _last_human_answer(messages).strip()
#134	self._waiting_for_answer = False
#135	if user_message.lower() in self.exit_words:
#136	self._exit_after_output = True
#137	return self._completion("human_output", {"message": "Assistant: goodbye."})
#138
#139	reply = self.responder.reply(list(self.history), user_message).strip()
#140	if not reply:
#141	reply = "(empty response)"
#142	self.history.append({"role": "user", "content": user_message})
#143	self.history.append({"role": "assistant", "content": reply})
#144	self.turns += 1
#145	if self.turns >= self.max_turns:
#146	self._exit_after_output = True
#147	return self._completion("human_output", {"message": f"Assistant: {reply}"})
#148
#149	def _completion(self, name: str, args: dict[str, Any]) -> LLMCompletion:
#150	return LLMCompletion(content="",
#151	tool_calls=[{"id": f"human_chat_{self.calls}", "name": name, "arguments": json.dumps(args)}], )
#152
#153
#154	class ModelResponder:
#155	def __init__(self, *, system_prompt: str):
#156	self.system_prompt = system_prompt
#157	self.client = LLMClient.from_env()
#158	self._runtime: Runtime \| None = None
#159	self._pid: str \| None = None
#160
#161	def bind_runtime(self, runtime: Runtime, pid: str) -> None:
#162	self._runtime = runtime
#163	self._pid = pid
#164
#165	def reply(self, history: list[dict[str, str]], user_message: str) -> str:
#166	messages = [{"role": "system", "content": self.system_prompt}]
#167	messages.extend(history)
#168	messages.append({"role": "user", "content": user_message})
#169	call_id = new_id("llmcall")
#170	created_at = utc_now()
#171	try:
#172	completion = self.client.complete_with_metadata(messages, json_mode=False)
#173	except Exception as exc:
#174	self._record_llm_call(call_id=call_id, messages=messages, created_at=created_at, error=str(exc))
#175	raise
#176	self._record_llm_call(call_id=call_id, messages=messages, created_at=created_at, completion=completion)
#177	return completion.content
#178
#179	def _record_llm_call(
#180	self,
#181	*,
#182	call_id: str,
#183	messages: list[dict[str, str]],
#184	created_at: str,
#185	completion: LLMCompletion \| None = None,
#186	error: str \| None = None,
#187	) -> None:
#188	if self._runtime is None:
#189	return
#190	self._runtime.store.insert_llm_call(
#191	LLMCallRecord(
#192	call_id=call_id,
#193	pid=self._pid,
#194	image_id=CHAT_IMAGE_ID,
#195	purpose="script_human_chat_reply",
#196	status="error" if error else "ok",
#197	api=completion.api if completion else None,
#198	model=completion.model if completion else self.client.model,
#199	request_id=completion.request_id if completion else None,
#200	response_id=completion.response_id if completion else None,
#201	messages=messages,
#202	tools=[],
#203	request_options={
#204	"json_mode": False,
#205	"client_class": type(self.client).__name__,
#206	"real_llm_client": True,
#207	},
#208	response_content=completion.content if completion else "",
#209	tool_calls=completion.tool_calls if completion else [],
#210	reasoning=completion.reasoning if completion else None,
#211	usage=completion.usage if completion else {},
#212	raw_response=completion.raw if completion else None,
#213	error=error,
#214	created_at=created_at,
#215	completed_at=utc_now(),
#216	)
#217	)
#218
#219
#220	class EchoResponder:
#221	def reply(self, history: list[dict[str, str]], user_message: str) -> str:
#222	return f"Echo: {user_message}"
#223
#224
#225	def chat_image() -> AgentImage:
#226	return AgentImage(image_id=CHAT_IMAGE_ID, name=CHAT_IMAGE_NAME, version="v0",
#227	system_prompt="Traditional human/LLM chat image with only human I/O and process exit tools.",
#228	default_tools=["ask_human", "human_output", "process_exit"], context_policy="recency_first",
#229	required_capabilities=[{"resource": _RUNTIME_DEFAULTS.default_human_resource, "rights": ["write"]}], )
#230
#231
#232	def _auto_input_fn(messages: Sequence[str], *, echo: bool):
#233	remaining = list(messages)
#234
#235	def input_fn(prompt: str) -> str:
#236	answer = remaining.pop(0) if remaining else "/exit"
#237	if echo:
#238	print(f"{prompt}{answer}", flush=True)
#239	return answer
#240
#241	return input_fn
#242
#243
#244	def _last_tool_result(messages: list[dict[str, str]], tool_name: str) -> dict[str, Any]:
#245	result = last_tool_result(messages, tool_name)
#246	if result is not None:
#247	return result
#248	raise AssertionError(f"no visible result for {tool_name}")
#249
#250
#251	def _last_human_answer(messages: list[dict[str, str]]) -> str:
#252	# Prefer recent events over Object Memory payload order; the latter can
#253	# include older tool results after context sorting.
#254	for event in reversed(recent_events(messages)):
#255	if event.get("type") != "human_response":
#256	continue
#257	payload = event.get("payload")
#258	if not isinstance(payload, dict):
#259	continue
#260	decision = payload.get("decision")
#261	if isinstance(decision, dict) and isinstance(decision.get("answer"), str):
#262	return decision["answer"]
#263	return str(_last_tool_result(messages, "ask_human").get("answer", ""))
#264
#265
#266	def _action_name(result: object) -> str \| None:
#267	if not isinstance(result, dict):
#268	return None
#269	action = result.get("action")
#270	if isinstance(action, dict):
#271	return action.get("action")
#272	return None
#273
#274
#275	if __name__ == "__main__":
#276	main()
#277

z6MkqRzA/yingqi-z20-Agent-libOS