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 json |
| #4 | import re |
| #5 | from typing import Any |
| #6 | |
| #7 | |
| #8 | def static_prefix(messages: list[dict[str, Any]]) -> dict[str, Any]: |
| #9 | text = _message_text(messages) |
| #10 | match = re.search(r"Static prefix:\n(?P<payload>.*?)\n\nAppend-only entries:", text, flags=re.DOTALL) |
| #11 | if match is None: |
| #12 | return {} |
| #13 | try: |
| #14 | payload = json.loads(match.group("payload")) |
| #15 | except json.JSONDecodeError: |
| #16 | return {} |
| #17 | return payload if isinstance(payload, dict) else {} |
| #18 | |
| #19 | |
| #20 | def entries(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| #21 | text = _message_text(messages) |
| #22 | if "Append-only entries:" not in text: |
| #23 | return [] |
| #24 | tail = text.split("Append-only entries:", 1)[1] |
| #25 | result: list[dict[str, Any]] = [] |
| #26 | for block in re.split(r"(?m)^---\s*$", tail): |
| #27 | block = block.strip() |
| #28 | if not block: |
| #29 | continue |
| #30 | try: |
| #31 | entry = json.loads(block) |
| #32 | except json.JSONDecodeError: |
| #33 | continue |
| #34 | if isinstance(entry, dict): |
| #35 | result.append(entry) |
| #36 | return result |
| #37 | |
| #38 | |
| #39 | def recent_events(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| #40 | result: list[dict[str, Any]] = [] |
| #41 | for entry in entries(messages): |
| #42 | if entry.get("kind") != "events_delta": |
| #43 | continue |
| #44 | events = entry.get("events") |
| #45 | if isinstance(events, list): |
| #46 | result.extend(event for event in events if isinstance(event, dict)) |
| #47 | return result |
| #48 | |
| #49 | |
| #50 | def tool_result_payloads(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| #51 | result: list[dict[str, Any]] = [] |
| #52 | for entry in entries(messages): |
| #53 | if entry.get("kind") != "memory_delta": |
| #54 | continue |
| #55 | objects = entry.get("objects") |
| #56 | if not isinstance(objects, list): |
| #57 | continue |
| #58 | for obj in objects: |
| #59 | if not isinstance(obj, dict): |
| #60 | continue |
| #61 | payload = obj.get("payload") |
| #62 | if isinstance(payload, dict) and isinstance(payload.get("tool_name"), str): |
| #63 | result.append(payload) |
| #64 | return result |
| #65 | |
| #66 | |
| #67 | def last_tool_result(messages: list[dict[str, Any]], tool_name: str) -> dict[str, Any] | None: |
| #68 | for payload in reversed(tool_result_payloads(messages)): |
| #69 | if payload.get("tool_name") != tool_name: |
| #70 | continue |
| #71 | result = payload.get("result") |
| #72 | if isinstance(result, dict): |
| #73 | return result |
| #74 | return None |
| #75 | |
| #76 | |
| #77 | def _message_text(messages: list[dict[str, Any]]) -> str: |
| #78 | return "\n".join(str(message.get("content", "")) for message in messages) |
| #79 |