feat(phase3): add The Librarian agent with library-desk integration

Library-Desk API Client:
- Async HTTP client with httpx for library-desk API
- HybridRAG search (vector + graph + web)
- Wiki operations (search, get, list, create, update)
- Smart page creation with HybridRAG research
- Semantic vector search and knowledge graph queries
- Dossier browsing and health checks

Librarian Tools (11 total):
- Research: hybrid_search, search_wiki, get_wiki_page, semantic_search
- Browse: list_dossiers, get_dossier_pages, explore_knowledge_graph
- Graph: find_related_entities
- Write: create_wiki_page, update_wiki_page, smart_create_wiki_page

Agent:
- PydanticAI agent with research assistant personality
- System prompt with research and writing workflows
- Streaming support via run_librarian_stream()

Capability:
- LIBRARIAN_CAPABILITY definition for Household Registry
- Automatic registration on startup

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-11 21:27:09 +01:00
co-authored by Claude Opus 4.5
parent 92c0d5d770
commit f6f37b341b
5 changed files with 1783 additions and 0 deletions
+30
View File
@@ -0,0 +1,30 @@
"""
The Librarian - Expert agent for research and knowledge management.
Connects to the library-desk API to provide:
- HybridRAG search (vector + graph + web)
- Wiki.js operations
- Knowledge graph queries
- Semantic search
"""
from src.agents.librarian.agent import (
get_librarian_agent,
run_librarian,
run_librarian_stream,
)
from src.agents.librarian.capability import (
LIBRARIAN_CAPABILITY,
get_librarian_capability,
register_librarian,
unregister_librarian,
)
__all__ = [
"LIBRARIAN_CAPABILITY",
"get_librarian_capability",
"get_librarian_agent",
"register_librarian",
"unregister_librarian",
"run_librarian",
"run_librarian_stream",
]
+286
View File
@@ -0,0 +1,286 @@
"""
The Librarian - Expert agent for research and knowledge management.
A PydanticAI agent that provides research assistance through
the library-desk API, offering:
- HybridRAG search across all knowledge sources
- Wiki and document management
- Semantic search and knowledge graph exploration
"""
from typing import Any, Optional
from pydantic_ai import Agent
from src.agents.librarian.tools import (
create_wiki_page,
explore_knowledge_graph,
find_related_entities,
get_dossier_pages,
get_wiki_page,
hybrid_search,
list_dossiers,
search_wiki,
semantic_search,
smart_create_wiki_page,
update_wiki_page,
)
from src.core.config import config
from src.core.logging_config import get_logger
logger = get_logger(__name__)
# Librarian system prompt
LIBRARIAN_SYSTEM_PROMPT = """You are The Librarian, an expert research assistant in the Tatlock household.
Your role is to help users find, understand, synthesize, and manage information from:
- The personal wiki (Wiki.js) containing documentation and notes
- The knowledge graph (Neo4j) with entities and relationships
- Vector embeddings (Qdrant) for semantic search
- Web search (SearXNG) for current information
## Your Personality
- Scholarly and thorough in your research
- Cite your sources and provide context
- Organize information clearly
- Suggest related topics when relevant
- Acknowledge limitations when information is incomplete
## Your Tools
### Research Tools
- **hybrid_search**: Your primary research tool - searches all sources at once
- **search_wiki**: Find specific wiki pages by keyword
- **semantic_search**: Find conceptually similar content
- **explore_knowledge_graph** / **find_related_entities**: Discover connections
- **list_dossiers** / **get_dossier_pages**: Browse knowledge collections
### Wiki Reading Tools
- **get_wiki_page**: Read full content of a wiki page by ID
- ALWAYS use this to fetch and read page content when summarizing
- Use after search_wiki to get the full text of a specific page
### Wiki Writing Tools
- **smart_create_wiki_page**: Create a page with automatic research (PREFERRED)
- **This is the DEFAULT choice when user asks to create a wiki page about a topic**
- When user says "Create a page about X" or "Add X to the wiki" without providing specific content, ALWAYS use this tool
- Automatically researches the topic from wiki, graph, and web
- Synthesizes content with proper source attribution
- Creates bidirectional links in knowledge graph
- **create_wiki_page**: Create a page with user-provided content
- ONLY use when user provides specific text/content they want added verbatim
- For simple notes, reminders, or quick additions with exact content
- **update_wiki_page**: Update an existing page (partial updates)
- Use when: "Update the page about X", "Fix this info", "Add to dossier"
- First search_wiki to find the page, then get_wiki_page to read it
- Only specify fields you want to change
## Research Approach
1. Start with hybrid_search for broad queries
2. Use search_wiki for specific document lookups
3. **ALWAYS use get_wiki_page to fetch full content** before summarizing a page
4. Use semantic_search when looking for conceptually similar content
5. Explore the knowledge graph to find connections between concepts
6. Synthesize and summarize findings clearly
## Writing Approach
When asked to create or update wiki content:
1. **"Create a page about X" (no specific content provided)**: Use smart_create_wiki_page
- This is the PREFERRED tool for topic-based page creation
- It researches first and creates comprehensive, well-sourced content
2. **User provides exact text to add**: Use create_wiki_page with their content
3. **Updating existing pages**:
- Search for the page with search_wiki
- Fetch full content with get_wiki_page
- Make edits and use update_wiki_page
4. **Organizing into dossiers**: Use update_wiki_page with just the tags field
## Response Format
Your responses are returned to Tatlock (the butler) who will synthesize them into a final answer for the user. Keep this in mind:
- Lead with the key findings or confirmation of action
- Include relevant sources and citations
- When summarizing wiki pages, fetch and read them first
- Note any gaps in available information
- Be concise but thorough - Tatlock will format the final response
- Structure your findings clearly so they can be easily integrated with other responses
"""
# Lazy initialization to avoid connection issues during imports
_librarian_agent: Optional[Agent[None, str]] = None
def _create_librarian_agent() -> Agent[None, str]:
"""Create the Librarian PydanticAI agent."""
# Import required classes for Ollama configuration
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.ollama import OllamaProvider
# PydanticAI expects Ollama base URL to end with /v1
clean_host = str(config.OLLAMA_HOST).rstrip('/')
base_url = f"{clean_host}/v1"
# Create Ollama model with provider
model = OpenAIChatModel(
model_name=config.OLLAMA_DEFAULT_MODEL,
provider=OllamaProvider(base_url=base_url)
)
agent: Agent[None, str] = Agent(
model=model,
system_prompt=LIBRARIAN_SYSTEM_PROMPT,
retries=2,
)
# Register research tools
agent.tool_plain(hybrid_search)
agent.tool_plain(search_wiki)
agent.tool_plain(semantic_search)
agent.tool_plain(list_dossiers)
agent.tool_plain(get_dossier_pages)
agent.tool_plain(explore_knowledge_graph)
agent.tool_plain(find_related_entities)
# Register wiki read tools
agent.tool_plain(get_wiki_page)
# Register wiki write tools
agent.tool_plain(create_wiki_page)
agent.tool_plain(update_wiki_page)
agent.tool_plain(smart_create_wiki_page)
logger.info(
"librarian_agent_created",
model=config.OLLAMA_DEFAULT_MODEL,
tool_count=11,
)
return agent
def get_librarian_agent() -> Agent[None, str]:
"""
Get the Librarian agent instance (lazy initialization).
Returns:
PydanticAI Agent configured for research tasks
"""
global _librarian_agent
if _librarian_agent is None:
_librarian_agent = _create_librarian_agent()
return _librarian_agent
async def run_librarian(
task: str,
context: str = "",
message_history: Optional[list[Any]] = None,
) -> str:
"""
Execute a research task with The Librarian.
This is the main entry point for delegating research tasks
to The Librarian from Tatlock or other agents.
Args:
task: The research task or question
context: Additional context from conversation
message_history: Optional conversation history
Returns:
Research results and findings
Example:
result = await run_librarian(
task="Find information about Docker networking",
context="User is setting up a homelab",
)
"""
agent = get_librarian_agent()
# Build prompt with context if provided
prompt = task
if context:
prompt = f"Context: {context}\n\nTask: {task}"
logger.info(
"librarian_task_started",
task=task[:100],
has_context=bool(context),
has_history=bool(message_history),
)
try:
result = await agent.run(
prompt,
message_history=message_history,
)
logger.info(
"librarian_task_completed",
task=task[:50],
output_length=len(result.output),
)
return result.output
except Exception as e:
logger.error(
"librarian_task_error",
task=task[:50],
error=str(e),
exc_info=True,
)
return f"The Librarian encountered an error: {str(e)}"
async def run_librarian_stream(
task: str,
context: str = "",
message_history: Optional[list[Any]] = None,
):
"""
Execute a research task with streaming output.
Yields text deltas as The Librarian generates the response.
Args:
task: The research task or question
context: Additional context from conversation
message_history: Optional conversation history
Yields:
str: Text deltas from the response
Example:
async for delta in run_librarian_stream("Find Docker docs"):
print(delta, end="", flush=True)
"""
agent = get_librarian_agent()
# Build prompt with context if provided
prompt = task
if context:
prompt = f"Context: {context}\n\nTask: {task}"
logger.info(
"librarian_stream_started",
task=task[:100],
)
try:
async with agent.run_stream(
prompt,
message_history=message_history,
) as response:
async for delta in response.stream_text(delta=True):
yield delta
logger.info("librarian_stream_completed", task=task[:50])
except Exception as e:
logger.error(
"librarian_stream_error",
task=task[:50],
error=str(e),
exc_info=True,
)
yield f"\n\nThe Librarian encountered an error: {str(e)}"
+81
View File
@@ -0,0 +1,81 @@
"""
Librarian capability registration for the Household Registry.
Defines The Librarian's capabilities and registers it as a
household member for coordination by the Steward and Tatlock.
"""
from src.agents.librarian.agent import get_librarian_agent
from src.agents.librarian.tools import LIBRARIAN_TOOLS
from src.core.household_registry import (
HouseholdCapability,
get_household_registry,
)
from src.core.logging_config import get_logger
logger = get_logger(__name__)
# The Librarian's capability summary for Steward coordination
LIBRARIAN_CAPABILITY = HouseholdCapability(
name="librarian",
role="The Librarian",
category="research",
description=(
"Research assistant providing knowledge search, wiki access, "
"semantic search, and knowledge graph exploration via library-desk API"
),
domains=[
"research",
"knowledge",
"information",
"wiki",
"documents",
"search",
"synthesis",
],
cost="medium", # Multiple API calls to library-desk
requires_network=True, # Needs library-desk API access
)
def get_librarian_capability() -> HouseholdCapability:
"""Get The Librarian's capability definition."""
return LIBRARIAN_CAPABILITY
def register_librarian() -> None:
"""
Register The Librarian with the Household Registry.
This makes The Librarian available for:
- Steward recommendations (via capability summary)
- Tatlock delegation (via agent reference)
- Tool scoping (via tool list)
"""
registry = get_household_registry()
# Check if already registered
if "librarian" in registry:
logger.debug("librarian_already_registered")
return
registry.register(
name="librarian",
capability=LIBRARIAN_CAPABILITY,
tools=LIBRARIAN_TOOLS,
agent=get_librarian_agent(),
)
logger.info(
"librarian_registered",
role=LIBRARIAN_CAPABILITY.role,
domains=LIBRARIAN_CAPABILITY.domains,
tool_count=len(LIBRARIAN_TOOLS),
)
def unregister_librarian() -> None:
"""Unregister The Librarian from the Household Registry."""
registry = get_household_registry()
registry.unregister("librarian")
logger.info("librarian_unregistered")
+685
View File
@@ -0,0 +1,685 @@
"""
HTTP client for the Library-Desk API.
Provides async methods for all relevant library-desk endpoints:
- HybridRAG queries
- Wiki operations
- Vector search
- Knowledge graph queries
"""
from typing import Any, Optional
import httpx
from pydantic import BaseModel, Field
from src.core.config import config
from src.core.logging_config import get_logger
logger = get_logger(__name__)
# ============================================================================
# Response Models
# ============================================================================
class WikiPage(BaseModel):
"""Wiki page from library-desk."""
id: int
path: str
title: str
description: Optional[str] = None
content: Optional[str] = None
tags: list[str] = Field(default_factory=list)
created_at: Optional[str] = None
updated_at: Optional[str] = None
class WikiSearchResult(BaseModel):
"""Search result from wiki search."""
id: int
path: str
title: str
description: Optional[str] = None
locale: Optional[str] = None
class VectorSearchResult(BaseModel):
"""Result from semantic vector search."""
page_id: int
page_path: str
page_title: str
chunk_text: str
score: float
chunk_index: int
class HybridSearchResult(BaseModel):
"""Result from HybridRAG search."""
source: str # "vector", "graph", "web"
title: str
content: str
url: Optional[str] = None
score: float
page_id: Optional[int] = None
metadata: dict[str, Any] = Field(default_factory=dict)
class HybridRAGResponse(BaseModel):
"""Full response from HybridRAG query."""
results: list[HybridSearchResult] = Field(default_factory=list)
keywords: list[str] = Field(default_factory=list)
synonyms: list[str] = Field(default_factory=list)
related_dossiers: list[str] = Field(default_factory=list)
formatted_context: str = ""
search_id: Optional[str] = None
timing: dict[str, float] = Field(default_factory=dict)
class GraphNode(BaseModel):
"""Node from knowledge graph."""
id: str
labels: list[str] = Field(default_factory=list)
properties: dict[str, Any] = Field(default_factory=dict)
class Dossier(BaseModel):
"""A dossier (tag-based collection)."""
name: str
page_count: int
class ResearchSummary(BaseModel):
"""Summary of research performed during smart-create."""
wiki_results: int = 0
web_results: int = 0
graph_entities: int = 0
keywords_extracted: int = 0
timing_ms: int = 0
class EntityLinking(BaseModel):
"""Entity linking results from smart-create."""
forward_links: int = 0
backward_links: int = 0
pages_updated: int = 0
class SmartCreateResponse(BaseModel):
"""Response from smart-create wiki page endpoint."""
page: WikiPage
research_summary: ResearchSummary = Field(default_factory=ResearchSummary)
sources_used: int = 0
search_id: Optional[str] = None
entity_linking: EntityLinking = Field(default_factory=EntityLinking)
# ============================================================================
# Client
# ============================================================================
class LibraryDeskClient:
"""
Async HTTP client for Library-Desk API.
Usage:
async with LibraryDeskClient() as client:
results = await client.hybrid_search("docker kubernetes")
"""
def __init__(
self,
base_url: Optional[str] = None,
api_key: Optional[str] = None,
timeout: int = 60,
):
"""
Initialize the client.
Args:
base_url: Library-desk API URL (defaults to config)
api_key: API key for authentication (defaults to config)
timeout: Request timeout in seconds
"""
self.base_url = base_url or str(config.LIBRARY_DESK_HOST)
self.api_key = api_key or config.LIBRARY_DESK_API_KEY
self.timeout = timeout
self._client: Optional[httpx.AsyncClient] = None
async def __aenter__(self) -> "LibraryDeskClient":
"""Create HTTP client on context entry."""
headers = {}
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
self._client = httpx.AsyncClient(
base_url=self.base_url,
headers=headers,
timeout=self.timeout,
)
return self
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
"""Close HTTP client on context exit."""
if self._client:
await self._client.aclose()
self._client = None
def _ensure_client(self) -> httpx.AsyncClient:
"""Ensure client is initialized."""
if self._client is None:
raise RuntimeError(
"Client not initialized. Use 'async with LibraryDeskClient() as client:'"
)
return self._client
# ========================================================================
# HybridRAG
# ========================================================================
async def hybrid_search(
self,
query: str,
user: str = "jpmschweitzer",
vector_limit: int = 10,
graph_limit: int = 10,
web_limit: int = 5,
enable_reranking: bool = True,
final_result_count: int = 10,
) -> HybridRAGResponse:
"""
Execute HybridRAG search combining vector, graph, and web results.
Args:
query: Search query
user: User identifier for multi-tenancy
vector_limit: Max results from vector search
graph_limit: Max results from graph search
web_limit: Max results from web search
enable_reranking: Whether to rerank with LLM
final_result_count: Number of final results after fusion
Returns:
HybridRAGResponse with ranked results and context
"""
client = self._ensure_client()
payload = {
"query": query,
"config": {
"vector_limit": vector_limit,
"graph_limit": graph_limit,
"web_limit": web_limit,
"enable_reranking": enable_reranking,
"final_result_count": final_result_count,
},
}
logger.info("library_desk_hybrid_search", query=query, user=user)
response = await client.post(
"/query/hybrid",
json=payload,
params={"user": user},
)
response.raise_for_status()
data = response.json()
# Parse results
results = []
for r in data.get("results", []):
results.append(HybridSearchResult(
source=r.get("source", "unknown"),
title=r.get("title", ""),
content=r.get("content", ""),
url=r.get("url"),
score=r.get("score", 0.0),
page_id=r.get("page_id"),
metadata=r.get("metadata", {}),
))
return HybridRAGResponse(
results=results,
keywords=data.get("keywords", []),
synonyms=data.get("synonyms", []),
related_dossiers=data.get("related_dossiers", []),
formatted_context=data.get("formatted_context", ""),
search_id=data.get("search_id"),
timing=data.get("timing", {}),
)
# ========================================================================
# Wiki Operations
# ========================================================================
async def search_wiki(
self,
query: str,
user: str = "jpmschweitzer",
limit: int = 20,
) -> list[WikiSearchResult]:
"""
Search wiki pages by text.
Args:
query: Search query
user: User identifier
limit: Maximum results
Returns:
List of matching wiki pages
"""
client = self._ensure_client()
logger.debug("library_desk_wiki_search", query=query, user=user)
response = await client.get(
"/wiki/search",
params={"q": query, "user": user, "limit": limit},
)
response.raise_for_status()
data = response.json()
return [WikiSearchResult(**r) for r in data.get("results", [])]
async def get_wiki_page(
self,
page_id: int,
user: str = "jpmschweitzer",
) -> WikiPage:
"""
Get a wiki page by ID.
Args:
page_id: Page ID
user: User identifier
Returns:
WikiPage with full content
"""
client = self._ensure_client()
response = await client.get(
f"/wiki/pages/{page_id}",
params={"user": user},
)
response.raise_for_status()
return WikiPage(**response.json())
async def list_wiki_pages(
self,
user: str = "jpmschweitzer",
tag: Optional[str] = None,
limit: int = 50,
) -> list[WikiPage]:
"""
List wiki pages, optionally filtered by tag.
Args:
user: User identifier
tag: Optional tag (dossier) to filter by
limit: Maximum pages to return
Returns:
List of wiki pages
"""
client = self._ensure_client()
params: dict[str, Any] = {"user": user, "limit": limit}
if tag:
params["tag"] = tag
response = await client.get("/wiki/pages", params=params)
response.raise_for_status()
data = response.json()
return [WikiPage(**p) for p in data.get("pages", [])]
async def create_wiki_page(
self,
title: str,
path: str,
content: str,
user: str = "jpmschweitzer",
description: str = "",
tags: Optional[list[str]] = None,
) -> WikiPage:
"""
Create a new wiki page.
Args:
title: Page title
path: Page path (e.g., "/projects/my-project")
content: Markdown content
user: User identifier
description: Short description
tags: List of tags (dossiers)
Returns:
Created WikiPage
"""
client = self._ensure_client()
payload = {
"title": title,
"path": path,
"content": content,
"user": user,
"description": description,
"tags": tags or [],
}
logger.info("library_desk_create_page", title=title, path=path)
response = await client.post("/wiki/pages", json=payload)
response.raise_for_status()
return WikiPage(**response.json())
async def update_wiki_page(
self,
page_id: int,
user: str = "jpmschweitzer",
content: Optional[str] = None,
title: Optional[str] = None,
tags: Optional[list[str]] = None,
description: Optional[str] = None,
) -> WikiPage:
"""
Update an existing wiki page.
Supports partial updates - only provided fields are updated.
Automatically triggers vector re-indexing and graph extraction.
Args:
page_id: ID of the page to update
user: User identifier
content: New content (optional)
title: New title (optional)
tags: New tags list (optional)
description: New description (optional)
Returns:
Updated WikiPage
"""
client = self._ensure_client()
# Build update payload with only provided fields
update_data: dict[str, Any] = {}
if content is not None:
update_data["content"] = content
if title is not None:
update_data["title"] = title
if tags is not None:
update_data["tags"] = tags
if description is not None:
update_data["description"] = description
logger.info(
"library_desk_update_page",
page_id=page_id,
fields=list(update_data.keys()),
)
response = await client.put(
f"/wiki/pages/{page_id}",
params={"user": user},
json=update_data,
)
response.raise_for_status()
return WikiPage(**response.json())
async def smart_create_wiki_page(
self,
topic: str,
tags: list[str],
user: str = "jpmschweitzer",
path: Optional[str] = None,
include_web_research: bool = True,
include_wiki_search: bool = True,
) -> SmartCreateResponse:
"""
Create a wiki page with HybridRAG research.
This endpoint:
1. Searches existing wiki, knowledge graph, and web for context
2. Uses LLM to synthesize findings into structured content
3. Creates the page with proper attribution
4. Automatically links entities bidirectionally
Args:
topic: The topic to research and create a page about
tags: List of tags (dossiers) for the page
user: User identifier
path: Optional custom path (auto-generated from topic if not provided)
include_web_research: Whether to include web search results
include_wiki_search: Whether to include existing wiki content
Returns:
SmartCreateResponse with page and research metadata
"""
client = self._ensure_client()
payload: dict[str, Any] = {
"topic": topic,
"tags": tags,
"user": user,
"include_web_research": include_web_research,
"include_wiki_search": include_wiki_search,
}
if path is not None:
payload["path"] = path
logger.info(
"library_desk_smart_create",
topic=topic,
tags=tags,
include_web=include_web_research,
)
response = await client.post("/wiki/pages/smart-create", json=payload)
response.raise_for_status()
data = response.json()
# Parse nested response
page = WikiPage(**data.get("page", {}))
research_summary = ResearchSummary(**data.get("research_summary", {}))
entity_linking = EntityLinking(**data.get("entity_linking", {}))
return SmartCreateResponse(
page=page,
research_summary=research_summary,
sources_used=data.get("sources_used", 0),
search_id=data.get("search_id"),
entity_linking=entity_linking,
)
async def list_dossiers(
self,
user: str = "jpmschweitzer",
) -> list[Dossier]:
"""
List all dossiers (tag collections) for a user.
Args:
user: User identifier
Returns:
List of dossiers with page counts
"""
client = self._ensure_client()
response = await client.get(
"/wiki/dossiers",
params={"user": user},
)
response.raise_for_status()
data = response.json()
return [Dossier(**d) for d in data.get("dossiers", [])]
# ========================================================================
# Vector Search
# ========================================================================
async def semantic_search(
self,
query: str,
user: str = "jpmschweitzer",
limit: int = 10,
score_threshold: float = 0.5,
) -> list[VectorSearchResult]:
"""
Perform semantic (vector) search over documents.
Args:
query: Natural language query
user: User identifier
limit: Maximum results
score_threshold: Minimum similarity score
Returns:
List of matching document chunks with scores
"""
client = self._ensure_client()
payload = {
"query": query,
"user": user,
"limit": limit,
"score_threshold": score_threshold,
}
logger.debug("library_desk_semantic_search", query=query)
response = await client.post("/vector/search", json=payload)
response.raise_for_status()
data = response.json()
return [VectorSearchResult(**r) for r in data.get("results", [])]
# ========================================================================
# Knowledge Graph
# ========================================================================
async def query_graph(
self,
cypher_query: str,
user: str = "jpmschweitzer",
parameters: Optional[dict[str, Any]] = None,
) -> list[dict[str, Any]]:
"""
Execute a Cypher query on the knowledge graph.
Note: Query is automatically scoped to user's data.
Args:
cypher_query: Cypher query string
user: User identifier
parameters: Query parameters
Returns:
List of result records
"""
client = self._ensure_client()
payload = {
"query": cypher_query,
"user": user,
"parameters": parameters or {},
}
logger.debug("library_desk_graph_query", query=cypher_query[:100])
response = await client.post("/graph/query", json=payload)
response.raise_for_status()
return response.json().get("records", [])
async def list_graph_nodes(
self,
user: str = "jpmschweitzer",
node_type: Optional[str] = None,
limit: int = 100,
) -> list[GraphNode]:
"""
List nodes in the knowledge graph.
Args:
user: User identifier
node_type: Optional filter by type (Document, Person, Concept, etc.)
limit: Maximum nodes
Returns:
List of graph nodes
"""
client = self._ensure_client()
params: dict[str, Any] = {"user": user, "limit": limit}
if node_type:
params["node_type"] = node_type
response = await client.get("/graph/nodes", params=params)
response.raise_for_status()
data = response.json()
return [GraphNode(**n) for n in data.get("nodes", [])]
async def get_graph_node(
self,
node_id: str,
user: str = "jpmschweitzer",
) -> dict[str, Any]:
"""
Get detailed information about a graph node.
Args:
node_id: Node ID
user: User identifier
Returns:
Node with relationships and connected nodes
"""
client = self._ensure_client()
response = await client.get(
f"/graph/nodes/{node_id}",
params={"user": user},
)
response.raise_for_status()
return response.json()
# ========================================================================
# Health Check
# ========================================================================
async def health_check(self) -> bool:
"""
Check if library-desk is healthy.
Returns:
True if healthy, False otherwise
"""
try:
client = self._ensure_client()
response = await client.get("/health")
return response.status_code == 200
except Exception as e:
logger.warning("library_desk_health_check_failed", error=str(e))
return False
# Global client factory
async def get_library_client() -> LibraryDeskClient:
"""
Get a library-desk client instance.
Usage:
async with get_library_client() as client:
results = await client.hybrid_search("query")
"""
return LibraryDeskClient()
+701
View File
@@ -0,0 +1,701 @@
"""
Librarian tools for PydanticAI agent.
These tools wrap the library-desk API and are registered with
The Librarian agent for research and knowledge management tasks.
"""
from src.agents.librarian.client import LibraryDeskClient
from src.core.logging_config import get_logger
logger = get_logger(__name__)
# ============================================================================
# HybridRAG Search
# ============================================================================
async def hybrid_search(
query: str,
include_web: bool = True,
) -> str:
"""
Search across all knowledge sources using HybridRAG.
This is the primary research tool, combining:
- Vector search (semantic similarity over documents)
- Knowledge graph (entities and relationships)
- Web search (current information from SearXNG)
Results are fused and re-ranked by relevance.
Args:
query: Natural language research query
include_web: Whether to include web results (default: True)
Returns:
Formatted search results with sources and context
Examples:
hybrid_search("How does Docker orchestration work with Kubernetes?")
hybrid_search("What projects use Neo4j?", include_web=False)
"""
try:
async with LibraryDeskClient() as client:
response = await client.hybrid_search(
query=query,
web_limit=5 if include_web else 0,
)
if not response.results:
return f"No results found for '{query}'"
# Format results
output_parts = [f"## Search Results for: {query}\n"]
# Add keywords if extracted
if response.keywords:
output_parts.append(f"**Keywords:** {', '.join(response.keywords)}")
# Add related dossiers
if response.related_dossiers:
output_parts.append(
f"**Related Dossiers:** {', '.join(response.related_dossiers)}"
)
output_parts.append("")
# Add results
for i, result in enumerate(response.results, 1):
source_icon = {
"vector": "📄",
"graph": "🔗",
"web": "🌐",
}.get(result.source, "")
output_parts.append(
f"{i}. {source_icon} **{result.title}** (score: {result.score:.2f})"
)
if result.url:
output_parts.append(f" URL: {result.url}")
output_parts.append(f" {result.content[:300]}...")
output_parts.append("")
logger.info(
"librarian_hybrid_search",
query=query,
result_count=len(response.results),
)
return "\n".join(output_parts)
except Exception as e:
logger.error("librarian_hybrid_search_error", error=str(e), query=query)
return f"Error searching: {str(e)}"
# ============================================================================
# Wiki Operations
# ============================================================================
async def search_wiki(
query: str,
limit: int = 10,
) -> str:
"""
Search the personal wiki for relevant pages.
Performs full-text search over wiki page titles, descriptions,
and content. Use this for finding specific documents.
Args:
query: Search query
limit: Maximum results (default: 10)
Returns:
List of matching wiki pages with paths and descriptions
Examples:
search_wiki("docker setup guide")
search_wiki("architecture", limit=5)
"""
try:
async with LibraryDeskClient() as client:
results = await client.search_wiki(query=query, limit=limit)
if not results:
return f"No wiki pages found for '{query}'"
output_parts = [f"## Wiki Search: {query}\n"]
for i, page in enumerate(results, 1):
output_parts.append(f"{i}. **{page.title}**")
output_parts.append(f" Path: {page.path}")
if page.description:
output_parts.append(f" {page.description}")
output_parts.append("")
return "\n".join(output_parts)
except Exception as e:
logger.error("librarian_wiki_search_error", error=str(e))
return f"Error searching wiki: {str(e)}"
async def get_wiki_page(
page_id: int,
) -> str:
"""
Get the full content of a wiki page.
Use this after searching to read the complete content
of a specific page.
Args:
page_id: The page ID from search results
Returns:
Full page content including title, path, and markdown content
Examples:
get_wiki_page(42)
"""
try:
async with LibraryDeskClient() as client:
page = await client.get_wiki_page(page_id=page_id)
output_parts = [
f"# {page.title}",
f"**Path:** {page.path}",
]
if page.description:
output_parts.append(f"**Description:** {page.description}")
if page.tags:
output_parts.append(f"**Tags:** {', '.join(page.tags)}")
output_parts.append("")
output_parts.append(page.content or "(No content)")
return "\n".join(output_parts)
except Exception as e:
logger.error("librarian_get_page_error", error=str(e), page_id=page_id)
return f"Error getting page {page_id}: {str(e)}"
async def list_dossiers() -> str:
"""
List all research dossiers (tag collections).
Dossiers are collections of wiki pages grouped by tag.
Use this to discover what knowledge collections exist.
Returns:
List of dossiers with page counts
Examples:
list_dossiers()
"""
try:
async with LibraryDeskClient() as client:
dossiers = await client.list_dossiers()
if not dossiers:
return "No dossiers found"
output_parts = ["## Research Dossiers\n"]
for dossier in dossiers:
output_parts.append(
f"- **{dossier.name}** ({dossier.page_count} pages)"
)
return "\n".join(output_parts)
except Exception as e:
logger.error("librarian_list_dossiers_error", error=str(e))
return f"Error listing dossiers: {str(e)}"
async def get_dossier_pages(
dossier_name: str,
limit: int = 20,
) -> str:
"""
Get all pages in a dossier.
Retrieves pages tagged with the specified dossier name.
Args:
dossier_name: Name of the dossier/tag
limit: Maximum pages to return
Returns:
List of pages in the dossier
Examples:
get_dossier_pages("projects")
get_dossier_pages("architecture", limit=10)
"""
try:
async with LibraryDeskClient() as client:
pages = await client.list_wiki_pages(tag=dossier_name, limit=limit)
if not pages:
return f"No pages found in dossier '{dossier_name}'"
output_parts = [f"## Dossier: {dossier_name}\n"]
for page in pages:
output_parts.append(f"- **{page.title}** ({page.path})")
if page.description:
output_parts.append(f" {page.description}")
return "\n".join(output_parts)
except Exception as e:
logger.error("librarian_get_dossier_error", error=str(e))
return f"Error getting dossier: {str(e)}"
# ============================================================================
# Semantic Search
# ============================================================================
async def semantic_search(
query: str,
limit: int = 10,
) -> str:
"""
Perform semantic (vector) search over documents.
Finds documents similar in meaning to the query,
even if they don't contain the exact words.
Args:
query: Natural language query
limit: Maximum results
Returns:
Matching document chunks with similarity scores
Examples:
semantic_search("containerization best practices")
semantic_search("how to handle authentication")
"""
try:
async with LibraryDeskClient() as client:
results = await client.semantic_search(query=query, limit=limit)
if not results:
return f"No semantically similar content found for '{query}'"
output_parts = [f"## Semantic Search: {query}\n"]
for i, result in enumerate(results, 1):
output_parts.append(
f"{i}. **{result.page_title}** (score: {result.score:.2f})"
)
output_parts.append(f" Path: {result.page_path}")
output_parts.append(f" {result.chunk_text[:200]}...")
output_parts.append("")
return "\n".join(output_parts)
except Exception as e:
logger.error("librarian_semantic_search_error", error=str(e))
return f"Error in semantic search: {str(e)}"
# ============================================================================
# Knowledge Graph
# ============================================================================
async def explore_knowledge_graph(
entity_type: str = "Document",
limit: int = 20,
) -> str:
"""
Explore entities in the knowledge graph.
Lists nodes of a specific type to understand what's
in the knowledge base.
Args:
entity_type: Type of entity (Document, Person, Project, Concept, Technology)
limit: Maximum nodes to return
Returns:
List of entities with their properties
Examples:
explore_knowledge_graph("Person")
explore_knowledge_graph("Technology", limit=50)
"""
try:
async with LibraryDeskClient() as client:
nodes = await client.list_graph_nodes(
node_type=entity_type,
limit=limit,
)
if not nodes:
return f"No {entity_type} nodes found in knowledge graph"
output_parts = [f"## Knowledge Graph: {entity_type} Entities\n"]
for node in nodes:
name = node.properties.get("name", node.properties.get("title", node.id))
output_parts.append(f"- **{name}**")
# Show a few key properties
for key in ["description", "url", "path"]:
if key in node.properties:
output_parts.append(f" {key}: {node.properties[key]}")
return "\n".join(output_parts)
except Exception as e:
logger.error("librarian_explore_graph_error", error=str(e))
return f"Error exploring knowledge graph: {str(e)}"
async def find_related_entities(
entity_name: str,
) -> str:
"""
Find entities related to a given concept or entity.
Queries the knowledge graph to find documents, people,
and concepts connected to the specified entity.
Args:
entity_name: Name of the entity to find relationships for
Returns:
Related entities and their relationships
Examples:
find_related_entities("Docker")
find_related_entities("Kubernetes")
"""
try:
async with LibraryDeskClient() as client:
# Find entities mentioning or related to the search term
cypher = """
MATCH (n)
WHERE toLower(n.name) CONTAINS toLower($name)
OR toLower(n.title) CONTAINS toLower($name)
OPTIONAL MATCH (n)-[r]-(related)
RETURN n, collect(DISTINCT {type: type(r), node: related})[0..10] as relationships
LIMIT 10
"""
results = await client.query_graph(
cypher,
parameters={"name": entity_name},
)
if not results:
return f"No entities found related to '{entity_name}'"
output_parts = [f"## Entities Related to: {entity_name}\n"]
for record in results:
node = record.get("n", {})
relationships = record.get("relationships", [])
name = node.get("name", node.get("title", "Unknown"))
labels = node.get("labels", [])
output_parts.append(f"### {name}")
if labels:
output_parts.append(f"Type: {', '.join(labels)}")
if relationships:
output_parts.append("**Connections:**")
for rel in relationships[:5]: # Limit to 5 relationships
rel_type = rel.get("type", "RELATED_TO")
related_node = rel.get("node", {})
related_name = related_node.get(
"name", related_node.get("title", "Unknown")
)
output_parts.append(f" - {rel_type}{related_name}")
output_parts.append("")
return "\n".join(output_parts)
except Exception as e:
logger.error("librarian_find_related_error", error=str(e))
return f"Error finding related entities: {str(e)}"
# ============================================================================
# Wiki Write Operations
# ============================================================================
async def update_wiki_page(
page_id: int,
content: str | None = None,
title: str | None = None,
tags: list[str] | None = None,
description: str | None = None,
) -> str:
"""
Update an existing wiki page.
Supports partial updates - only specify the fields you want to change.
Changes trigger automatic vector re-indexing and knowledge graph updates.
Use this for:
- Correcting information in a page
- Adding content to an existing page
- Updating tags to organize pages into dossiers
- Fixing descriptions or titles
Args:
page_id: ID of the page to update (get from search_wiki results)
content: New markdown content (optional - only if changing content)
title: New title (optional - only if renaming)
tags: New tag list (optional - replaces existing tags)
description: New description (optional)
Returns:
Confirmation with updated page details
Examples:
update_wiki_page(42, content="# Updated Content\\n\\nNew information here")
update_wiki_page(42, tags=["projects", "devops"]) # Add to dossiers
update_wiki_page(42, description="Updated description")
"""
try:
async with LibraryDeskClient() as client:
page = await client.update_wiki_page(
page_id=page_id,
content=content,
title=title,
tags=tags,
description=description,
)
# Build update summary
updated_fields = []
if content is not None:
updated_fields.append("content")
if title is not None:
updated_fields.append("title")
if tags is not None:
updated_fields.append("tags")
if description is not None:
updated_fields.append("description")
output_parts = [
f"## Page Updated: {page.title}",
f"**Path:** {page.path}",
f"**Updated fields:** {', '.join(updated_fields)}",
]
if page.tags:
output_parts.append(f"**Tags:** {', '.join(page.tags)}")
output_parts.append("\n*Vector embeddings and knowledge graph will be updated automatically.*")
logger.info(
"librarian_update_page",
page_id=page_id,
updated_fields=updated_fields,
)
return "\n".join(output_parts)
except Exception as e:
logger.error("librarian_update_page_error", error=str(e), page_id=page_id)
return f"Error updating page {page_id}: {str(e)}"
async def create_wiki_page(
title: str,
path: str,
content: str,
tags: list[str],
description: str = "",
) -> str:
"""
Create a new wiki page with user-provided content.
Use this when:
- User provides specific content to add
- Creating simple notes or reminders
- The content is already known/composed
For research-backed pages where you need to gather information first,
use smart_create_wiki_page instead.
Args:
title: Page title
path: Page path (e.g., "/projects/my-project" or "/notes/meeting-2024")
content: Markdown content for the page
tags: List of tags/dossiers (e.g., ["projects", "devops"])
description: Short description of the page
Returns:
Confirmation with created page details
Examples:
create_wiki_page(
title="SSL Renewal Reminder",
path="/reminders/ssl-renewal",
content="# SSL Renewal\\n\\nRemember to renew SSL cert on Jan 15",
tags=["reminders", "infrastructure"],
description="Certificate renewal reminder"
)
"""
try:
async with LibraryDeskClient() as client:
page = await client.create_wiki_page(
title=title,
path=path,
content=content,
tags=tags,
description=description,
)
output_parts = [
f"## Page Created: {page.title}",
f"**ID:** {page.id}",
f"**Path:** {page.path}",
]
if page.tags:
output_parts.append(f"**Tags:** {', '.join(page.tags)}")
if page.description:
output_parts.append(f"**Description:** {page.description}")
output_parts.append("\n*Vector embeddings and knowledge graph will be updated automatically.*")
logger.info(
"librarian_create_page",
page_id=page.id,
title=title,
path=path,
)
return "\n".join(output_parts)
except Exception as e:
logger.error("librarian_create_page_error", error=str(e), title=title)
return f"Error creating page: {str(e)}"
async def smart_create_wiki_page(
topic: str,
tags: list[str],
path: str | None = None,
include_web_research: bool = True,
include_wiki_search: bool = True,
) -> str:
"""
Create a wiki page with automatic research and content synthesis.
This is the RECOMMENDED way to create pages about topics. It will:
1. Search existing wiki, knowledge graph, and web for relevant information
2. Use an LLM to synthesize findings into well-structured content
3. Create the page with proper source attribution
4. Automatically link entities bidirectionally in the knowledge graph
Use this when:
- User says "Create a page about X"
- User says "Add information about X to the wiki"
- You need to research a topic before writing
- The topic would benefit from existing knowledge context
Args:
topic: The topic to research and create a page about
tags: List of tags/dossiers for categorization
path: Optional custom path (auto-generated from topic if not provided)
include_web_research: Whether to search the web (default: True)
include_wiki_search: Whether to search existing wiki (default: True)
Returns:
Summary of created page with research statistics
Examples:
smart_create_wiki_page("Docker Compose", tags=["technology", "devops"])
smart_create_wiki_page("Home network architecture", tags=["infrastructure"], include_web_research=False)
"""
try:
async with LibraryDeskClient() as client:
response = await client.smart_create_wiki_page(
topic=topic,
tags=tags,
path=path,
include_web_research=include_web_research,
include_wiki_search=include_wiki_search,
)
page = response.page
research = response.research_summary
linking = response.entity_linking
output_parts = [
f"## Page Created: {page.title}",
f"**ID:** {page.id}",
f"**Path:** {page.path}",
]
if page.tags:
output_parts.append(f"**Tags:** {', '.join(page.tags)}")
# Research summary
output_parts.append("\n### Research Summary")
output_parts.append(f"- **Wiki results used:** {research.wiki_results}")
output_parts.append(f"- **Web results used:** {research.web_results}")
output_parts.append(f"- **Graph entities found:** {research.graph_entities}")
output_parts.append(f"- **Keywords extracted:** {research.keywords_extracted}")
output_parts.append(f"- **Total sources:** {response.sources_used}")
output_parts.append(f"- **Research time:** {research.timing_ms}ms")
# Entity linking
if linking.forward_links > 0 or linking.backward_links > 0:
output_parts.append("\n### Knowledge Graph Updates")
output_parts.append(f"- **Forward links created:** {linking.forward_links}")
output_parts.append(f"- **Backward links created:** {linking.backward_links}")
output_parts.append(f"- **Related pages updated:** {linking.pages_updated}")
logger.info(
"librarian_smart_create",
topic=topic,
page_id=page.id,
sources_used=response.sources_used,
)
return "\n".join(output_parts)
except Exception as e:
logger.error("librarian_smart_create_error", error=str(e), topic=topic)
return f"Error creating page about '{topic}': {str(e)}"
# ============================================================================
# Tool Collection for Registration
# ============================================================================
# All tools available to The Librarian
LIBRARIAN_TOOLS = [
# Research tools
hybrid_search,
search_wiki,
get_wiki_page,
list_dossiers,
get_dossier_pages,
semantic_search,
explore_knowledge_graph,
find_related_entities,
# Write tools
create_wiki_page,
update_wiki_page,
smart_create_wiki_page,
]