From f6f37b341b81a56c7987121a0d8d15ee49f9c676 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Thu, 11 Dec 2025 21:27:09 +0100 Subject: [PATCH] feat(phase3): add The Librarian agent with library-desk integration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- src/agents/librarian/__init__.py | 30 ++ src/agents/librarian/agent.py | 286 ++++++++++++ src/agents/librarian/capability.py | 81 ++++ src/agents/librarian/client.py | 685 ++++++++++++++++++++++++++++ src/agents/librarian/tools.py | 701 +++++++++++++++++++++++++++++ 5 files changed, 1783 insertions(+) create mode 100644 src/agents/librarian/__init__.py create mode 100644 src/agents/librarian/agent.py create mode 100644 src/agents/librarian/capability.py create mode 100644 src/agents/librarian/client.py create mode 100644 src/agents/librarian/tools.py diff --git a/src/agents/librarian/__init__.py b/src/agents/librarian/__init__.py new file mode 100644 index 0000000..2a0d6ed --- /dev/null +++ b/src/agents/librarian/__init__.py @@ -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", +] diff --git a/src/agents/librarian/agent.py b/src/agents/librarian/agent.py new file mode 100644 index 0000000..91ce17f --- /dev/null +++ b/src/agents/librarian/agent.py @@ -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)}" diff --git a/src/agents/librarian/capability.py b/src/agents/librarian/capability.py new file mode 100644 index 0000000..3a80904 --- /dev/null +++ b/src/agents/librarian/capability.py @@ -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") diff --git a/src/agents/librarian/client.py b/src/agents/librarian/client.py new file mode 100644 index 0000000..1452174 --- /dev/null +++ b/src/agents/librarian/client.py @@ -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() diff --git a/src/agents/librarian/tools.py b/src/agents/librarian/tools.py new file mode 100644 index 0000000..9dc26a3 --- /dev/null +++ b/src/agents/librarian/tools.py @@ -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, +]