repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
Mirrored from https://github.com/yingqi-z20/Agent-libOS
stars
latest
clone command
git clone gitlawb://did:key:z6MkqRzA...RfoM/yingqi-z20-Agen...git clone gitlawb://did:key:z6MkqRzA.../yingqi-z20-Agen...d98dd2c9IPC1d ago| #1 | from __future__ import annotations |
| #2 | |
| #3 | import asyncio |
| #4 | import tempfile |
| #5 | import unittest |
| #6 | |
| #7 | from agent_libos import Runtime |
| #8 | from agent_libos.llm.client import LLMCompletion |
| #9 | from scripts.human_llm_chat import EchoResponder, ModelResponder, run_chat |
| #10 | |
| #11 | |
| #12 | class HumanLLMChatScriptTests(unittest.TestCase): |
| #13 | def test_chat_uses_human_question_and_output_tools(self) -> None: |
| #14 | report = asyncio.run( |
| #15 | run_chat( |
| #16 | responder=EchoResponder(), |
| #17 | max_turns=5, |
| #18 | auto_messages=["hello", "/exit"], |
| #19 | echo=False, |
| #20 | ) |
| #21 | ) |
| #22 | |
| #23 | self.assertEqual(report["process_status"], "exited") |
| #24 | self.assertEqual(report["turns"], 1) |
| #25 | self.assertEqual(report["history"], [{"role": "user", "content": "hello"}, {"role": "assistant", "content": "Echo: hello"}]) |
| #26 | self.assertIn("Assistant: Echo: hello", report["outputs"]) |
| #27 | self.assertIn("Assistant: goodbye.", report["outputs"]) |
| #28 | self.assertEqual( |
| #29 | report["actions"], |
| #30 | [None, "ask_human", "human_output", None, "ask_human", "human_output", "process_exit"], |
| #31 | ) |
| #32 | |
| #33 | def test_model_responder_persists_nested_text_llm_call(self) -> None: |
| #34 | responder = ModelResponder.__new__(ModelResponder) |
| #35 | responder.system_prompt = "System prompt" |
| #36 | responder.client = FakeTextLLMClient() |
| #37 | responder._runtime = None |
| #38 | responder._pid = None |
| #39 | |
| #40 | with tempfile.TemporaryDirectory() as temp_dir: |
| #41 | db = f"{temp_dir}/runtime.sqlite" |
| #42 | report = asyncio.run( |
| #43 | run_chat( |
| #44 | db=db, |
| #45 | responder=responder, |
| #46 | max_turns=5, |
| #47 | auto_messages=["hello", "/exit"], |
| #48 | echo=False, |
| #49 | ) |
| #50 | ) |
| #51 | |
| #52 | runtime = Runtime.open(db) |
| #53 | try: |
| #54 | calls = [call for call in runtime.store.list_llm_calls(report["pid"]) if call.purpose == "script_human_chat_reply"] |
| #55 | |
| #56 | self.assertEqual(len(calls), 1) |
| #57 | self.assertEqual(calls[0].response_content, "model reply") |
| #58 | self.assertEqual(calls[0].usage["total_tokens"], 6) |
| #59 | self.assertEqual(calls[0].reasoning, {"summary": "fake text response"}) |
| #60 | self.assertEqual(calls[0].messages[-1]["content"], "hello") |
| #61 | finally: |
| #62 | runtime.close() |
| #63 | |
| #64 | |
| #65 | class FakeTextLLMClient: |
| #66 | model = "fake-text-model" |
| #67 | |
| #68 | def complete_with_metadata(self, messages, *, json_mode: bool) -> LLMCompletion: |
| #69 | return LLMCompletion( |
| #70 | content="model reply", |
| #71 | tool_calls=[], |
| #72 | raw={"id": "fake_raw"}, |
| #73 | api="chat", |
| #74 | response_id="fake_resp", |
| #75 | request_id="fake_req", |
| #76 | model=self.model, |
| #77 | usage={"prompt_tokens": 4, "completion_tokens": 2, "total_tokens": 6}, |
| #78 | reasoning={"summary": "fake text response"}, |
| #79 | ) |
| #80 | |
| #81 | |
| #82 | if __name__ == "__main__": |
| #83 | unittest.main() |
| #84 |