mirror of
https://github.com/imartinez/privateGPT.git
synced 2026-07-17 01:48:03 +00:00
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
"""Implement context stack builder."""
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from private_gpt.components.chat.models.chat_config_models import ToolSpec
|
|
from private_gpt.components.context.errors import (
|
|
ToolNameConflictError,
|
|
)
|
|
from private_gpt.components.context.models.context_layer import (
|
|
AnyContextLayer,
|
|
DocumentLayer,
|
|
SkillBodyLayer,
|
|
SkillCatalogEntry,
|
|
SkillCatalogLayer,
|
|
ToolDefinitionsLayer,
|
|
UserInstructionsLayer,
|
|
)
|
|
from private_gpt.components.context.models.context_stack import ContextStack
|
|
from private_gpt.components.context.models.layer_type import LayerType
|
|
from private_gpt.components.engines.citations.types import Document
|
|
|
|
|
|
class ContextStackBuilder(BaseModel):
|
|
"""Build validated context stacks with tool conflict checks."""
|
|
|
|
user_instructions: str | None = None
|
|
layers: list[AnyContextLayer] = Field(default_factory=list)
|
|
|
|
def set_user_instructions(self, text: str) -> "ContextStackBuilder":
|
|
"""Set user instructions and append the user-instructions layer."""
|
|
self.user_instructions = text
|
|
self.layers.append(
|
|
UserInstructionsLayer(
|
|
text=text,
|
|
source="platform",
|
|
)
|
|
)
|
|
return self
|
|
|
|
def add_skill_catalog(
|
|
self, skills: list[SkillCatalogEntry]
|
|
) -> "ContextStackBuilder":
|
|
"""Add the skill catalog layer from names and descriptions."""
|
|
self.layers.append(SkillCatalogLayer(entries=skills, source="catalog"))
|
|
return self
|
|
|
|
def add_activated_skill(
|
|
self, skill_id: str, name: str, version: str, instructions: str
|
|
) -> "ContextStackBuilder":
|
|
"""Add one activated skill body layer."""
|
|
self.layers.append(
|
|
SkillBodyLayer(
|
|
skill_id=skill_id,
|
|
name=name,
|
|
version=version,
|
|
instructions=instructions,
|
|
source=name,
|
|
)
|
|
)
|
|
return self
|
|
|
|
def add_tools(
|
|
self,
|
|
tools: list[ToolSpec],
|
|
source: str,
|
|
) -> "ContextStackBuilder":
|
|
"""Add a tool definition layer after checking for name conflicts."""
|
|
self.layers.append(ToolDefinitionsLayer(tools=tools, source=source))
|
|
return self
|
|
|
|
def add_document(
|
|
self,
|
|
document: Document,
|
|
source: str,
|
|
) -> "ContextStackBuilder":
|
|
"""Add a document layer after checking document budget."""
|
|
self.layers.append(DocumentLayer(document=document, source=source))
|
|
return self
|
|
|
|
def build(self) -> ContextStack:
|
|
"""Build an immutable stack after conflict checks."""
|
|
self.validate_tool_name_uniqueness()
|
|
return ContextStack(layers=list(self.layers))
|
|
|
|
def validate_tool_name_uniqueness(self) -> None:
|
|
"""Validate global tool uniqueness across tool-definition layers."""
|
|
seen: dict[str, str] = {}
|
|
for layer in self.layers:
|
|
if layer.type is not LayerType.TOOL_DEFINITIONS:
|
|
continue
|
|
if not isinstance(layer, ToolDefinitionsLayer):
|
|
continue
|
|
for tool in layer.tools:
|
|
if tool.name is None:
|
|
continue
|
|
existing = seen.get(tool.name)
|
|
if existing is not None:
|
|
raise ToolNameConflictError(
|
|
"Tool name conflict: "
|
|
f"name='{tool.name}', "
|
|
f"layer_a='{existing}', layer_b='{layer.source}'"
|
|
)
|
|
seen[tool.name] = layer.source
|