added files

cognee/__main__.py

@@ -0,0 +1,4 @@
+from cognee.cli._cognee import main
+
+if __name__ == "__main__":
+    main()

cognee/cli/__init__.py

@@ -0,0 +1,10 @@
+from cognee.cli.reference import SupportsCliCommand
+from cognee.cli.exceptions import CliCommandException
+
+DEFAULT_DOCS_URL = "https://docs.cognee.ai"
+
+__all__ = [
+    "SupportsCliCommand",
+    "CliCommandException",
+    "DEFAULT_DOCS_URL",
+]

cognee/cli/_cognee.py

@@ -0,0 +1,180 @@
+import sys
+import os
+import argparse
+from typing import Any, Sequence, Dict, Type, cast, List
+import click
+
+try:
+    import rich_argparse
+    from rich.markdown import Markdown
+
+    HAS_RICH = True
+except ImportError:
+    HAS_RICH = False
+
+from cognee.cli import SupportsCliCommand, DEFAULT_DOCS_URL
+from cognee.cli.config import CLI_DESCRIPTION
+from cognee.cli import debug
+import cognee.cli.echo as fmt
+from cognee.cli.exceptions import CliCommandException
+
+
+ACTION_EXECUTED = False
+
+
+def print_help(parser: argparse.ArgumentParser) -> None:
+    if not ACTION_EXECUTED:
+        parser.print_help()
+
+
+class DebugAction(argparse.Action):
+    def __init__(
+        self,
+        option_strings: Sequence[str],
+        dest: Any = argparse.SUPPRESS,
+        default: Any = argparse.SUPPRESS,
+        help: str = None,
+    ) -> None:
+        super(DebugAction, self).__init__(
+            option_strings=option_strings, dest=dest, default=default, nargs=0, help=help
+        )
+
+    def __call__(
+        self,
+        parser: argparse.ArgumentParser,
+        namespace: argparse.Namespace,
+        values: Any,
+        option_string: str = None,
+    ) -> None:
+        # Enable debug mode for stack traces
+        debug.enable_debug()
+        fmt.note("Debug mode enabled. Full stack traces will be shown.")
+
+
+# Debug functionality is now in cognee.cli.debug module
+
+
+def _discover_commands() -> List[Type[SupportsCliCommand]]:
+    """Discover all available CLI commands"""
+    # Import commands dynamically to avoid early cognee initialization
+    commands = []
+
+    command_modules = [
+        ("cognee.cli.commands.add_command", "AddCommand"),
+        ("cognee.cli.commands.search_command", "SearchCommand"),
+        ("cognee.cli.commands.cognify_command", "CognifyCommand"),
+        ("cognee.cli.commands.delete_command", "DeleteCommand"),
+        ("cognee.cli.commands.config_command", "ConfigCommand"),
+    ]
+
+    for module_path, class_name in command_modules:
+        try:
+            module = __import__(module_path, fromlist=[class_name])
+            command_class = getattr(module, class_name)
+            commands.append(command_class)
+        except (ImportError, AttributeError) as e:
+            fmt.warning(f"Failed to load command {class_name}: {e}")
+
+    return commands
+
+
+def _create_parser() -> tuple[argparse.ArgumentParser, Dict[str, SupportsCliCommand]]:
+    parser = argparse.ArgumentParser(
+        description=f"{CLI_DESCRIPTION} Further help is available at {DEFAULT_DOCS_URL}."
+    )
+
+    # Get version dynamically
+    try:
+        from cognee.version import get_cognee_version
+
+        version = get_cognee_version()
+    except ImportError:
+        version = "unknown"
+
+    parser.add_argument("--version", action="version", version=f"cognee {version}")
+    parser.add_argument(
+        "--debug",
+        action=DebugAction,
+        help="Enable debug mode to show full stack traces on exceptions",
+    )
+
+    subparsers = parser.add_subparsers(title="Available commands", dest="command")
+
+    # Discover and install commands
+    command_classes = _discover_commands()
+    installed_commands: Dict[str, SupportsCliCommand] = {}
+
+    for command_class in command_classes:
+        command = command_class()
+        if command.command in installed_commands:
+            continue
+
+        command_parser = subparsers.add_parser(
+            command.command,
+            help=command.help_string,
+            description=command.description if hasattr(command, "description") else None,
+        )
+        command.configure_parser(command_parser)
+        installed_commands[command.command] = command
+
+    # Add rich formatting if available
+    if HAS_RICH:
+
+        def add_formatter_class(parser: argparse.ArgumentParser) -> None:
+            parser.formatter_class = rich_argparse.RichHelpFormatter
+
+            if parser.description:
+                parser.description = Markdown(parser.description, style="argparse.text")
+            for action in parser._actions:
+                if isinstance(action, argparse._SubParsersAction):
+                    for _subcmd, subparser in action.choices.items():
+                        add_formatter_class(subparser)
+
+        add_formatter_class(parser)
+
+    return parser, installed_commands
+
+
+def main() -> int:
+    """Main CLI entry point"""
+    parser, installed_commands = _create_parser()
+    args = parser.parse_args()
+
+    if cmd := installed_commands.get(args.command):
+        try:
+            cmd.execute(args)
+        except Exception as ex:
+            docs_url = cmd.docs_url if hasattr(cmd, "docs_url") else DEFAULT_DOCS_URL
+            error_code = -1
+            raiseable_exception = ex
+
+            # Handle CLI-specific exceptions
+            if isinstance(ex, CliCommandException):
+                error_code = ex.error_code
+                docs_url = ex.docs_url or docs_url
+                raiseable_exception = ex.raiseable_exception
+
+            # Print exception
+            if raiseable_exception:
+                fmt.error(str(ex))
+
+            fmt.note(f"Please refer to our docs at '{docs_url}' for further assistance.")
+
+            if debug.is_debug_enabled() and raiseable_exception:
+                raise raiseable_exception
+
+            return error_code
+    else:
+        print_help(parser)
+        return -1
+
+    return 0
+
+
+def _main() -> None:
+    """Script entry point"""
+    sys.exit(main())
+
+
+if __name__ == "__main__":
+    sys.exit(main())

cognee/cli/commands/__init__.py

@@ -0,0 +1 @@
+# CLI Commands package

cognee/cli/commands/add_command.py

@@ -0,0 +1,80 @@
+import argparse
+import asyncio
+from typing import Optional
+
+from cognee.cli.reference import SupportsCliCommand
+from cognee.cli import DEFAULT_DOCS_URL
+import cognee.cli.echo as fmt
+from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
+
+
+class AddCommand(SupportsCliCommand):
+    command = "add"
+    help_string = "Add data to Cognee for knowledge graph processing"
+    docs_url = DEFAULT_DOCS_URL
+    description = """
+Add data to Cognee for knowledge graph processing.
+
+This is the first step in the Cognee workflow - it ingests raw data and prepares it
+for processing. The function accepts various data formats including text, files, and
+binary streams, then stores them in a specified dataset for further processing.
+
+Supported Input Types:
+- **Text strings**: Direct text content
+- **File paths**: Local file paths (absolute paths starting with "/")  
+- **File URLs**: "file:///absolute/path" or "file://relative/path"
+- **S3 paths**: "s3://bucket-name/path/to/file"
+- **Lists**: Multiple files or text strings in a single call
+
+Supported File Formats:
+- Text files (.txt, .md, .csv)
+- PDFs (.pdf)
+- Images (.png, .jpg, .jpeg) - extracted via OCR/vision models
+- Audio files (.mp3, .wav) - transcribed to text
+- Code files (.py, .js, .ts, etc.) - parsed for structure and content
+- Office documents (.docx, .pptx)
+
+After adding data, use `cognee cognify` to process it into knowledge graphs.
+    """
+
+    def configure_parser(self, parser: argparse.ArgumentParser) -> None:
+        parser.add_argument(
+            "data",
+            nargs="+",
+            help="Data to add: text content, file paths (/path/to/file), file URLs (file://path), S3 paths (s3://bucket/file), or mix of these",
+        )
+        parser.add_argument(
+            "--dataset-name",
+            "-d",
+            default="main_dataset",
+            help="Dataset name to organize your data (default: main_dataset)",
+        )
+
+    def execute(self, args: argparse.Namespace) -> None:
+        try:
+            # Import cognee here to avoid circular imports
+            import cognee
+
+            fmt.echo(f"Adding {len(args.data)} item(s) to dataset '{args.dataset_name}'...")
+
+            # Run the async add function
+            async def run_add():
+                try:
+                    # Pass all data items as a list to cognee.add if multiple items
+                    if len(args.data) == 1:
+                        data_to_add = args.data[0]
+                    else:
+                        data_to_add = args.data
+
+                    fmt.echo("Processing data...")
+                    await cognee.add(data=data_to_add, dataset_name=args.dataset_name)
+                    fmt.success(f"Successfully added data to dataset '{args.dataset_name}'")
+                except Exception as e:
+                    raise CliCommandInnerException(f"Failed to add data: {str(e)}")
+
+            asyncio.run(run_add())
+
+        except Exception as e:
+            if isinstance(e, CliCommandInnerException):
+                raise CliCommandException(str(e), error_code=1)
+            raise CliCommandException(f"Error adding data: {str(e)}", error_code=1)

cognee/cli/commands/cognify_command.py

@@ -0,0 +1,128 @@
+import argparse
+import asyncio
+from typing import Optional
+
+from cognee.cli.reference import SupportsCliCommand
+from cognee.cli import DEFAULT_DOCS_URL
+from cognee.cli.config import CHUNKER_CHOICES
+import cognee.cli.echo as fmt
+from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
+
+
+class CognifyCommand(SupportsCliCommand):
+    command = "cognify"
+    help_string = "Transform ingested data into a structured knowledge graph"
+    docs_url = DEFAULT_DOCS_URL
+    description = """
+Transform ingested data into a structured knowledge graph.
+
+This is the core processing step in Cognee that converts raw text and documents
+into an intelligent knowledge graph. It analyzes content, extracts entities and
+relationships, and creates semantic connections for enhanced search and reasoning.
+
+Processing Pipeline:
+1. **Document Classification**: Identifies document types and structures
+2. **Permission Validation**: Ensures user has processing rights  
+3. **Text Chunking**: Breaks content into semantically meaningful segments
+4. **Entity Extraction**: Identifies key concepts, people, places, organizations
+5. **Relationship Detection**: Discovers connections between entities
+6. **Graph Construction**: Builds semantic knowledge graph with embeddings
+7. **Content Summarization**: Creates hierarchical summaries for navigation
+
+After successful cognify processing, use `cognee search` to query the knowledge graph.
+    """
+
+    def configure_parser(self, parser: argparse.ArgumentParser) -> None:
+        parser.add_argument(
+            "--datasets",
+            "-d",
+            nargs="*",
+            help="Dataset name(s) to process. Processes all available data if not specified. Can be multiple: --datasets dataset1 dataset2",
+        )
+        parser.add_argument(
+            "--chunk-size",
+            type=int,
+            help="Maximum tokens per chunk. Auto-calculated based on LLM if not specified (~512-8192 tokens)",
+        )
+        parser.add_argument(
+            "--ontology-file", help="Path to RDF/OWL ontology file for domain-specific entity types"
+        )
+        parser.add_argument(
+            "--chunker",
+            choices=CHUNKER_CHOICES,
+            default="TextChunker",
+            help="Text chunking strategy (default: TextChunker)",
+        )
+        parser.add_argument(
+            "--background",
+            "-b",
+            action="store_true",
+            help="Run processing in background and return immediately (recommended for large datasets)",
+        )
+        parser.add_argument(
+            "--verbose", "-v", action="store_true", help="Show detailed progress information"
+        )
+
+    def execute(self, args: argparse.Namespace) -> None:
+        try:
+            # Import cognee here to avoid circular imports
+            import cognee
+
+            # Prepare datasets parameter
+            datasets = args.datasets if args.datasets else None
+            dataset_msg = f" for datasets {datasets}" if datasets else " for all available data"
+            fmt.echo(f"Starting cognification{dataset_msg}...")
+
+            if args.verbose:
+                fmt.note("This process will analyze your data and build knowledge graphs.")
+                fmt.note("Depending on data size, this may take several minutes.")
+                if args.background:
+                    fmt.note(
+                        "Running in background mode - the process will continue after this command exits."
+                    )
+
+            # Prepare chunker parameter - will be handled in the async function
+
+            # Run the async cognify function
+            async def run_cognify():
+                try:
+                    # Import chunker classes here
+                    from cognee.modules.chunking import TextChunker
+
+                    chunker_class = TextChunker  # Default
+                    if args.chunker == "LangchainChunker":
+                        try:
+                            from cognee.modules.chunking import LangchainChunker
+
+                            chunker_class = LangchainChunker
+                        except ImportError:
+                            fmt.warning("LangchainChunker not available, using TextChunker")
+
+                    result = await cognee.cognify(
+                        datasets=datasets,
+                        chunker=chunker_class,
+                        chunk_size=args.chunk_size,
+                        ontology_file_path=args.ontology_file,
+                        run_in_background=args.background,
+                    )
+                    return result
+                except Exception as e:
+                    raise CliCommandInnerException(f"Failed to cognify: {str(e)}")
+
+            result = asyncio.run(run_cognify())
+
+            if args.background:
+                fmt.success("Cognification started in background!")
+                if args.verbose and result:
+                    fmt.echo(
+                        "Background processing initiated. Use pipeline monitoring to track progress."
+                    )
+            else:
+                fmt.success("Cognification completed successfully!")
+                if args.verbose and result:
+                    fmt.echo(f"Processing results: {result}")
+
+        except Exception as e:
+            if isinstance(e, CliCommandInnerException):
+                raise CliCommandException(str(e), error_code=1)
+            raise CliCommandException(f"Error during cognification: {str(e)}", error_code=1)

cognee/cli/commands/config_command.py

@@ -0,0 +1,148 @@
+import argparse
+import json
+from typing import Optional, Any
+
+from cognee.cli.reference import SupportsCliCommand
+from cognee.cli import DEFAULT_DOCS_URL
+import cognee.cli.echo as fmt
+from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
+
+
+class ConfigCommand(SupportsCliCommand):
+    command = "config"
+    help_string = "Manage cognee configuration settings"
+    docs_url = DEFAULT_DOCS_URL
+    description = """
+The `cognee config` command allows you to view and modify configuration settings.
+
+You can:
+- View all current configuration settings
+- Get specific configuration values  
+- Set configuration values
+- Reset configuration to defaults
+
+Configuration changes will affect how cognee processes and stores data.
+    """
+
+    def configure_parser(self, parser: argparse.ArgumentParser) -> None:
+        subparsers = parser.add_subparsers(dest="config_action", help="Configuration actions")
+
+        # Get command
+        get_parser = subparsers.add_parser("get", help="Get configuration value(s)")
+        get_parser.add_argument(
+            "key", nargs="?", help="Configuration key to retrieve (shows all if not specified)"
+        )
+
+        # Set command
+        set_parser = subparsers.add_parser("set", help="Set configuration value")
+        set_parser.add_argument("key", help="Configuration key to set")
+        set_parser.add_argument("value", help="Configuration value to set")
+
+        # List command
+        subparsers.add_parser("list", help="List all configuration keys")
+
+        # Reset command
+        reset_parser = subparsers.add_parser("reset", help="Reset configuration to defaults")
+        reset_parser.add_argument(
+            "--force", "-f", action="store_true", help="Skip confirmation prompt"
+        )
+
+    def execute(self, args: argparse.Namespace) -> None:
+        try:
+            # Import cognee here to avoid circular imports
+            import cognee
+
+            if not hasattr(args, "config_action") or args.config_action is None:
+                fmt.error("Please specify a config action: get, set, list, or reset")
+                return
+
+            if args.config_action == "get":
+                self._handle_get(args)
+            elif args.config_action == "set":
+                self._handle_set(args)
+            elif args.config_action == "list":
+                self._handle_list(args)
+            elif args.config_action == "reset":
+                self._handle_reset(args)
+            else:
+                fmt.error(f"Unknown config action: {args.config_action}")
+
+        except Exception as e:
+            if isinstance(e, CliCommandInnerException):
+                raise CliCommandException(str(e), error_code=1)
+            raise CliCommandException(f"Error managing configuration: {str(e)}", error_code=1)
+
+    def _handle_get(self, args: argparse.Namespace) -> None:
+        try:
+            import cognee
+
+            if args.key:
+                # Get specific key
+                try:
+                    value = cognee.config.get(args.key)
+                    fmt.echo(f"{args.key}: {value}")
+                except Exception:
+                    fmt.error(f"Configuration key '{args.key}' not found")
+            else:
+                # Get all configuration
+                try:
+                    config_dict = (
+                        cognee.config.get_all() if hasattr(cognee.config, "get_all") else {}
+                    )
+                    if config_dict:
+                        fmt.echo("Current configuration:")
+                        for key, value in config_dict.items():
+                            fmt.echo(f"  {key}: {value}")
+                    else:
+                        fmt.echo("No configuration settings found")
+                except Exception:
+                    fmt.note("Configuration viewing not fully implemented yet")
+
+        except Exception as e:
+            raise CliCommandInnerException(f"Failed to get configuration: {str(e)}")
+
+    def _handle_set(self, args: argparse.Namespace) -> None:
+        try:
+            import cognee
+
+            # Try to parse value as JSON, otherwise treat as string
+            try:
+                value = json.loads(args.value)
+            except json.JSONDecodeError:
+                value = args.value
+
+            try:
+                cognee.config.set(args.key, value)
+                fmt.success(f"Set {args.key} = {value}")
+            except Exception:
+                fmt.error(f"Failed to set configuration key '{args.key}'")
+
+        except Exception as e:
+            raise CliCommandInnerException(f"Failed to set configuration: {str(e)}")
+
+    def _handle_list(self, args: argparse.Namespace) -> None:
+        try:
+            import cognee
+
+            # This would need to be implemented in cognee.config
+            fmt.note("Available configuration keys:")
+            fmt.echo("  LLM_MODEL")
+            fmt.echo("  VECTOR_DB_URL")
+            fmt.echo("  GRAPH_DB_URL")
+            fmt.echo("  (Use 'cognee config get' to see current values)")
+
+        except Exception as e:
+            raise CliCommandInnerException(f"Failed to list configuration: {str(e)}")
+
+    def _handle_reset(self, args: argparse.Namespace) -> None:
+        try:
+            if not args.force:
+                if not fmt.confirm("Reset all configuration to defaults?"):
+                    fmt.echo("Reset cancelled.")
+                    return
+
+            fmt.note("Configuration reset not fully implemented yet")
+            fmt.echo("This would reset all settings to their default values")
+
+        except Exception as e:
+            raise CliCommandInnerException(f"Failed to reset configuration: {str(e)}")

cognee/cli/commands/delete_command.py

@@ -0,0 +1,80 @@
+import argparse
+import asyncio
+from typing import Optional
+
+from cognee.cli.reference import SupportsCliCommand
+from cognee.cli import DEFAULT_DOCS_URL
+import cognee.cli.echo as fmt
+from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
+
+
+class DeleteCommand(SupportsCliCommand):
+    command = "delete"
+    help_string = "Delete data from cognee knowledge base"
+    docs_url = DEFAULT_DOCS_URL
+    description = """
+The `cognee delete` command removes data from your knowledge base.
+
+You can delete:
+- Specific datasets by name
+- All data (with confirmation)
+- Data for specific users
+
+Be careful with deletion operations as they are irreversible.
+    """
+
+    def configure_parser(self, parser: argparse.ArgumentParser) -> None:
+        parser.add_argument("--dataset-name", "-d", help="Specific dataset to delete")
+        parser.add_argument("--user-id", "-u", help="User ID to delete data for")
+        parser.add_argument(
+            "--all", action="store_true", help="Delete all data (requires confirmation)"
+        )
+        parser.add_argument("--force", "-f", action="store_true", help="Skip confirmation prompts")
+
+    def execute(self, args: argparse.Namespace) -> None:
+        try:
+            # Import cognee here to avoid circular imports
+            import cognee
+
+            # Validate arguments
+            if not any([args.dataset_name, args.user_id, args.all]):
+                fmt.error("Please specify what to delete: --dataset-name, --user-id, or --all")
+                return
+
+            # Build confirmation message
+            if args.all:
+                confirm_msg = "Delete ALL data from cognee?"
+                operation = "all data"
+            elif args.dataset_name:
+                confirm_msg = f"Delete dataset '{args.dataset_name}'?"
+                operation = f"dataset '{args.dataset_name}'"
+            elif args.user_id:
+                confirm_msg = f"Delete all data for user '{args.user_id}'?"
+                operation = f"data for user '{args.user_id}'"
+
+            # Confirm deletion unless forced
+            if not args.force:
+                fmt.warning("This operation is irreversible!")
+                if not fmt.confirm(confirm_msg):
+                    fmt.echo("Deletion cancelled.")
+                    return
+
+            fmt.echo(f"Deleting {operation}...")
+
+            # Run the async delete function
+            async def run_delete():
+                try:
+                    if args.all:
+                        await cognee.delete(dataset_name=None, user_id=args.user_id)
+                    else:
+                        await cognee.delete(dataset_name=args.dataset_name, user_id=args.user_id)
+                except Exception as e:
+                    raise CliCommandInnerException(f"Failed to delete: {str(e)}")
+
+            asyncio.run(run_delete())
+            fmt.success(f"Successfully deleted {operation}")
+
+        except Exception as e:
+            if isinstance(e, CliCommandInnerException):
+                raise CliCommandException(str(e), error_code=1)
+            raise CliCommandException(f"Error deleting data: {str(e)}", error_code=1)

cognee/cli/commands/search_command.py

@@ -0,0 +1,149 @@
+import argparse
+import asyncio
+import json
+from typing import Optional
+
+from cognee.cli.reference import SupportsCliCommand
+from cognee.cli import DEFAULT_DOCS_URL
+from cognee.cli.config import SEARCH_TYPE_CHOICES, OUTPUT_FORMAT_CHOICES
+import cognee.cli.echo as fmt
+from cognee.cli.exceptions import CliCommandException, CliCommandInnerException
+
+
+class SearchCommand(SupportsCliCommand):
+    command = "search"
+    help_string = "Search and query the knowledge graph for insights, information, and connections"
+    docs_url = DEFAULT_DOCS_URL
+    description = """
+Search and query the knowledge graph for insights, information, and connections.
+
+This is the final step in the Cognee workflow that retrieves information from the
+processed knowledge graph. It supports multiple search modes optimized for different
+use cases - from simple fact retrieval to complex reasoning and code analysis.
+
+Search Types & Use Cases:
+
+**GRAPH_COMPLETION** (Default - Recommended):
+    Natural language Q&A using full graph context and LLM reasoning.
+    Best for: Complex questions, analysis, summaries, insights.
+
+**RAG_COMPLETION**:
+    Traditional RAG using document chunks without graph structure.
+    Best for: Direct document retrieval, specific fact-finding.
+
+**INSIGHTS**:
+    Structured entity relationships and semantic connections.
+    Best for: Understanding concept relationships, knowledge mapping.
+
+**CHUNKS**:
+    Raw text segments that match the query semantically.
+    Best for: Finding specific passages, citations, exact content.
+
+**SUMMARIES**:
+    Pre-generated hierarchical summaries of content.
+    Best for: Quick overviews, document abstracts, topic summaries.
+
+**CODE**:
+    Code-specific search with syntax and semantic understanding.
+    Best for: Finding functions, classes, implementation patterns.
+    """
+
+    def configure_parser(self, parser: argparse.ArgumentParser) -> None:
+        parser.add_argument("query_text", help="Your question or search query in natural language")
+        parser.add_argument(
+            "--query-type",
+            "-t",
+            choices=SEARCH_TYPE_CHOICES,
+            default="GRAPH_COMPLETION",
+            help="Search mode (default: GRAPH_COMPLETION for conversational AI responses)",
+        )
+        parser.add_argument(
+            "--datasets",
+            "-d",
+            nargs="*",
+            help="Dataset name(s) to search within. Searches all accessible datasets if not specified",
+        )
+        parser.add_argument(
+            "--top-k",
+            "-k",
+            type=int,
+            default=10,
+            help="Maximum number of results to return (default: 10, max: 100)",
+        )
+        parser.add_argument(
+            "--system-prompt",
+            help="Custom system prompt file for LLM-based search types (default: answer_simple_question.txt)",
+        )
+        parser.add_argument(
+            "--output-format",
+            "-f",
+            choices=OUTPUT_FORMAT_CHOICES,
+            default="pretty",
+            help="Output format (default: pretty)",
+        )
+
+    def execute(self, args: argparse.Namespace) -> None:
+        try:
+            # Import cognee here to avoid circular imports
+            import cognee
+            from cognee.modules.search.types import SearchType
+
+            # Convert string to SearchType enum
+            query_type = SearchType[args.query_type]
+
+            datasets_msg = (
+                f" in datasets {args.datasets}" if args.datasets else " across all datasets"
+            )
+            fmt.echo(f"Searching for: '{args.query_text}' (type: {args.query_type}){datasets_msg}")
+
+            # Run the async search function
+            async def run_search():
+                try:
+                    results = await cognee.search(
+                        query_text=args.query_text,
+                        query_type=query_type,
+                        datasets=args.datasets,
+                        system_prompt_path=args.system_prompt or "answer_simple_question.txt",
+                        top_k=args.top_k,
+                    )
+                    return results
+                except Exception as e:
+                    raise CliCommandInnerException(f"Failed to search: {str(e)}")
+
+            results = asyncio.run(run_search())
+
+            # Format and display results
+            if args.output_format == "json":
+                fmt.echo(json.dumps(results, indent=2, default=str))
+            elif args.output_format == "simple":
+                for i, result in enumerate(results, 1):
+                    fmt.echo(f"{i}. {result}")
+            else:  # pretty format
+                if not results:
+                    fmt.warning("No results found for your query.")
+                    return
+
+                fmt.echo(f"\nFound {len(results)} result(s) using {args.query_type}:")
+                fmt.echo("=" * 60)
+
+                if args.query_type in ["GRAPH_COMPLETION", "RAG_COMPLETION"]:
+                    # These return conversational responses
+                    for i, result in enumerate(results, 1):
+                        fmt.echo(f"{fmt.bold('Response:')} {result}")
+                        if i < len(results):
+                            fmt.echo("-" * 40)
+                elif args.query_type == "CHUNKS":
+                    # These return text chunks
+                    for i, result in enumerate(results, 1):
+                        fmt.echo(f"{fmt.bold(f'Chunk {i}:')} {result}")
+                        fmt.echo()
+                else:
+                    # Generic formatting for other types
+                    for i, result in enumerate(results, 1):
+                        fmt.echo(f"{fmt.bold(f'Result {i}:')} {result}")
+                        fmt.echo()
+
+        except Exception as e:
+            if isinstance(e, CliCommandInnerException):
+                raise CliCommandException(str(e), error_code=1)
+            raise CliCommandException(f"Error searching: {str(e)}", error_code=1)

cognee/cli/config.py

@@ -0,0 +1,33 @@
+"""
+CLI configuration and constants to avoid hardcoded values
+"""
+
+# CLI Constants
+CLI_DESCRIPTION = "Cognee CLI - Manage your knowledge graphs and cognitive processing pipelines."
+DEFAULT_DOCS_URL = "https://docs.cognee.ai"
+
+# Command descriptions - these should match the actual command implementations
+COMMAND_DESCRIPTIONS = {
+    "add": "Add data to Cognee for knowledge graph processing",
+    "search": "Search and query the knowledge graph for insights, information, and connections",
+    "cognify": "Transform ingested data into a structured knowledge graph",
+    "delete": "Delete data from cognee knowledge base",
+    "config": "Manage cognee configuration settings",
+}
+
+# Search type choices
+SEARCH_TYPE_CHOICES = [
+    "GRAPH_COMPLETION",
+    "RAG_COMPLETION",
+    "INSIGHTS",
+    "CHUNKS",
+    "SUMMARIES",
+    "CODE",
+    "CYPHER",
+]
+
+# Chunker choices
+CHUNKER_CHOICES = ["TextChunker", "LangchainChunker"]
+
+# Output format choices
+OUTPUT_FORMAT_CHOICES = ["json", "pretty", "simple"]

cognee/cli/debug.py

@@ -0,0 +1,21 @@
+"""Provides a global debug setting for the CLI - following dlt patterns"""
+
+_DEBUG_FLAG = False
+
+
+def enable_debug() -> None:
+    """Enable debug mode for CLI"""
+    global _DEBUG_FLAG
+    _DEBUG_FLAG = True
+
+
+def disable_debug() -> None:
+    """Disable debug mode for CLI"""
+    global _DEBUG_FLAG
+    _DEBUG_FLAG = False
+
+
+def is_debug_enabled() -> bool:
+    """Check if debug mode is enabled"""
+    global _DEBUG_FLAG
+    return _DEBUG_FLAG

cognee/cli/echo.py

@@ -0,0 +1,45 @@
+"""CLI output formatting utilities"""
+
+import sys
+import click
+from typing import Any
+
+
+def echo(message: str = "", color: str = None, err: bool = False) -> None:
+    """Echo a message to stdout or stderr with optional color"""
+    click.secho(message, fg=color, err=err)
+
+
+def note(message: str) -> None:
+    """Print a note in blue"""
+    echo(f"Note: {message}", color="blue")
+
+
+def warning(message: str) -> None:
+    """Print a warning in yellow"""
+    echo(f"Warning: {message}", color="yellow")
+
+
+def error(message: str) -> None:
+    """Print an error in red"""
+    echo(f"Error: {message}", color="red", err=True)
+
+
+def success(message: str) -> None:
+    """Print a success message in green"""
+    echo(f"Success: {message}", color="green")
+
+
+def bold(text: str) -> str:
+    """Make text bold"""
+    return click.style(text, bold=True)
+
+
+def confirm(message: str, default: bool = False) -> bool:
+    """Ask for user confirmation"""
+    return click.confirm(message, default=default)
+
+
+def prompt(message: str, default: Any = None) -> str:
+    """Prompt user for input"""
+    return click.prompt(message, default=default)

cognee/cli/exceptions.py

@@ -0,0 +1,23 @@
+from typing import Optional
+
+
+class CliCommandException(Exception):
+    """Exception raised by CLI commands with additional context"""
+
+    def __init__(
+        self,
+        message: str,
+        error_code: int = -1,
+        docs_url: Optional[str] = None,
+        raiseable_exception: Optional[Exception] = None,
+    ) -> None:
+        super().__init__(message)
+        self.error_code = error_code
+        self.docs_url = docs_url
+        self.raiseable_exception = raiseable_exception
+
+
+class CliCommandInnerException(Exception):
+    """Inner exception for wrapping other exceptions in CLI context"""
+
+    pass

cognee/cli/minimal_cli.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+"""
+Minimal CLI entry point for cognee that avoids early initialization
+"""
+
+import sys
+import os
+from typing import Any, Sequence
+
+# CRITICAL: Prevent verbose logging initialization for CLI-only usage
+# This must be set before any cognee imports to be effective
+os.environ["COGNEE_MINIMAL_LOGGING"] = "true"
+
+
+def get_version() -> str:
+    """Get cognee version without importing the main package"""
+    try:
+        # Try to get version from pyproject.toml first (for development)
+        from pathlib import Path
+
+        pyproject_path = Path(__file__).parent.parent.parent / "pyproject.toml"
+        if pyproject_path.exists():
+            with open(pyproject_path, encoding="utf-8") as f:
+                for line in f:
+                    if line.startswith("version"):
+                        version = line.split("=")[1].strip("'\"\n ")
+                        return f"{version}-local"
+
+        # Fallback to installed package version
+        import importlib.metadata
+
+        return importlib.metadata.version("cognee")
+    except Exception:
+        return "unknown"
+
+
+def get_command_info() -> dict:
+    """Get command information without importing cognee"""
+    return {
+        "add": "Add data to Cognee for knowledge graph processing",
+        "search": "Search and query the knowledge graph for insights, information, and connections",
+        "cognify": "Transform ingested data into a structured knowledge graph",
+        "delete": "Delete data from cognee knowledge base",
+        "config": "Manage cognee configuration settings",
+    }
+
+
+def print_help() -> None:
+    """Print help message with dynamic command descriptions"""
+    commands = get_command_info()
+    command_list = "\n".join(f"    {cmd:<12} {desc}" for cmd, desc in commands.items())
+
+    print(f"""
+usage: cognee [-h] [--version] [--debug] {{{"|".join(commands.keys())}}} ...
+
+Cognee CLI - Manage your knowledge graphs and cognitive processing pipelines.
+
+options:
+  -h, --help            show this help message and exit
+  --version             show program's version number and exit
+  --debug               Enable debug mode to show full stack traces on exceptions
+
+Available commands:
+  {{{",".join(commands.keys())}}}
+{command_list}
+
+For more information on each command, use: cognee <command> --help
+""")
+
+
+def main() -> int:
+    """Minimal CLI main function"""
+    # Handle help and version without any imports - purely static
+    if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] in ["-h", "--help"]):
+        print_help()
+        return 0
+
+    if len(sys.argv) == 2 and sys.argv[1] == "--version":
+        print(f"cognee {get_version()}")
+        return 0
+
+    # For actual commands, import the full CLI with minimal logging
+    try:
+        from cognee.cli._cognee import main as full_main
+
+        return full_main()
+    except Exception as e:
+        if "--debug" in sys.argv:
+            raise
+        print(f"Error: {e}")
+        print("Use --debug for full stack trace")
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

cognee/cli/reference.py

@@ -0,0 +1,23 @@
+from typing import Protocol, Optional
+import argparse
+
+
+class SupportsCliCommand(Protocol):
+    """Protocol for defining one cognee cli command"""
+
+    command: str
+    """name of the command"""
+    help_string: str
+    """the help string for argparse"""
+    description: Optional[str]
+    """the more detailed description for argparse, may include markdown for the docs"""
+    docs_url: Optional[str]
+    """the default docs url to be printed in case of an exception"""
+
+    def configure_parser(self, parser: argparse.ArgumentParser) -> None:
+        """Configures the parser for the given argument"""
+        ...
+
+    def execute(self, args: argparse.Namespace) -> None:
+        """Executes the command with the given arguments"""
+        ...

cognee/cli/suppress_logging.py

@@ -0,0 +1,12 @@
+"""
+Module to suppress verbose logging before any cognee imports.
+This must be imported before any other cognee modules.
+"""
+
+import os
+
+# Set CLI mode to suppress verbose logging
+os.environ["COGNEE_CLI_MODE"] = "true"
+
+# Also set log level to ERROR for extra safety
+os.environ["LOG_LEVEL"] = "ERROR"

cognee/shared/logging_utils.py

@@ -221,13 +221,22 @@ def cleanup_old_logs(logs_dir, max_files):
 
         # Remove old files that exceed the maximum
         if len(log_files) > max_files:
+            deleted_count = 0
             for old_file in log_files[max_files:]:
                 try:
                     old_file.unlink()
-                    logger.info(f"Deleted old log file: {old_file}")
+                    deleted_count += 1
+                    # Only log individual files in non-CLI mode
+                    if os.getenv("COGNEE_CLI_MODE") != "true":
+                        logger.info(f"Deleted old log file: {old_file}")
                 except Exception as e:
+                    # Always log errors
                     logger.error(f"Failed to delete old log file {old_file}: {e}")
 
+            # In CLI mode, show compact summary
+            if os.getenv("COGNEE_CLI_MODE") == "true" and deleted_count > 0:
+                logger.info(f"Cleaned up {deleted_count} old log files")
+
         return True
     except Exception as e:
         logger.error(f"Error cleaning up log files: {e}")
@@ -246,6 +255,11 @@ def setup_logging(log_level=None, name=None):
     """
     global _is_structlog_configured
 
+    # Check if we should use minimal logging (for CLI)
+    if os.getenv("COGNEE_MINIMAL_LOGGING") == "true":
+        return _setup_minimal_logging(log_level, name)
+
+    # Regular detailed logging for non-CLI usage
     log_level = log_level if log_level else log_levels[os.getenv("LOG_LEVEL", "INFO")]
 
     # Configure external library logging early to suppress verbose output
@@ -387,23 +401,60 @@ def setup_logging(log_level=None, name=None):
 
     # Get a configured logger and log system information
     logger = structlog.get_logger(name if name else __name__)
-    logger.info(
-        "Logging initialized",
-        python_version=PYTHON_VERSION,
-        structlog_version=STRUCTLOG_VERSION,
-        cognee_version=COGNEE_VERSION,
-        os_info=OS_INFO,
-    )
 
-    logger.info("Want to learn more? Visit the Cognee documentation: https://docs.cognee.ai")
+    # Provide compact logging for CLI mode, detailed for regular mode
+    if os.getenv("COGNEE_CLI_MODE") == "true":
+        # Compact initialization for CLI
+        logger.info(f"cognee {COGNEE_VERSION} initialized")
+        log_database_configuration_compact(logger)
+    else:
+        # Detailed initialization for regular usage
+        logger.info(
+            "Logging initialized",
+            python_version=PYTHON_VERSION,
+            structlog_version=STRUCTLOG_VERSION,
+            cognee_version=COGNEE_VERSION,
+            os_info=OS_INFO,
+        )
 
-    # Log database configuration
-    log_database_configuration(logger)
+        logger.info("Want to learn more? Visit the Cognee documentation: https://docs.cognee.ai")
+
+        # Log database configuration
+        log_database_configuration(logger)
 
     # Return the configured logger
     return logger
 
 
+def _setup_minimal_logging(log_level=None, name=None):
+    """Setup minimal logging for CLI usage - based on dlt patterns"""
+    global _is_structlog_configured
+
+    # Use ERROR level for minimal logging, or user-specified level
+    log_level = log_level if log_level else log_levels.get("ERROR", logging.ERROR)
+
+    # Configure external library logging to be quiet
+    configure_external_library_logging()
+
+    # Create a simple logger without verbose initialization
+    logger = logging.getLogger(name if name else __name__)
+
+    if not logger.handlers:
+        # Simple console handler for errors only
+        handler = logging.StreamHandler(sys.stderr)
+        formatter = logging.Formatter(fmt="%(levelname)s: %(message)s", style="%")
+        handler.setFormatter(formatter)
+        handler.setLevel(log_level)
+        logger.addHandler(handler)
+        logger.setLevel(log_level)
+
+    # Skip all the verbose initialization, file handlers, cleanup, etc.
+    # Just return a basic logger that only shows important messages
+    _is_structlog_configured = True  # Prevent double initialization
+
+    return logger
+
+
 def get_log_file_location():
     """Return the file path of the log file in use, if any."""
     root_logger = logging.getLogger()

pyproject.toml

@@ -58,7 +58,9 @@ dependencies = [
     "pympler>=1.1,<2.0.0",
     "onnxruntime>=1.0.0,<2.0.0",
     "pylance>=0.22.0,<1.0.0",
-    "kuzu (==0.11.0)"
+    "kuzu (==0.11.0)",
+    "click>=8.0.0,<9.0.0",
+    "rich-argparse>=1.1.0,<2.0.0"
 ]
 
 [project.optional-dependencies]
@@ -143,6 +145,9 @@ debug = ["debugpy>=1.8.9,<2.0.0"]
 Homepage = "https://www.cognee.ai"
 Repository = "https://github.com/topoteretes/cognee"
 
+[project.scripts]
+cognee = "cognee.cli.minimal_cli:main"
+
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"