PlanOpticon

planopticon / docs / api / agent.md
Source Rendered
Blame History Raw 408 lines
1
# Agent API Reference
2
3
::: video_processor.agent.agent_loop
4
5
::: video_processor.agent.skills.base
6
7
::: video_processor.agent.kb_context
8
9
---
10
11
## Overview
12
13
The agent module implements a planning agent that synthesizes knowledge from processed video content into actionable artifacts such as project plans, PRDs, task breakdowns, and roadmaps. The agent operates on knowledge graphs loaded via `KBContext` and uses a skill-based architecture for extensibility.
14
15
**Key components:**
16
17
- **`PlanningAgent`** -- orchestrates skill selection and execution based on user requests
18
- **`AgentContext`** -- shared state passed between skills during execution
19
- **`Skill`** (ABC) -- base class for pluggable agent capabilities
20
- **`Artifact`** -- output produced by skill execution
21
- **`KBContext`** -- loads and merges multiple knowledge graph sources
22
23
---
24
25
## PlanningAgent
26
27
```python
28
from video_processor.agent.agent_loop import PlanningAgent
29
```
30
31
AI agent that synthesizes knowledge into planning artifacts. Uses an LLM to select which skills to execute for a given request, or falls back to keyword matching when no LLM is available.
32
33
### Constructor
34
35
```python
36
def __init__(self, context: AgentContext)
37
```
38
39
| Parameter | Type | Description |
40
|---|---|---|
41
| `context` | `AgentContext` | Shared context containing knowledge graph, query engine, and provider |
42
43
### from_kb_paths()
44
45
```python
46
@classmethod
47
def from_kb_paths(
48
cls,
49
kb_paths: List[Path],
50
provider_manager=None,
51
) -> PlanningAgent
52
```
53
54
Factory method that creates an agent from one or more knowledge base file paths. Handles loading and merging knowledge graphs automatically.
55
56
**Parameters:**
57
58
| Parameter | Type | Default | Description |
59
|---|---|---|---|
60
| `kb_paths` | `List[Path]` | *required* | Paths to `.db` or `.json` knowledge graph files, or directories to search |
61
| `provider_manager` | `ProviderManager` | `None` | LLM provider for agent operations |
62
63
**Returns:** `PlanningAgent` -- configured agent with loaded knowledge base.
64
65
```python
66
from pathlib import Path
67
from video_processor.agent.agent_loop import PlanningAgent
68
from video_processor.providers.manager import ProviderManager
69
70
agent = PlanningAgent.from_kb_paths(
71
kb_paths=[Path("results/knowledge_graph.db")],
72
provider_manager=ProviderManager(),
73
)
74
```
75
76
### execute()
77
78
```python
79
def execute(self, request: str) -> List[Artifact]
80
```
81
82
Execute a user request by selecting and running appropriate skills.
83
84
**Process:**
85
86
1. Build a context summary from the knowledge base statistics
87
2. Format available skills with their descriptions
88
3. Ask the LLM to select skills and parameters (or use keyword matching as fallback)
89
4. Execute selected skills in order, accumulating artifacts
90
91
**Parameters:**
92
93
| Parameter | Type | Description |
94
|---|---|---|
95
| `request` | `str` | Natural language request (e.g., "Generate a project plan") |
96
97
**Returns:** `List[Artifact]` -- generated artifacts from skill execution.
98
99
**LLM mode:** The LLM receives the knowledge base summary, available skills, and user request, then returns a JSON array of `{"skill": "name", "params": {}}` objects to execute.
100
101
**Keyword fallback:** Without an LLM, skills are matched by splitting the skill name into words and checking if any appear in the request text.
102
103
```python
104
artifacts = agent.execute("Create a PRD and task breakdown")
105
for artifact in artifacts:
106
print(f"--- {artifact.name} ({artifact.artifact_type}) ---")
107
print(artifact.content[:500])
108
```
109
110
### chat()
111
112
```python
113
def chat(self, message: str) -> str
114
```
115
116
Interactive chat mode. Maintains conversation history and provides contextual responses about the loaded knowledge base.
117
118
**Parameters:**
119
120
| Parameter | Type | Description |
121
|---|---|---|
122
| `message` | `str` | User message |
123
124
**Returns:** `str` -- assistant response.
125
126
The chat mode provides the LLM with:
127
128
- Knowledge base statistics (entity counts, relationship counts)
129
- List of previously generated artifacts
130
- Full conversation history
131
- Available REPL commands (e.g., `/entities`, `/search`, `/plan`, `/export`)
132
133
**Requires** a configured `provider_manager`. Returns a static error message if no LLM is available.
134
135
```python
136
response = agent.chat("What technologies were discussed in the meetings?")
137
print(response)
138
139
response = agent.chat("Which of those have the most dependencies?")
140
print(response)
141
```
142
143
---
144
145
## AgentContext
146
147
```python
148
from video_processor.agent.skills.base import AgentContext
149
```
150
151
Shared state dataclass passed to all skills during execution. Accumulates artifacts and conversation history across the agent session.
152
153
| Field | Type | Default | Description |
154
|---|---|---|---|
155
| `knowledge_graph` | `Any` | `None` | `KnowledgeGraph` instance |
156
| `query_engine` | `Any` | `None` | `GraphQueryEngine` instance for querying the KG |
157
| `provider_manager` | `Any` | `None` | `ProviderManager` instance for LLM calls |
158
| `planning_entities` | `List[Any]` | `[]` | Extracted `PlanningEntity` instances |
159
| `user_requirements` | `Dict[str, Any]` | `{}` | User-specified requirements and constraints |
160
| `conversation_history` | `List[Dict[str, str]]` | `[]` | Chat message history (`role`, `content` dicts) |
161
| `artifacts` | `List[Artifact]` | `[]` | Previously generated artifacts |
162
| `config` | `Dict[str, Any]` | `{}` | Additional configuration |
163
164
```python
165
from video_processor.agent.skills.base import AgentContext
166
167
context = AgentContext(
168
knowledge_graph=kg,
169
query_engine=engine,
170
provider_manager=pm,
171
config={"output_format": "markdown"},
172
)
173
```
174
175
---
176
177
## Skill (ABC)
178
179
```python
180
from video_processor.agent.skills.base import Skill
181
```
182
183
Base class for agent skills. Each skill represents a discrete capability that produces an artifact from the agent context.
184
185
**Class attributes:**
186
187
| Attribute | Type | Description |
188
|---|---|---|
189
| `name` | `str` | Skill identifier (e.g., `"project_plan"`, `"prd"`) |
190
| `description` | `str` | Human-readable description shown to the LLM for skill selection |
191
192
### execute()
193
194
```python
195
@abstractmethod
196
def execute(self, context: AgentContext, **kwargs) -> Artifact
197
```
198
199
Execute this skill and return an artifact. Receives the shared agent context and any parameters selected by the LLM planner.
200
201
### can_execute()
202
203
```python
204
def can_execute(self, context: AgentContext) -> bool
205
```
206
207
Check if this skill can execute given the current context. The default implementation requires both `knowledge_graph` and `provider_manager` to be set. Override for skills with different requirements.
208
209
**Returns:** `bool`
210
211
### Implementing a custom skill
212
213
```python
214
from video_processor.agent.skills.base import Skill, Artifact, AgentContext, register_skill
215
216
class SummarySkill(Skill):
217
name = "summary"
218
description = "Generate a concise summary of the knowledge base"
219
220
def execute(self, context: AgentContext, **kwargs) -> Artifact:
221
stats = context.query_engine.stats()
222
prompt = f"Summarize this knowledge base:\n{stats.to_text()}"
223
content = context.provider_manager.chat(
224
[{"role": "user", "content": prompt}]
225
)
226
return Artifact(
227
name="Knowledge Base Summary",
228
content=content,
229
artifact_type="document",
230
format="markdown",
231
)
232
233
def can_execute(self, context: AgentContext) -> bool:
234
return context.query_engine is not None and context.provider_manager is not None
235
236
# Register the skill so the agent can discover it
237
register_skill(SummarySkill())
238
```
239
240
---
241
242
## Artifact
243
244
```python
245
from video_processor.agent.skills.base import Artifact
246
```
247
248
Dataclass representing the output of a skill execution.
249
250
| Field | Type | Default | Description |
251
|---|---|---|---|
252
| `name` | `str` | *required* | Human-readable artifact name |
253
| `content` | `str` | *required* | Generated content (Markdown, JSON, Mermaid, etc.) |
254
| `artifact_type` | `str` | *required* | Type: `"project_plan"`, `"prd"`, `"roadmap"`, `"task_list"`, `"document"`, `"issues"` |
255
| `format` | `str` | `"markdown"` | Content format: `"markdown"`, `"json"`, `"mermaid"` |
256
| `metadata` | `Dict[str, Any]` | `{}` | Additional metadata |
257
258
---
259
260
## Skill Registry Functions
261
262
### register_skill()
263
264
```python
265
def register_skill(skill: Skill) -> None
266
```
267
268
Register a skill instance in the global registry. Skills must be registered before the agent can discover and execute them.
269
270
### get_skill()
271
272
```python
273
def get_skill(name: str) -> Optional[Skill]
274
```
275
276
Look up a registered skill by name.
277
278
**Returns:** `Optional[Skill]` -- the skill instance, or `None` if not found.
279
280
### list_skills()
281
282
```python
283
def list_skills() -> List[Skill]
284
```
285
286
Return all registered skill instances.
287
288
---
289
290
## KBContext
291
292
```python
293
from video_processor.agent.kb_context import KBContext
294
```
295
296
Loads and merges multiple knowledge graph sources into a unified context for agent consumption. Supports both FalkorDB (`.db`) and JSON (`.json`) formats, and can auto-discover graphs in a directory tree.
297
298
### Constructor
299
300
```python
301
def __init__(self)
302
```
303
304
Creates an empty context. Use `add_source()` to add knowledge graph paths, then `load()` to initialize.
305
306
### add_source()
307
308
```python
309
def add_source(self, path) -> None
310
```
311
312
Add a knowledge graph source.
313
314
**Parameters:**
315
316
| Parameter | Type | Description |
317
|---|---|---|
318
| `path` | `str \| Path` | Path to a `.db` file, `.json` file, or directory to search for knowledge graphs |
319
320
If `path` is a directory, it is searched recursively for knowledge graph files using `find_knowledge_graphs()`.
321
322
**Raises:** `FileNotFoundError` if the path does not exist.
323
324
### load()
325
326
```python
327
def load(self, provider_manager=None) -> KBContext
328
```
329
330
Load and merge all added sources into a single knowledge graph and query engine.
331
332
**Parameters:**
333
334
| Parameter | Type | Default | Description |
335
|---|---|---|---|
336
| `provider_manager` | `ProviderManager` | `None` | LLM provider for the knowledge graph and query engine |
337
338
**Returns:** `KBContext` -- self, for method chaining.
339
340
### Properties
341
342
| Property | Type | Description |
343
|---|---|---|
344
| `knowledge_graph` | `KnowledgeGraph` | The merged knowledge graph (raises `RuntimeError` if not loaded) |
345
| `query_engine` | `GraphQueryEngine` | Query engine for the merged graph (raises `RuntimeError` if not loaded) |
346
| `sources` | `List[Path]` | List of resolved source paths |
347
348
### summary()
349
350
```python
351
def summary(self) -> str
352
```
353
354
Generate a brief text summary of the loaded knowledge base, including entity counts by type and relationship counts.
355
356
**Returns:** `str` -- multi-line summary text.
357
358
### auto_discover()
359
360
```python
361
@classmethod
362
def auto_discover(
363
cls,
364
start_dir: Optional[Path] = None,
365
provider_manager=None,
366
) -> KBContext
367
```
368
369
Factory method that creates a `KBContext` by auto-discovering knowledge graphs near `start_dir` (defaults to current directory).
370
371
**Returns:** `KBContext` -- loaded context (may have zero sources if none found).
372
373
### Usage examples
374
375
```python
376
from pathlib import Path
377
from video_processor.agent.kb_context import KBContext
378
379
# Manual source management
380
kb = KBContext()
381
kb.add_source(Path("project_a/knowledge_graph.db"))
382
kb.add_source(Path("project_b/results/")) # searches directory
383
kb.load(provider_manager=pm)
384
385
print(kb.summary())
386
# Knowledge base: 3 source(s)
387
# Entities: 142
388
# Relationships: 89
389
# Entity types:
390
# technology: 45
391
# person: 23
392
# concept: 74
393
394
# Auto-discover from current directory
395
kb = KBContext.auto_discover()
396
397
# Use with the agent
398
from video_processor.agent.agent_loop import PlanningAgent
399
from video_processor.agent.skills.base import AgentContext
400
401
context = AgentContext(
402
knowledge_graph=kb.knowledge_graph,
403
query_engine=kb.query_engine,
404
provider_manager=pm,
405
)
406
agent = PlanningAgent(context)
407
```
408

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button