forked from HKUDS/OpenSpace
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcontext.py
More file actions
65 lines (49 loc) · 2.02 KB
/
Copy pathcontext.py
File metadata and controls
65 lines (49 loc) · 2.02 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
"""Skill-context and skill-registry helpers for GroundingAgent.
Pure side-effect functions that operate on agent instance state.
Extracted from grounding_agent.py (Epic 5.7).
"""
from __future__ import annotations
from typing import TYPE_CHECKING, List, Optional
from openspace.utils.logging import Logger
if TYPE_CHECKING:
from openspace.skill_engine import SkillRegistry
logger = Logger.get_logger("openspace.agents.grounding_agent")
def set_skill_context(
agent,
context: str,
skill_ids: Optional[List[str]] = None,
) -> None:
"""Inject skill guidance into the agent's system prompt.
Called by ``OpenSpace.execute()`` before ``process()`` when skills
are matched. The context is a formatted string built by
``SkillRegistry.build_context_injection()``.
Args:
agent: GroundingAgent instance.
context: Formatted skill content for system prompt injection.
skill_ids: skill_id values of injected skills.
"""
agent._skill_context = context if context else None
agent._active_skill_ids = skill_ids or []
if agent._skill_context:
logger.info(
f"Skill context set: {', '.join(agent._active_skill_ids) or '(unnamed)'}"
)
def clear_skill_context(agent) -> None:
"""Remove skill guidance (used before fallback execution)."""
if agent._skill_context:
logger.info(
f"Skill context cleared (was: {', '.join(agent._active_skill_ids)})"
)
agent._skill_context = None
agent._active_skill_ids = []
def has_skill_context(agent) -> bool:
"""Return True if skill context is currently set."""
return agent._skill_context is not None
def set_skill_registry(agent, registry: Optional["SkillRegistry"]) -> None:
"""Attach a SkillRegistry so the agent can offer ``retrieve_skill`` as a tool."""
agent._skill_registry = registry
if registry:
count = len(registry.list_skills())
logger.info(
f"Skill registry attached ({count} skill(s) available for mid-iteration retrieval)"
)