yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

yingqi-z20-Agent-libOS — gitlawb

#1	from __future__ import annotations
#2
#3	import asyncio
#4	import unittest
#5	from types import SimpleNamespace
#6	from typing import Any
#7
#8	from agent_libos.llm.client import LLMClient
#9
#10
#11	class LLMClientTests(unittest.TestCase):
#12	def test_responses_action_request_converts_chat_tools_and_parses_function_calls(self) -> None:
#13	response = SimpleNamespace(
#14	id="resp_123",
#15	model="gpt-test",
#16	usage=SimpleNamespace(input_tokens=11, output_tokens=3, total_tokens=14),
#17	output_text="",
#18	output=[
#19	SimpleNamespace(type="reasoning", summary=[{"text": "choose write_text_file"}]),
#20	SimpleNamespace(
#21	type="function_call",
#22	id="fc_1",
#23	call_id="call_1",
#24	name="write_text_file",
#25	arguments='{"path":"out.txt","content":"ok"}',
#26	)
#27	],
#28	)
#29	fake = FakeAsyncOpenAI(responses=FakeResponses(response))
#30	client = LLMClient(model="gpt-test", api_key="key", api_mode="responses")
#31	client._async_client = fake
#32
#33	completion = asyncio.run(
#34	client.acomplete_action(
#35	messages=[
#36	{"role": "system", "content": "system rules"},
#37	{"role": "user", "content": "write a file"},
#38	],
#39	tools=[
#40	{
#41	"type": "function",
#42	"function": {
#43	"name": "write_text_file",
#44	"description": "Write text.",
#45	"parameters": {"type": "object", "properties": {"path": {"type": "string"}}},
#46	},
#47	}
#48	],
#49	)
#50	)
#51
#52	payload = fake.responses.payloads[0]
#53	self.assertEqual(payload["instructions"], "system rules")
#54	self.assertEqual(payload["input"], [{"role": "user", "content": "write a file"}])
#55	self.assertEqual(payload["tools"][0]["name"], "write_text_file")
#56	self.assertEqual(payload["tools"][0]["type"], "function")
#57	self.assertFalse(payload["tools"][0]["strict"])
#58	self.assertFalse(payload["parallel_tool_calls"])
#59	self.assertFalse(payload["store"])
#60	self.assertEqual(completion.api, "responses")
#61	self.assertEqual(completion.response_id, "resp_123")
#62	self.assertEqual(completion.tool_calls[0]["call_id"], "call_1")
#63	self.assertEqual(completion.tool_calls[0]["name"], "write_text_file")
#64	self.assertEqual(completion.usage["total_tokens"], 14)
#65	self.assertEqual(completion.reasoning[0]["summary"][0]["text"], "choose write_text_file")
#66
#67	def test_responses_text_request_uses_json_mode_when_requested(self) -> None:
#68	response = SimpleNamespace(id="resp_json", model="gpt-test", output_text='{"ok":true}', output=[])
#69	fake = FakeAsyncOpenAI(responses=FakeResponses(response))
#70	client = LLMClient(model="gpt-test", api_key="key", api_mode="responses", verbosity="low")
#71	client._async_client = fake
#72
#73	content = asyncio.run(client.acomplete([{"role": "user", "content": "return json"}], json_mode=True))
#74
#75	payload = fake.responses.payloads[0]
#76	self.assertEqual(content, '{"ok":true}')
#77	self.assertEqual(payload["text"]["format"], {"type": "json_object"})
#78	self.assertEqual(payload["text"]["verbosity"], "low")
#79
#80	def test_auto_mode_uses_chat_for_custom_base_url(self) -> None:
#81	chat_completion = SimpleNamespace(
#82	id="chatcmpl_123",
#83	model="compat-model",
#84	usage=SimpleNamespace(prompt_tokens=7, completion_tokens=2, total_tokens=9),
#85	choices=[
#86	SimpleNamespace(
#87	finish_reason="tool_calls",
#88	message=SimpleNamespace(
#89	content="",
#90	reasoning_content="select process_exit",
#91	tool_calls=[
#92	SimpleNamespace(
#93	id="tool_1",
#94	function=SimpleNamespace(name="process_exit", arguments='{"payload":{"ok":true}}'),
#95	)
#96	],
#97	),
#98	)
#99	],
#100	)
#101	fake = FakeAsyncOpenAI(chat=FakeChat(FakeChatCompletions(chat_completion)))
#102	client = LLMClient(
#103	base_url="https://example.com/compatible/v1",
#104	model="compat-model",
#105	api_key="key",
#106	api_mode="auto",
#107	)
#108	client._async_client = fake
#109
#110	completion = asyncio.run(
#111	client.acomplete_action(
#112	messages=[{"role": "user", "content": "exit"}],
#113	tools=[
#114	{
#115	"type": "function",
#116	"function": {
#117	"name": "process_exit",
#118	"description": "Exit.",
#119	"parameters": {"type": "object", "properties": {}},
#120	},
#121	}
#122	],
#123	)
#124	)
#125
#126	self.assertEqual(fake.chat.completions.payloads[0]["model"], "compat-model")
#127	self.assertFalse(fake.responses.payloads)
#128	self.assertEqual(completion.api, "chat")
#129	self.assertEqual(completion.tool_calls[0]["name"], "process_exit")
#130	self.assertEqual(completion.usage["total_tokens"], 9)
#131	self.assertEqual(completion.reasoning, "select process_exit")
#132
#133	def test_custom_chat_empty_response_retries_with_thinking_disabled(self) -> None:
#134	empty = SimpleNamespace(
#135	id="chatcmpl_empty",
#136	model="compat-model",
#137	choices=[SimpleNamespace(finish_reason="length", message=SimpleNamespace(content="", tool_calls=[]))],
#138	)
#139	ok = SimpleNamespace(
#140	id="chatcmpl_ok",
#141	model="compat-model",
#142	choices=[SimpleNamespace(finish_reason="stop", message=SimpleNamespace(content="OK", tool_calls=[]))],
#143	)
#144	completions = FakeChatCompletions([empty, ok])
#145	fake = FakeAsyncOpenAI(chat=FakeChat(completions))
#146	client = LLMClient(
#147	base_url="https://example.com/compatible/v1",
#148	model="compat-model",
#149	api_key="key",
#150	api_mode="chat",
#151	)
#152	client._async_client = fake
#153
#154	content = asyncio.run(client.acomplete([{"role": "user", "content": "say OK"}], json_mode=False))
#155
#156	self.assertEqual(content, "OK")
#157	self.assertEqual(len(completions.payloads), 2)
#158	self.assertEqual(completions.payloads[1]["extra_body"], {"enable_thinking": False})
#159
#160
#161	class FakeAsyncOpenAI:
#162	def __init__(self, responses: Any \| None = None, chat: Any \| None = None):
#163	self.responses = responses or FakeResponses(SimpleNamespace(id="unused", model="unused", output_text="", output=[]))
#164	self.chat = chat or FakeChat(FakeChatCompletions(SimpleNamespace(choices=[])))
#165
#166
#167	class FakeResponses:
#168	def __init__(self, response: Any):
#169	self.response = response
#170	self.payloads: list[dict[str, Any]] = []
#171
#172	async def create(self, **payload: Any) -> Any:
#173	self.payloads.append(payload)
#174	return self.response
#175
#176
#177	class FakeChat:
#178	def __init__(self, completions: Any):
#179	self.completions = completions
#180
#181
#182	class FakeChatCompletions:
#183	def __init__(self, completion: Any):
#184	self.completions = list(completion) if isinstance(completion, list) else [completion]
#185	self.payloads: list[dict[str, Any]] = []
#186
#187	async def create(self, **payload: Any) -> Any:
#188	self.payloads.append(payload)
#189	return self.completions.pop(0)
#190
#191
#192	if __name__ == "__main__":
#193	unittest.main()
#194

z6MkqRzA/yingqi-z20-Agent-libOS