feat: add doc search and doc list commands

Semantic search against the agos-system knowledge corpus.
Lists documentation sources with ingest status.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
2026-08-21 19:28:57 +05:30
parent f3b6b3f1a7
commit 73396c97ea
3 changed files with 114 additions and 0 deletions

View File

@@ -1262,6 +1262,12 @@ def system_metrics_stub():
console.print("[yellow]Note: Full metrics require observability implementation[/yellow]")
@cli.group()
def doc():
"""Documentation search and browse commands"""
pass
install_api_commands(
cli=cli,
agent_group=agent,
@@ -1269,6 +1275,7 @@ install_api_commands(
workflow_group=workflow,
policy_group=policy,
system_group=system,
doc_group=doc,
console=console,
emit_log=_emit_log,
auth_file=_AUTH_FILE,

View File

@@ -19,6 +19,7 @@ def install_api_commands(
workflow_group,
policy_group,
system_group,
doc_group,
console,
emit_log,
auth_file: Path,
@@ -1365,3 +1366,81 @@ def install_api_commands(
installed_count = sum(1 for item in installables_payload if item.get('installed'))
identity_table.add_row('Installables Enabled', str(installed_count))
console.print(identity_table)
# ── Doc commands ──────────────────────────────────────────────────────────
_DOC_CORPUS_NAME = 'agos-system'
def _resolve_doc_corpus_id(correlation_id: str, api_url: Optional[str] = None) -> str:
"""Look up the agos-system corpus by name and return its corpus_id."""
payload = _request_api('GET', '/corpora?limit=100', correlation_id, api_url=api_url)
for c in (payload.get('corpora') or []):
if c.get('name') == _DOC_CORPUS_NAME:
return c['corpus_id']
raise click.ClickException(
f'Documentation corpus "{_DOC_CORPUS_NAME}" not found. '
f'Seed it first: python seeders/seed_cli_docs.py'
)
@doc_group.command(name='search')
@click.argument('query', nargs=-1, required=True)
@click.option('--top-k', default=5, type=int, help='Number of results to return')
@api_url_option
@json_output_option
def doc_search_command(query: Sequence[str], top_k: int, api_url: Optional[str], json_output: bool):
"""Search Agos documentation by keyword or natural language query."""
correlation_id = f'cli_doc_search_{int(time.time() * 1000)}'
resolved_api_url = _resolve_api_url(api_url)
query_text = ' '.join(query).strip()
if not query_text:
raise click.ClickException('Usage: agos doc search <query>')
corpus_id = _resolve_doc_corpus_id(correlation_id, api_url=resolved_api_url)
payload = _request_api(
'POST',
f'/corpora/{corpus_id}/query',
correlation_id,
api_url=resolved_api_url,
json_body={'query': query_text, 'top_k': top_k},
)
if json_output:
_print_json(payload)
return
_show_target(resolved_api_url)
chunks = payload.get('chunks') or payload.get('results') or []
if not chunks:
console.print(f'[dim]No results for: {query_text}[/dim]')
return
console.print(f'[bold blue]Documentation results for:[/bold blue] {query_text}\n')
for i, chunk in enumerate(chunks, 1):
score = chunk.get('score') or chunk.get('similarity') or 0
source = chunk.get('source_name') or chunk.get('source') or '—'
text = (chunk.get('text') or chunk.get('content') or '').strip()
# Truncate long chunks for display
if len(text) > 500:
text = text[:497] + '...'
score_color = 'green' if score > 0.7 else ('yellow' if score > 0.4 else 'dim')
console.print(f'[{score_color}]#{i} score: {score:.3f}[/{score_color}] [cyan]{source}[/cyan]')
console.print(f' {text}\n')
@doc_group.command(name='list')
@api_url_option
@json_output_option
def doc_list_command(api_url: Optional[str], json_output: bool):
"""List all documentation sources in the knowledge base."""
correlation_id = f'cli_doc_list_{int(time.time() * 1000)}'
resolved_api_url = _resolve_api_url(api_url)
corpus_id = _resolve_doc_corpus_id(correlation_id, api_url=resolved_api_url)
payload = _request_api('GET', f'/corpora/{corpus_id}/sources', correlation_id, api_url=resolved_api_url)
if json_output:
_print_json(payload)
return
_show_target(resolved_api_url)
sources = payload.get('sources') or []
if not sources:
console.print('[dim]No documentation sources found.[/dim]')
return
_render_table(
'Documentation Sources',
[('Name', 'name'), ('Type', 'source_type'), ('Status', 'ingest_status'), ('Chunks', 'chunk_count'), ('Added', 'created_at')],
sources,
)

View File

@@ -18,6 +18,7 @@ _MANUAL: Dict[str, Dict[str, object]] = {
'agos agent list',
'agos task list --status running',
'agos workflow run wf_autoblogger_v1 --agent-id agent_123',
'agos doc search "how to run agents"',
'agos help agent',
'agos man workflow run',
],
@@ -419,6 +420,33 @@ _MANUAL: Dict[str, Dict[str, object]] = {
'summary': 'Show the legacy local metrics view.',
'examples': ['agos system metrics'],
},
'doc': {
'summary': (
'Search and browse Agos documentation from the terminal. '
'Queries the agos-system knowledge corpus via semantic search.'
),
'examples': [
'agos doc search how to create an agent',
'agos doc search "workflow automation"',
'agos doc list',
],
},
'doc search': {
'summary': 'Semantic search across Agos documentation. Returns ranked results with relevance scores.',
'examples': [
'agos doc search how to create an agent',
'agos doc search "PKCE login flow"',
'agos doc search plugins --top-k 10',
'agos doc search workflows --json-output',
],
},
'doc list': {
'summary': 'List all documentation sources in the knowledge base.',
'examples': [
'agos doc list',
'agos doc list --json-output',
],
},
}