knowledge_config.py 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. """Knowledge extraction and injection configuration.
  2. This value object belongs to the runner configuration layer. Keeping it out
  3. of the builtin HTTP tool module prevents core configuration imports from
  4. triggering builtin tool registration.
  5. """
  6. from __future__ import annotations
  7. import subprocess
  8. from dataclasses import dataclass
  9. from typing import Dict, List, Optional
  10. from agent.core.prompts import COMPLETION_REFLECT_PROMPT, build_reflect_prompt
  11. @dataclass
  12. class KnowledgeConfig:
  13. """Configure knowledge extraction, review, and prompt injection."""
  14. enable_extraction: bool = True
  15. reflect_prompt: str = ""
  16. enable_completion_extraction: bool = True
  17. completion_reflect_prompt: str = ""
  18. reflect_auto_commit: bool = False
  19. enable_injection: bool = True
  20. owner: str = ""
  21. default_tags: Optional[Dict[str, str]] = None
  22. default_scopes: Optional[List[str]] = None
  23. default_search_types: Optional[List[str]] = None
  24. default_search_owner: str = ""
  25. def get_reflect_prompt(self) -> str:
  26. """Return the configured compression-reflection prompt."""
  27. return self.reflect_prompt or build_reflect_prompt()
  28. def get_completion_reflect_prompt(self) -> str:
  29. """Return the configured completion-reflection prompt."""
  30. return self.completion_reflect_prompt or COMPLETION_REFLECT_PROMPT
  31. @property
  32. def resolved_owner(self) -> str:
  33. """Resolve owner from explicit config, Git identity, then ``agent``."""
  34. if self.owner:
  35. return self.owner
  36. try:
  37. result = subprocess.run(
  38. ["git", "config", "user.email"],
  39. capture_output=True,
  40. text=True,
  41. timeout=2,
  42. )
  43. if result.returncode == 0 and result.stdout.strip():
  44. return result.stdout.strip()
  45. except (OSError, subprocess.SubprocessError):
  46. pass
  47. return "agent"
  48. def get_owner(self, agent_id: str = "agent") -> str:
  49. """Return the resolved owner, namespaced for non-default agents."""
  50. owner = self.resolved_owner
  51. if owner == "agent" and agent_id != "agent":
  52. return f"agent:{agent_id}"
  53. return owner
  54. __all__ = ["KnowledgeConfig"]