Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion examples/basic.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,9 @@
""",
prefixes={"ex": "http://example.org/knowledge-mapper/basic#"},
)
def example_answer_ki():
def example_answer_ki(binding_set, info):
logger.info("Handling a call to the example answer KI.")
return binding_set


if __name__ == "__main__":
Expand Down
84 changes: 84 additions & 0 deletions examples/custom-settings/custom_settings.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
"""Example: combining application config with KnowledgeBaseSettings.

A good pattern is to subclass KnowledgeBaseSettings to add application-specific
settings, while still supporting the standard config sources (YAML/JSON file, env vars,
CLI args) for the KB configuration. Other setups are possible as well.
This example shows this method and how to build a KB, register KI's from settings for
each type of interaction.

Configuration is loaded automatically from (highest priority first):
1. CLI arguments --kb_id, --db_host, ...
2. Environment vars KB_ID, DB_HOST, ...
3. config.yaml / config.json
4. Field default values

Run:
python custom_settings.py # use config.yaml / env vars
python custom_settings.py --kb_id http://my/kb # override via CLI
KB_ID=http://my/kb python custom_settings.py # override via env var
"""

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent))

from pydantic_settings import CliSettingsSource, SettingsConfigDict
from shared import get_example_logger

from src import KnowledgeBase, KnowledgeBaseSettings
from src.ke.models import BindingSet, KnowledgeInteractionInfo

EXAMPLE_NAME = "custom-settings"
logger = get_example_logger(EXAMPLE_NAME)


class AppSettings(KnowledgeBaseSettings):
model_config = SettingsConfigDict(
yaml_file="custom-settings/settings.yaml",
cli_parse_args=True,
extra="ignore",
)

# Application-specific fields
db_host: str = "localhost"
db_port: int = 5432
debug: bool = False

@classmethod
def settings_customise_sources(cls, settings_cls, **kwargs): # type: ignore
return (
CliSettingsSource(settings_cls, cli_parse_args=True),
*super().settings_customise_sources(settings_cls, **kwargs),
)


settings = AppSettings() # type: ignore
kb = KnowledgeBase.from_settings(settings)
kb.ki_from_settings_with_default_handler("ask-from-settings")
kb.ki_from_settings_with_default_handler("post-from-settings")


@kb.ki_from_settings("answer-from-settings")
def example_answer_from_settings(
binding_set: BindingSet, info: KnowledgeInteractionInfo
) -> BindingSet:
return binding_set


@kb.ki_from_settings("react-from-settings")
def example_react_from_settings(
binding_set: BindingSet, info: KnowledgeInteractionInfo
) -> BindingSet:
return binding_set


if __name__ == "__main__":
ask_ctx = kb.ki_registry["ask-from-settings"]
post_ctx = kb.ki_registry["post-from-settings"]

logger.info(f"KB id: {kb.info.id}")
logger.info(f"DB host:port: {settings.db_host}:{settings.db_port}")
logger.info(f"Debug: {settings.debug}")
logger.info(f"ASK KI name: {ask_ctx.info.name}")
logger.info(f"POST KI name: {post_ctx.info.name}")
43 changes: 43 additions & 0 deletions examples/custom-settings/settings.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
# Application-specific settings
db_host: "localhost"
db_port: 5432
debug: false

# Knowledge Base settings
knowledge_base:
id: "http://example.org/knowledge-mapper/custom-settings#kb"
name: "custom-knowledge-base"
description: "A knowledge base with custom settings loaded."

knowledge_engine_endpoint: "http://localhost:8280/rest"

knowledge_interactions:
- name: ask-from-settings
type: AskKnowledgeInteraction
prefixes:
ex: "http://example.org/knowledge-mapper/custom-settings#"
graph_pattern: |
?s ?p ?o .
- name: answer-from-settings
type: AnswerKnowledgeInteraction
prefixes:
ex: "http://example.org/knowledge-mapper/custom-settings#"
graph_pattern: |
?s ?p ?o .
- name: post-from-settings
type: PostKnowledgeInteraction
prefixes:
ex: "http://example.org/knowledge-mapper/custom-settings#"
argument_graph_pattern: |
?s ?p ?o .
result_graph_pattern: |
?s ?p ?o .
- name: react-from-settings
type: ReactKnowledgeInteraction
prefixes:
ex: "http://example.org/knowledge-mapper/custom-settings#"
argument_graph_pattern: |
?s ?p ?o .
result_graph_pattern: |
?s ?p ?o .

2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"pydantic>=2.12.5",
"pydantic-settings>=2.13.1",
"pydantic-settings[yaml]>=2.13.1",
"requests>=2.32.5",
]

Expand Down
1 change: 1 addition & 0 deletions src/__init__.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import logging

from .knowledge_base import KnowledgeBase
from .settings import KnowledgeBaseSettings

__version__ = "0.1.0a0"

Expand Down
13 changes: 11 additions & 2 deletions src/ke/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ def register_ki(
"""
...

def poll_ki_call(self, kb_id: str) -> PollResult:
def poll_ki_call(self, kb_id: str) -> tuple[PollResult, HandleRequest | None]:
"""Poll the KE runtime for an incoming KI call for the given KB.

Raises:
Expand All @@ -129,12 +129,17 @@ def poll_ki_call(self, kb_id: str) -> PollResult:
"""
...

@property
def ke_url(self) -> str:
"""Return the base URL of the KE runtime this client is communicating with."""
...


class Client(ClientProtocol):
"""HTTP client for the Knowledge Engine REST API."""

def __init__(self, ke_url: str):
self.ke_url = ke_url
self._ke_url = ke_url

def ke_is_available(self) -> bool:
try:
Expand Down Expand Up @@ -282,3 +287,7 @@ def post_handle_response(

if not response.ok:
raise UnexpectedHttpResponseError(response)

@property
def ke_url(self) -> str:
return self._ke_url
15 changes: 8 additions & 7 deletions src/ke/models.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from enum import StrEnum
from typing import Annotated

from pydantic import BaseModel, ConfigDict, Field
from pydantic.alias_generators import to_camel
Expand All @@ -13,9 +14,9 @@ class BindingModel(BaseModel):
class KnowledgeBaseInfo(BaseModel):
model_config = ConfigDict(extra="allow", frozen=True, populate_by_name=True)

id: str = Field(..., alias="knowledgeBaseId")
name: str = Field(..., alias="knowledgeBaseName")
description: str = Field(..., alias="knowledgeBaseDescription")
id: Annotated[str, Field(..., alias="knowledgeBaseId")]
name: Annotated[str, Field(..., alias="knowledgeBaseName")]
description: Annotated[str, Field(..., alias="knowledgeBaseDescription")]


class KiTypes(StrEnum):
Expand All @@ -30,10 +31,10 @@ class KnowledgeInteractionInfo(BaseModel):
alias_generator=to_camel, extra="allow", frozen=True, populate_by_name=True
)

type: KiTypes = Field(..., alias="knowledgeInteractionType")
id: str | None = Field(default=None, alias="knowledgeInteractionId")
name: str = Field(..., alias="knowledgeInteractionName")
prefixes: dict[str, str] = Field(default_factory=dict)
type: Annotated[KiTypes, Field(..., alias="knowledgeInteractionType")]
id: Annotated[str | None, Field(..., alias="knowledgeInteractionId")] = None
name: Annotated[str, Field(..., alias="knowledgeInteractionName")]
prefixes: Annotated[dict[str, str], Field(default_factory=dict)]


class AskAnswerInteractionInfo(KnowledgeInteractionInfo):
Expand Down
Loading
Loading