Files
librefang-registry/plugins/auto-summarizer/hooks/ingest.py
T
Evan Hu afcb260554 feat: add 6 production-ready plugins
All plugins are stdlib-only Python with no external dependencies.

- auto-summarizer: extractive conversation summary for long context
  compression, persists per-agent summaries to disk
- conversation-logger: JSONL audit logs per agent with ISO 8601
  timestamps, auto-creates log directory tree
- guardrails: safety filter detecting PII (email, phone, SSN, CC),
  prompt injection patterns, and credential exposure via regex
- keyword-memory: extracts entities (emails, URLs, dates, technical
  terms like camelCase/snake_case/dotted identifiers) as memories
- sentiment-tracker: keyword-based sentiment scoring with intensifiers
  and negation handling, only injects context for non-neutral sentiment
- todo-tracker: detects action items via 7 task patterns, tracks
  completion, deduplicates, persists per-agent with 20-item FIFO limit
2026-03-21 02:51:58 +09:00

50 lines
1.3 KiB
Python

#!/usr/bin/env python3
"""Auto-summarizer ingest hook.
Returns the stored conversation summary as a memory fragment so agents
maintain awareness of prior conversation context.
Receives via stdin:
{"type": "ingest", "agent_id": "...", "message": "user message text"}
Prints to stdout:
{"type": "ingest_result", "memories": [{"content": "..."}]}
"""
import json
import os
import sys
def get_summary_path(agent_id):
"""Return the path to the summary file for the given agent."""
home = os.path.expanduser("~")
return os.path.join(
home, ".librefang", "plugins", "auto-summarizer", f"{agent_id}.summary"
)
def main():
request = json.loads(sys.stdin.read())
agent_id = request.get("agent_id", "unknown")
memories = []
summary_path = get_summary_path(agent_id)
if os.path.isfile(summary_path):
try:
with open(summary_path, "r", encoding="utf-8") as f:
summary = f.read().strip()
if summary:
memories.append(
{"content": f"[summary] Conversation so far: {summary}"}
)
except (OSError, IOError):
# If we cannot read the file, return no memories silently.
pass
print(json.dumps({"type": "ingest_result", "memories": memories}))
if __name__ == "__main__":
main()