2 Commits

Author SHA1 Message Date
subochev 25771a0c33 docs(diagrams): agent architecture overview with pre-rendered SVG
ci / JVM build + tests (push) Failing after 1m57s
PlantUML diagrams for future agent architecture (Android, multi-user
chat, sub-agents, A2A):
- 01-module-layers.md — целевая модульная структура
- 02-agent-composition.md — AgentBuilder DSL + MemoryBackend.exposesTools()
- 03-multi-user-chat.md — mention-detection sequence
- 04-sub-agents.md — spawnChild + Flow<SubAgentEvent> + A2A
- 05-android-stack.md — что меняется на Android vs Standalone

Каждый .md включает пред-рендеренный SVG (показывается во всех markdown
viewers без PlantUML plugin) + PlantUML source в code block (для
редактирования). SVG нужен потому что PlantUML требует Graphviz dot
для рендеринга — без него IntelliJ/VSCode выдают ошибку.

Регенерация SVG после правки PlantUML-source:
  docker run --rm -v "$PWD:/work" plantuml/plantuml -tsvg /work/docs/diagrams/*.md
2026-09-18 20:02:41 +03:00
subochev 78cbe9b463 refactor(standalone): split ChatConversation into components
Decompose 1415-line god class into focused components:
  - ConversationState (shared mutable state)
  - ConversationEvents (SharedFlow + policy)
  - ContextBuilder (prefix/memory helpers)
  - CompactionCoordinator (compaction + LiteConv rebuild)
  - ToolDispatcher (single tool-call execution)
  - BackgroundScheduler (review/reflection/mining triggers)
  - ConversationLoop (orchestrator, implements ProtoConversation)

ChatConversation becomes a typealias. Public API preserved.
2026-09-18 03:00:24 +03:00
20 changed files with 1994 additions and 1352 deletions
+158
View File
@@ -0,0 +1,158 @@
# 01 — Слои модулей (целевое состояние)
Целевая модульная структура agentik. Снизу вверх:
**приложения → runtime → домен → абстракции → платформенные impl**.
![Module Layers](./01-module-layers.svg)
PlantUML source (для редактирования; требует Graphviz `dot` для рендеринга):
```plantuml
@startuml agentik-module-layers
skinparam componentStyle rectangle
skinparam ranksep 60
skinparam nodesep 30
skinparam packageStyle rectangle
title agentik — слои модулей (целевое состояние)
' --- Applications: entry points (thin wrappers) ---
package "Applications\n(entry points, тонкие)" {
[Standalone\nHTTP+AG-UI+A2A] as Standalone
[AgentikCli\nREPL] as Cli
[AgentikAndroid\nCompose UI] as Android
}
' --- Agent runtime ---
package "Agent Runtime\n(композиция, lifecycle)" {
[AgentCore\nBaseAgent] as AgentCore
[AgentBuilder\nDSL] as Builder
}
' --- Background work ---
package "Background Work\n(event-driven triggers)" {
[BackgroundEvents\nbus + events] as Ev
[BackgroundScheduler\npolicy] as Sched
}
' --- Domain logic (generic, переиспользуется) ---
package "Domain Logic\n(generic tools)" {
[LlmTools\nReflector/Reviewer/Miner] as LlmT
[McpBridge\nMCP-SDK → LiteTool] as Mcp
[Skills\nparse + store] as Skills
}
' --- Storage abstractions + impls ---
package "Storage\n(abstractions)" as StoragePkg {
[StorageCore\ninterfaces] as StorageCore
}
package "Storage\n(JVM impls)" {
[StorageSqlite\nJDBC] as StorageSql
[StorageInmemory\ntests] as StorageInmem
}
package "Storage\n(Android impl)" {
[StorageSqliteAndroid\nRoom/sqlite] as StorageSqlA
}
' --- Memory backends ---
package "Memory\n(abstractions)" {
[MemoryApi\nMemorySystem/MemoryTools] as MemApi
}
package "Memory\n(impls)" {
[MemoryMd\nHermes §-files] as MemMd
[MemoryVector\nJVector+JVM] as MemVec
[MemoryVectorAndroid\nONNX+ANN] as MemVecA
}
' --- LLM backends ---
package "LLM\n(abstractions)" {
[LitertApi\nLiteLlm контракт] as Litert
}
package "LLM\n(impls)" {
[LitertOpenai\nHTTP] as LitertO
[LitertGoogle\nLiteRT JVM] as LitertG
[LitertAndroid\nLiteRT Android] as LitertA
}
' --- Inter-app protocol ---
package "Inter-app" {
[Proto\nAgent/Conversation] as Proto
[A2AServer] as A2A
}
' --- Зависимости (приложения → runtime → домен → абстракции → платформенные импл) ---
Standalone ..> Builder
Cli ..> Builder
Android ..> Builder
Builder ..> AgentCore
AgentCore ..> Proto
AgentCore ..> StorageCore
AgentCore ..> MemApi
AgentCore ..> Litert
AgentCore ..> Mcp
AgentCore ..> Skills
Sched ..> Ev
AgentCore ..> Sched
AgentCore ..> Ev
Mcp ..> Litert
LlmT ..> Litert
MemMd ..> MemApi
MemVec ..> MemApi
MemVecA ..> MemApi
StorageSql ..> StorageCore
StorageInmem ..> StorageCore
StorageSqlA ..> StorageCore
LitertO ..> Litert
LitertG ..> Litert
LitertA ..> Litert
Standalone ..> A2A
Standalone ..> LitertO
Standalone ..> LitertG
Standalone ..> StorageSql
Standalone ..> MemMd
Standalone ..> MemVec
Standalone ..> Mcp
Android ..> LitertA
Android ..> StorageSqlA
Android ..> MemMd
Android ..> MemVecA
@enduml
```
## Что показывает
- **Applications** — три точки входа: web-сервер, CLI REPL, Android-приложение. Каждое тонкое, не содержит бизнес-логики.
- **Agent Runtime** — `BaseAgent` + `AgentBuilder` DSL. Вся композиция и lifecycle.
- **Background Work** — `BackgroundEvents` (event-bus) + `BackgroundScheduler` (policy подписки). Event-driven, не interval-polling.
- **Domain Logic** — generic переиспользуемые модули (`:llm-tools`, `:mcp-bridge`, `:skills`).
- **Storage / Memory / LLM** — каждая с абстракцией и одним или несколькими impl (JVM-only или Android-only).
- **Inter-app** — `:proto` контракты + `:a2a-server` для межагентного общения.
## Текущее состояние vs целевое
✅ Уже сделано (в этом цикле правок):
- `:llm-tools` extracted
- `:mcp-bridge` extracted
- `BackgroundScheduler` стал event-driven
- `ConversationLoop` стал отдельным компонентом (typealias `ChatConversation`)
⏳ Не сделано:
- `:agent-core` (выделить `BaseAgent` + builder в отдельный KMP-модуль)
- `:background-events` (выделить events + scheduler — пока в `:standalone`)
- `:storage-sqlite-android`
- `:memory-vector-android`
- `:litert-android`
- `:agentik-android` (само приложение)
File diff suppressed because one or more lines are too long

After

Width:  |  Height:  |  Size: 38 KiB

+127
View File
@@ -0,0 +1,127 @@
# 02 — Agent Builder: композиция (целевое API)
Как `AgentBuilder` собирает `BaseAgent` из компонентов. **Memory backend сам объявляет свои tools** — builder их авто-мержит. BackgroundScheduler подписан на события, не interval-poll.
![Agent Composition](./02-agent-composition.svg)
PlantUML source (для редактирования; требует Graphviz `dot` для рендеринга):
```plantuml
@startuml agent-composition
skinparam componentStyle rectangle
title Agent Builder — композиция (целевое API)
' --- Builder ---
rectangle "AgentBuilder" as Builder {
rectangle "llm: LiteLlm (обязательно)" as Llm
rectangle "storage: StorageBundle (обязательно)" as Storage
rectangle "memory: MemorySystem (обязательно)" as Mem
rectangle "soul: SoulProvider (default NoopSoul)" as Soul
rectangle "tools: List<NamedTool> (авто-сборка из backends)" as Tools
rectangle "background: BackgroundConfig (default EmptyBg)" as Bg
rectangle "toolset: List<ToolsetContribution> (default empty)" as Ts
}
' --- Backends with their tool side-effects ---
rectangle "MemoryMd" as MdMem {
interface "MemorySystem" as MemSys
interface "List<NamedTool>" as MdTools
note right
MemoryMd.exposesTools() →
memory_save / memory_read /
memory_list / memory_delete
end note
}
rectangle "McpRegistry" as McpReg {
interface "List<NamedTool>" as McpTools
note right
McpRegistry.namedTools →
server__tool1, server__tool2,
...
end note
}
rectangle "BackgroundEvents" as Events {
interface "MutableSharedFlow<CompactionEvent|ToolCallEvent|LifecycleEvent>" as Flow
note right
Эмитится из:
- CompactionCoordinator
- ToolDispatcher
- ConversationLoop.close()
end note
}
rectangle "BackgroundScheduler" as Sched {
interface "policy: trigger + debounce" as Policy
note right
Подписан на Events.
НИКАКОГО interval-polling.
end note
}
' --- Получаемый Agent ---
rectangle "BaseAgent\n(impl: ConversationLoop)" as Agent {
rectangle "send / interrupt / events" as API
rectangle "BackgroundScheduler\nподписка" as Sub
}
' --- Стрелки зависимостей ---
Builder --> Llm
Builder --> Storage
Builder --> Mem
Builder --> Soul
Builder --> Tools
Builder --> Bg
Builder --> Ts
MdMem --> Mem : implements
MdMem --> MdTools : exposes
McpReg --> McpTools : exposes
Bg --> Events : subscribes-to
Bg --> Sched : holds
Tools <-- MdTools : auto-merge
Tools <-- McpTools : auto-merge
Builder --> Agent : build()
Agent --> API
Agent --> Sub
@enduml
```
## Целевой Kotlin DSL
```kotlin
val agent = agentBuilder {
// Обязательные
llm(OpenAiLlm.fromEnv()) // или LitertAndroid.onDevice(context)
storage(SqliteStorage(path)) // или SqliteStorage.android(context)
memory(MemoryMd(root = "~/memory")) // или MemoryVector(embedding = HttpEmbedding(...))
// Опциональные
soul(FileSoul("~/SOUL.md")) // или HttpSoul(url), NoopSoul()
background {
// triggers: OnClosing (reflection+mining), OnCompaction(minTurns=10, mining=true)
// event-driven, не interval
}
tools {
// memoryMd.exposesTools() + mcpRegistry.namedTools авто-подцепляются
+FileReadTool(root = "/data")
}
toolset {
+MemoryToolsToolset(memoryMd)
}
}.build()
```
## Ключевые решения
- **`MemoryBackend.exposesTools()`** — backend декларирует свои tools. Не «подставить любой backend», а «backend сообщает что он умеет». Это убирает coupling «какие tools совместимы с какими backends».
- **Builder требует только `llm + storage + memory`** как обязательные. Всё остальное — опционально с разумными default'ами (`NoopSoul`, `EmptyBackground`, `empty toolset`).
- **`BaseAgent`** — реализация `ConversationLoop` через builder. Конструктор принимает все нужные компоненты. **Один и тот же `BaseAgent` в `:standalone`, `:agentik-cli`, `:agentik-android`** — отличается только wiring через builder.
- **BackgroundScheduler подписан на `BackgroundEvents`** — это даёт event-driven по умолчанию. `OnEvery(n)` interval-режим — опциональный fallback (не default).
File diff suppressed because one or more lines are too long

After

Width:  |  Height:  |  Size: 24 KiB

+129
View File
@@ -0,0 +1,129 @@
# 03 — Multi-user chat с mention-detection
Точка 1 из планов. Один `BaseAgent` обслуживает N пользователей. Отвечает только когда addressed (mention или admin-команда).
![Multi-user chat](./03-multi-user-chat.svg)
PlantUML source (для редактирования; требует Graphviz `dot` для рендеринга):
```plantuml
@startuml multi-user-chat
skinparam componentStyle rectangle
skinparam participantPadding 15
skinparam boxPadding 10
title Multi-user chat с mention-detection (точка 1 из планов)
' --- Участники ---
actor "User A" as UA
actor "User B" as UB
actor "User C\n(админ)" as UC
participant "Telegram /\nSlack /\nMatrix" as Channel
participant "AgentRuntime\n(BaseAgent)" as Runtime
participant "MentionDetector" as Detector
participant "SoulProvider" as Soul
participant "MemorySystem\n(MdMemory)" as Memory
participant "LlmBackend\n(LiteLlm)" as Llm
' --- Сценарий ---
UA -> Channel : "@bot, что нового?"
UB -> Channel : "люблю котов"
UC -> Channel : "/bot status"
Channel -> Runtime : событие чата
' --- Внутри Runtime ---
Runtime -> Detector : isMentioned(message, botName)
note right of Detector
variants:
- SimpleMentionDetector (regex: @bot)
- LlmMentionDetector (mini-classifier)
- AdminCommandDetector (/command)
end note
Detector --> Runtime : MatchResult{isMentioned, isCommand}
alt isMentioned или isCommand
Runtime -> Soul : read()
Runtime -> Memory : prefetch(query, topK)
Runtime -> Llm : send(system + history + memory + user)
Llm --> Runtime : response + tool_calls
Runtime -> Memory : save(decision)
Runtime --> Channel : ответ в нужный канал/thread
else NOT mentioned и NOT command
Runtime -> Runtime : drop (если не админ)
note right
Не отвечаем, но возможно:
- запоминаем факт (memory-only update)
- summary на long conversation
end note
end
@enduml
```
## Ключевые модули (что нужно будет добавить)
### `MentionDetector` interface
```kotlin
interface MentionDetector {
data class Result(
val isMentioned: Boolean,
val isAdminCommand: Boolean,
val isPrivateMessage: Boolean, // DM — всегда отвечаем
)
fun detect(message: ChatMessage, botName: String): Result
}
```
Имплементации:
- `SimpleMentionDetector` — regex `@bot`, `/command` (дешёво, latency ~0)
- `LlmMentionDetector` — маленькая классификация через тот же LLM (точнее, но +1 LLM-вызов на каждое сообщение)
- `HybridMentionDetector` — fast regex → fallback на LLM только если ambiguous
### `ChatAdapter` interface
```kotlin
interface ChatAdapter {
val channel: String // "telegram" / "slack" / "matrix"
suspend fun listen(onMessage: (ChatMessage) -> Unit): Job
suspend fun reply(messageId: String, text: String, threadId: String? = null)
suspend fun isAdmin(userId: String): Boolean
}
```
Имплементации per platform. Каждая адаптирует формат platform → `ChatMessage`.
### Конфигурация builder'а
```kotlin
agentBuilder {
llm(...)
storage(...)
memory(...)
soul(...)
background { ... }
chat {
mentionDetector = HybridMentionDetector(regex = "@bot|@Agent", llmClassifier = false)
chatAdapter = TelegramChatAdapter(token = "...")
// На каждое сообщение:
// 1. mentionDetector.detect()
// 2. если isMentioned || isAdminCommand || isPrivate → process
// 3. иначе — опционально memory-only save (тихий режим)
}
}
```
## Что это даёт
- Один `BaseAgent` обслуживает чат целиком (один LLM, одна память — общий контекст команды)
- `@bot` — explicit invocation, не «agent отвечает на всё подряд»
- `/bot status` / `/bot clear-memory` — admin-команды (отдельный канал, без LLM)
- DM — всегда отвечает (это личное обращение)
- В групповом чате без mention — agent может **молча учить** (memory update без ответа). Полезно для «запомнил что Вася любит котов».
## Текущее состояние vs целевое
⏳ Ничего из этого нет. Сейчас `:standalone` — это HTTP API, к которому подключаются clients. Для multi-user chat нужен новый `:chat-adapter-telegram` (или -slack / -matrix) модуль + `MentionDetector` interface.
File diff suppressed because one or more lines are too long

After

Width:  |  Height:  |  Size: 18 KiB

+119
View File
@@ -0,0 +1,119 @@
# 04 — Sub-agents + A2A между агентами
Точки 2 и 3 из планов. Orchestrator-агент spawn'ит sub-агентов с изолированным контекстом. Независимые агенты общаются через A2A.
![Sub-agents + A2A](./04-sub-agents.svg)
PlantUML source (для редактирования; требует Graphviz `dot` для рендеринга):
```plantuml
@startuml sub-agents-and-a2a
skinparam componentStyle rectangle
title Sub-agents + A2A между агентами (точка 2+3 из планов)
' --- Orchestrator ---
rectangle "OrchestratorAgent\n(BaseAgent + tools)" as Orch {
rectangle "ConversationLoop\n(main user)" as MainConv
}
' --- Sub-agent spawn ---
rectangle "subAgent(\n task: String,\n config: AgentConfig\n): Flow<SubAgentEvent>" as SpawnAPI
note right of SpawnAPI
Spawn API — НЕ отдельный модуль,
а convenience поверх BaseAgent:
val sub = agent.spawnChild(config) {
systemPrompt = "..."
tools = [ReadTool, WriteTool]
memory = EmptyMemory // изолированно
}
sub.events.collect { ... }
end note
' --- Дочерний агент (изолированный контекст) ---
rectangle "SubAgent\n(изолированный scope)" as Sub {
rectangle "ConversationLoop\n(child)" as SubConv
rectangle "backgroundScope\n(lifecycle scoped)" as SubBg
}
' --- A2A между независимыми агентами ---
rectangle "Agent A\n(BaseAgent)" as AgentA
rectangle "Agent B\n(BaseAgent)" as AgentB
rectangle "A2A Server\n(:a2a-server)" as A2ASrv
AgentA -> A2ASrv : POST /\n(application/json)
A2ASrv -> AgentB : dispatch(message)
AgentB --> A2ASrv : response
A2ASrv --> AgentA : SSE / JSON-RPC
' --- Стрелки ---
Orch -> SpawnAPI : calls
SpawnAPI -> Sub : creates with custom config
Sub -> SubBg : has its own
Orch -> Orch : main flow continues
Sub --> Orch : Flow<SubAgentEvent> emits\n(Started / ToolCalled / ToolResult /\nAssistantMessage / Done / Failed)
Orch -> A2ASrv : can also delegate to remote agent
@enduml
```
## Sub-agents API
```kotlin
sealed interface SubAgentEvent {
data class Started(val taskId: String) : SubAgentEvent
data class AssistantMessage(val text: String) : SubAgentEvent
data class ToolCalled(val toolName: String, val args: JsonObject) : SubAgentEvent
data class ToolResult(val toolName: String, val result: String) : SubAgentEvent
data class Done(val taskId: String, val finalResult: String) : SubAgentEvent
data class Failed(val taskId: String, val error: String) : SubAgentEvent
}
interface BaseAgent {
// ... existing methods ...
/**
* Spawn дочерний агент с изолированным контекстом (memory, system prompt,
* tools). Возвращает Flow событий жизненного цикла + результата.
* Cancellation родителя НЕ отменяет sub-agent — sub-agent живёт до Done/Failed.
*/
fun spawnChild(config: SubAgentConfig): Flow<SubAgentEvent>
}
data class SubAgentConfig(
val systemPrompt: String,
val tools: List<NamedTool> = emptyList(),
val memory: MemorySystem = EmptyMemory(),
val model: LiteLlm? = null, // если null — делит LLM родителя
val maxTurns: Int = 10,
val timeoutMs: Long = 60_000,
)
```
## Зачем изолированный scope
Sub-agent получает **свою копию контекста**, не делит memory с родителем. Это критично:
- `research_subagent` — должен исследовать тему, не отвечать на основные сообщения пользователя
- `summarize_subagent` — суммаризировать документ, не трогать основной диалог
- `code_review_subagent` — ревьюить PR, не видеть разговор
Если нужно расшарить контекст — это explicit через `sharedMemory: SharedMemoryHandle` параметр, не default.
## A2A между независимыми агентами
Уже есть `:a2a-server` модуль (см. `standalone/build.gradle.kts` — `implementation(libs.a2a.server)`). Использовался для AG-UI/A2A протокола в `:standalone`. Можно переиспользовать для межагентного общения.
Сценарий: orchestrator-agent не может сам решить задачу → делегирует remote-агенту через A2A → получает response → продолжает. Это уже работающая инфраструктура.
## Текущее состояние vs целевое
✅ Уже есть:
- `:a2a-server` подключён
- `BaseAgent.spawnChild` — **не существует**, но `ConversationLoop` уже умеет создавать изолированный scope через свой `agentScope` — нужна только обёртка
⏳ Не сделано:
- `SubAgentConfig` + `Flow<SubAgentEvent>` API
- `EmptyMemory` (null-object для изолированного scope)
- Lifecycle management (parent dies → child должен complete or be cancelled?)
- Сериализация sub-agent state для отладки (event log)
File diff suppressed because one or more lines are too long

After

Width:  |  Height:  |  Size: 14 KiB

+171
View File
@@ -0,0 +1,171 @@
# 05 — Android Agent Stack
Что меняется vs `:standalone`. Цель: `BaseAgent` тот же самый, но platform-impl разные (Storage, LLM, Vector Memory, MCP).
![Android Agent Stack](./05-android-stack.svg)
PlantUML source (для редактирования; требует Graphviz `dot` для рендеринга):
```plantuml
@startuml android-agent-stack
skinparam componentStyle rectangle
title Android Agent Stack — что меняется vs Standalone
' --- Android side ---
package "Android Application" {
[MainActivity\n(Compose)] as Activity
[AndroidAgentRunner\n(workmanager / service)] as Runner
[AndroidAgentBuilder] as AndroidBuilder
}
package "Android-specific impls" {
[StorageSqliteAndroid\n(Room/sqlite)] as StorageA
[MemoryVectorAndroid\n(ONNX runtime + ANN)] as MemVecA
[LitertAndroid\n(NNAPI delegate)] as LitertA
[SoulFileAndroid\n(context.filesDir)] as SoulA
[McpRegistry\nstdio: ProcessBuilder] as McpA
}
' --- Shared (KMP) ---
package "Agent Runtime (shared)" {
[AgentCore\nBaseAgent] as AgentCore
[AgentBuilder] as Builder
}
package "Domain (shared)" {
[LlmTools\ncommonMain] as LlmT
[BackgroundEvents\ncommonMain] as Ev
[McpBridge\njvmMain] as McpB
[Skills\ncommonMain] as Skills
}
package "Memory (shared impl)" {
[MemoryMd\n(commonMain)] as MemMd
[MemoryApi\ninterfaces] as MemApi
}
' --- Зависимости ---
Activity --> Runner
Runner --> AndroidBuilder
AndroidBuilder --> AgentCore
AndroidBuilder --> StorageA
AndroidBuilder --> MemVecA
AndroidBuilder --> LitertA
AndroidBuilder --> SoulA
AndroidBuilder --> McpA
AgentCore --> LlmT
AgentCore --> Ev
AgentCore --> McpB
AgentCore --> Skills
AgentCore --> MemMd
' --- Главные отличия от Standalone ---
note right of LitertA
On-device inference.
LiteRT с NNAPI delegate →
работает на CPU/GPU/NPU
прямо на устройстве, без сети.
vs Standalone: HTTP-only
(OpenAI-compatible).
end note
note right of StorageA
android.database.sqlite
через Room или сырой API.
vs Standalone: JDBC +
Sqlite-JDBC driver
(только JVM).
end note
note right of MemVecA
JVector JVM-only. На Android
нужна альтернатива —
ONNX Runtime + какой-нибудь
ANN (Annoy/HNSW).
Или пока без vector memory,
только MemoryMd.
end note
note right of McpA
MCP через ProcessBuilder
на Android работает, но
subprocess lifecycle
сложнее (foreground service
нужен для долгого subprocess).
end note
@enduml
```
## Что общего с `:standalone`
**`BaseAgent`, `BackgroundScheduler`, `LlmTools`, `McpBridge`, `Skills`, `MemoryMd` — всё KMP (commonMain).** Android-agent = `:standalone` с другим wiring'ом. Не нужно переписывать agent logic.
## Что другое
| Компонент | `:standalone` (JVM) | `:agentik-android` (Android) | Сложность |
|---|---|---|---|
| Storage | `:storage-sqlite` (JDBC + Sqlite-JDBC) | `:storage-sqlite-android` (Room или raw) | Низкая — тот же `StorageBundle` interface |
| LLM | `:litert-openai` (HTTP), `:litert-google` (LiteRT JVM) | `:litert-android` (LiteRT Android, NNAPI delegate) | Средняя — нужен новый модуль |
| Vector memory | `:memory-vector` (JVector) | `:memory-vector-android` (ONNX Runtime + HNSW/Annoy) | Высокая — JVector JVM-only, нужна альтернатива |
| SOUL provider | `FileSoulProvider` (path) | `SoulFileAndroid` (`context.filesDir`) | Низкая |
| MCP | `McpRegistry` (ProcessBuilder, stdio subprocess) | Тот же `McpRegistry`, но subprocess в foreground service | Средняя — нужен Android service |
| Embedding | `HttpEmbeddingClient` (HTTP) | Тот же ИЛИ on-device (ONNX) | Средняя |
## Минимальный Android agent (v1)
Если не нужны все фичи сразу — минимум:
```kotlin
val agent = androidAgentBuilder(context) {
llm(LitertAndroid.onDevice(context, modelPath = "/data/local/tmp/model.litertlm"))
storage(SqliteStorage.android(context, "agent.db"))
memory(MemoryMd.root(context.filesDir.resolve("memory")))
soul(FileSoul(context.filesDir.resolve("SOUL.md")))
background {
// OnClosing + OnCompaction работают так же как на JVM
}
}
```
Без MCP, без vector memory (только MemoryMd на файлах), только on-device LLM. Достаточно для off-line агента.
## Foreground service для MCP
Если нужны MCP-серверы (например, локальный file-system MCP) — subprocess нужен foreground service чтобы Android не убил его при выключении экрана. Это добавляет сложности:
```kotlin
class McpForegroundService : Service() {
override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
startForeground(NOTIFICATION_ID, notification)
val proc = ProcessBuilder(command, args).start()
// ... route stdio to McpLiteToolAdapter ...
return START_STICKY
}
}
```
Пока можно без этого (только если MCP нужен на Android).
## Текущее состояние vs целевое
✅ KMP-ready:
- `:llm-tools` (commonMain, платформо-агностик)
- `:mcp-bridge` (jvmMain — Android-вариант через `:mcp-bridge-android`)
- `:skills` (commonMain)
- `:memory-md` (commonMain)
- `:proto` (commonMain)
⏳ Не существует:
- `:storage-sqlite-android`
- `:memory-vector-android`
- `:litert-android`
- `:agentik-android` (само приложение)
- `:agent-core` (выделить BaseAgent + builder)
- `:background-events` (выделить events + scheduler)
File diff suppressed because one or more lines are too long

After

Width:  |  Height:  |  Size: 25 KiB

+64
View File
@@ -0,0 +1,64 @@
# agentik — диаграммы архитектуры
PlantUML-схемы для обсуждения будущей структуры (Android agent, multi-user chat, sub-agents, A2A). Это **целевое состояние**, не текущее.
## Файлы
Каждый `.md` содержит:
- Краткое описание (что показывает)
- **Пред-рендеренный SVG** (`![Diagram](file.svg)`) — гарантированно показывается **везде**
- PlantUML source в ` ```plantuml ` блоке — для редактирования (требует Graphviz `dot` для рендеринга)
- Дополнительный markdown-текст (что нужно сделать, текущее vs целевое)
| Файл | Что показывает |
|---|---|
| [01-module-layers.md](./01-module-layers.md) | Целевая модульная структура (приложения → runtime → домен → абстракции → платформенные impl). Что в каком слое и кто от кого зависит. |
| [02-agent-composition.md](./02-agent-composition.md) | Как `AgentBuilder` собирает `BaseAgent` из компонентов. Memory backend сам объявляет свои tools. BackgroundScheduler подписан на события (НЕ interval-poll). |
| [03-multi-user-chat.md](./03-multi-user-chat.md) | Сценарий: чат с N пользователями, mention-detection, админ-команды, agent отвечает только когда addressed. |
| [04-sub-agents.md](./04-sub-agents.md) | Orchestrator spawn'ит sub-agent с изолированным контекстом, получает `Flow<SubAgentEvent>`. A2A между независимыми агентами через `:a2a-server`. |
| [05-android-stack.md](./05-android-stack.md) | Что меняется на Android: on-device LLM (NNAPI), Room/sqlite, ONNX-based vector memory, foreground-service для MCP subprocess. |
## Почему SVG + PlantUML source
PlantUML требует Java + (для component/class/deployment диаграмм) Graphviz `dot`. Если `dot` не установлен — рендерер падает с ошибкой "Executable dot does not exist".
Решение: **пре-рендерим в SVG один раз** и вставляем как `<img>`. Диаграмма гарантированно показывается в любом markdown-viewer (GitHub, IntelliJ, VSCode, GitLab) без зависимостей. PlantUML source в code block остаётся для редактирования.
## Как редактировать диаграмму
1. Меняешь PlantUML-source в ` ```plantuml ` блоке `.md` файла.
2. Ре-рендеришь SVG:
```bash
mkdir -p /tmp/plantuml-work && chmod 777 /tmp/plantuml-work
cp docs/diagrams/*.md /tmp/plantuml-work/
docker run --rm -v /tmp/plantuml-work:/work plantuml/plantuml -tsvg /work/*.md
cp /tmp/plantuml-work/*.svg docs/diagrams/
```
3. Проверяешь что SVG обновился:
```bash
ls -la docs/diagrams/*.svg
```
4. Коммитишь оба файла: `.md` (source) и `.svg` (rendered).
Требует Docker (или локального PlantUML+Graphviz). `apt install graphviz` для Arch/Manjaro.
## Контекст
Текущий код движется в эту сторону:
- `:llm-tools` extracted ✅
- `:mcp-bridge` extracted ✅
- `BackgroundScheduler` стал event-driven ✅
- `ConversationLoop` стал отдельным компонентом ✅
Не сделано (см. детали в каждом .md):
- `:agent-core` (выделить `BaseAgent` + builder)
- `:background-events` (выделить events + scheduler)
- `:storage-sqlite-android`, `:memory-vector-android`, `:litert-android`
- `:agentik-android` (само приложение)
- `MentionDetector` interface + adapters для multi-user chat
- `BaseAgent.spawnChild` + `Flow<SubAgentEvent>`
Подробнее:
- `STANDALONE-REVIEW.md` — что плохо в текущем коде
- `MEMORY-DESIGN.md` — детали memory архитектуры
- `STANDALONE.md` — текущий standalone
@@ -0,0 +1,170 @@
package pw.binom.agentik.standalone.agent
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import mu.KotlinLogging
import pw.binom.agentik.memory.ConversationTurn
import pw.binom.agentik.memory.MemoryReviewDecision
import pw.binom.agentik.memory.MemoryReviewer
import pw.binom.agentik.memory.MemoryStore
import pw.binom.agentik.memory.ReviewedTurn
import pw.binom.agentik.skills.SkillStore
import pw.binom.agentik.storage.Content
import pw.binom.agentik.storage.MessageRecord
import pw.binom.agentik.storage.ReflectionStore
import pw.binom.agentik.storage.WorkingMemoryEntry
import pw.binom.agentik.storage.WorkingMemoryStore
import pw.binom.agentik.standalone.agent.memory.materializeReviewNote
internal data class BackgroundConfig(
val memoryReviewer: MemoryReviewer?,
val memoryStore: MemoryStore?,
val memoryReviewInterval: Int = 0,
val reflectionStore: ReflectionStore?,
val reflector: LlmReflector?,
val reflectionInterval: Int,
val skillMiner: SkillMiner?,
val skillMiningStore: SkillStore?,
val skillMiningInterval: Int,
)
internal class BackgroundScheduler(
private val state: ConversationState,
private val workingMemory: WorkingMemoryStore,
private val config: BackgroundConfig,
) {
private val log = KotlinLogging.logger {}
fun maybeScheduleReview(
userRecord: MessageRecord.UserMessage,
assistantContent: List<Content>,
) {
val reviewer = config.memoryReviewer ?: return
val store = config.memoryStore ?: return
if (state.isTemporal) return
if (config.memoryReviewInterval > 0) {
val userTurnCount = countUserTurnsBlocking()
if (userTurnCount % config.memoryReviewInterval != 0) return
}
val userText = userRecord.content.filterIsInstance<Content.Text>()
.joinToString("\n") { it.body }
val assistantText = assistantContent.filterIsInstance<Content.Text>()
.joinToString("\n") { it.body }
if (userText.isBlank() || assistantText.isBlank()) return
val convId = state.id
state.agentScope.launch {
try {
val decision: MemoryReviewDecision = reviewer.review(
ReviewedTurn(
userMessage = userText,
assistantMessage = assistantText,
conversationId = convId,
),
)
for (n in decision.toSave) {
val note = materializeReviewNote(n, conversationId = null)
runCatching { store.upsert(note) }
.onFailure { log.warn(it) { "review upsert failed: ${it.message}" } }
}
for (id in decision.toDelete) {
runCatching { store.delete(id) }
.onFailure { log.warn(it) { "review delete failed: ${it.message}" } }
}
} catch (e: Throwable) {
log.warn(e) { "review failed for $convId: ${e.message}" }
}
}
}
fun maybeScheduleReflection(
userRecord: MessageRecord.UserMessage,
assistantContent: List<Content>,
) {
if (config.reflectionInterval <= 0) return
val reflector = config.reflector ?: return
val store = config.reflectionStore ?: return
if (state.isTemporal) return
val userText = userRecord.content.filterIsInstance<Content.Text>()
.joinToString("\n") { it.body }
val assistantText = assistantContent.filterIsInstance<Content.Text>()
.joinToString("\n") { it.body }
if (userText.isBlank() || assistantText.isBlank()) return
val userTurnCount = countUserTurnsBlocking()
if (userTurnCount % config.reflectionInterval != 0) return
val convId = state.id
state.agentScope.launch {
try {
val turns = listOf(
ConversationTurn(
userMessage = userText,
assistantMessage = assistantText,
)
)
val reflection = reflector.reflect(turns) ?: return@launch
val stamped = reflection.copy(conversationId = convId)
runCatching { store.insert(stamped) }
.onFailure { log.warn(it) { "reflection insert failed: ${it.message}" } }
log.info { "self-reflection score=${stamped.score}/5 conv=$convId spots=${stamped.weakSpots.size}" }
} catch (e: Throwable) {
log.warn(e) { "reflection failed for $convId: ${e.message}" }
}
}
}
fun maybeScheduleSkillMining(
userRecord: MessageRecord.UserMessage,
assistantContent: List<Content>,
) {
if (config.skillMiningInterval <= 0) return
val miner = config.skillMiner ?: return
val store = config.skillMiningStore ?: return
if (state.isTemporal) return
val userTurnCount = countUserTurnsBlocking()
if (userTurnCount % config.skillMiningInterval != 0) return
val convId = state.id
state.agentScope.launch {
try {
val turns = recentTurnsFromWorkingMemory(miner.maxTurns)
if (turns.isEmpty()) return@launch
val existing = store.catalog.skills
val mined = miner.mine(turns, existing)
for (s in mined) {
runCatching { store.upsert(s) }
.onFailure { log.warn(it) { "skill-mine upsert '${s.name}' failed: ${it.message}" } }
}
log.info { "skill-mine: conv=$convId turns=${turns.size} existing=${existing.size} mined=${mined.size}" }
} catch (e: Throwable) {
log.warn(e) { "skill-mine failed for $convId: ${e.message}" }
}
}
}
private fun countUserTurnsBlocking(): Int = runBlocking {
var count = 0
for (row in workingMemory.list(state.id)) {
if (row.entry is WorkingMemoryEntry.User) count++
}
count
}
private suspend fun recentTurnsFromWorkingMemory(maxTurns: Int): List<ConversationTurn> {
val rows = workingMemory.list(state.id)
val pairs = mutableListOf<ConversationTurn>()
var pendingUser: String? = null
for (row in rows) {
when (val e = row.entry) {
is WorkingMemoryEntry.User -> pendingUser = e.content.text()
is WorkingMemoryEntry.Assistant -> {
val user = pendingUser ?: ""
pendingUser = null
pairs += ConversationTurn(userMessage = user, assistantMessage = e.content.text())
}
else -> {}
}
}
return pairs.takeLast(maxTurns)
}
private fun List<Content>.text(): String =
filterIsInstance<Content.Text>().joinToString("\n") { it.body }
}
@@ -66,6 +66,14 @@ class ChatAgent(
private val memoryStore: pw.binom.agentik.memory.MemoryStore? = null, private val memoryStore: pw.binom.agentik.memory.MemoryStore? = null,
private val memoryPrefetcher: MemoryPrefetcher? = null, private val memoryPrefetcher: MemoryPrefetcher? = null,
private val memoryReviewer: MemoryReviewer? = null, private val memoryReviewer: MemoryReviewer? = null,
/**
* Через сколько пользовательских ходов запускать LLM-based memory review
* (см. [pw.binom.agentik.standalone.agent.LlmMemoryReviewer]). `0` —
* review выключен. Default: 0 (для безопасности — старый код без
* interval-gate приводил к ×2 LLM-call amplification, и [Main.kt] явно
* передаёт config.memoryReviewInterval).
*/
private val memoryReviewInterval: Int = 0,
/** /**
* Тело SOUL.md — markdown-описание персоны. Вставляется в самое начало * Тело SOUL.md — markdown-описание персоны. Вставляется в самое начало
* системного промпта, поверх базы, навыков и memory-guidance. `null` — * системного промпта, поверх базы, навыков и memory-guidance. `null` —
@@ -234,6 +242,7 @@ class ChatAgent(
memoryPrefetcher = memoryPrefetcher, memoryPrefetcher = memoryPrefetcher,
memoryReviewer = memoryReviewer, memoryReviewer = memoryReviewer,
memoryStoreForReview = memoryStore, memoryStoreForReview = memoryStore,
memoryReviewInterval = memoryReviewInterval,
contextWindow = contextWindow, contextWindow = contextWindow,
compressionThreshold = compressionThreshold, compressionThreshold = compressionThreshold,
contextCompactor = contextCompactor, contextCompactor = contextCompactor,
@@ -285,6 +294,7 @@ class ChatAgent(
memoryPrefetcher = memoryPrefetcher, memoryPrefetcher = memoryPrefetcher,
memoryReviewer = memoryReviewer, memoryReviewer = memoryReviewer,
memoryStoreForReview = memoryStore, memoryStoreForReview = memoryStore,
memoryReviewInterval = memoryReviewInterval,
contextWindow = contextWindow, contextWindow = contextWindow,
compressionThreshold = compressionThreshold, compressionThreshold = compressionThreshold,
contextCompactor = contextCompactor, contextCompactor = contextCompactor,
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,236 @@
package pw.binom.agentik.standalone.agent
import kotlinx.coroutines.CancellationException
import mu.KotlinLogging
import pw.binom.agentik.memory.ConversationTurn
import pw.binom.agentik.memory.MemoryReviewer
import pw.binom.agentik.memory.MemoryStore
import pw.binom.agentik.standalone.agent.memory.materializeReviewNote
import pw.binom.agentik.storage.Content
import pw.binom.agentik.storage.WorkingMemoryEntry
import pw.binom.agentik.storage.WorkingMemoryRow
import pw.binom.agentik.storage.WorkingMemoryStore
import pw.binom.litert.LiteContentPart
import pw.binom.litert.LiteConversation
import pw.binom.litert.LiteConversationConfig
import pw.binom.litert.LiteLlm
import pw.binom.litert.LiteMessage
import pw.binom.litert.LiteRole
internal class CompactionCoordinator(
private val state: ConversationState,
private val contextWindow: Int?,
private val compressionThreshold: Double,
private val contextCompactor: ContextCompactor?,
private val memoryReviewer: MemoryReviewer?,
private val memoryStoreForReview: MemoryStore?,
private val workingMemory: WorkingMemoryStore,
private val liteLlm: LiteLlm,
private val systemPrompt: String,
) {
private val log = KotlinLogging.logger {}
private val toolsCharsCached: Int by lazy(LazyThreadSafetyMode.PUBLICATION) {
state.tools.sumOf { it.tool.describe().length }
}
suspend fun compactPreTurnIfNeeded(): Boolean = compactPreTurn(force = false)
suspend fun forceCompactNow(): Boolean = compactPreTurn(force = true)
private suspend fun compactPreTurn(force: Boolean): Boolean {
val window = contextWindow ?: return false
val compactor = contextCompactor ?: return false
val wm = workingMemory.list(state.id)
if (wm.isEmpty()) return false
val systemText = systemPrompt
val history = wm.filter { it.entry is WorkingMemoryEntry.User || it.entry is WorkingMemoryEntry.Assistant }
val toolsChars = toolsCharsCached
val estimated = estimateTokens(
systemText = systemText,
history = history,
toolsChars = toolsChars,
)
if (!force && estimated.toDouble() / window < compressionThreshold) return false
val toCompact = if (history.size > KEEP_RECENT_TURNS) {
history.dropLast(KEEP_RECENT_TURNS)
} else {
history
}
if (toCompact.isEmpty()) return false
val turns = toCompact.mapNotNull { row ->
when (val e = row.entry) {
is WorkingMemoryEntry.User -> SummaryTurn(
userMessage = e.content.text(),
assistantMessage = "",
createdAt = row.createdAt,
)
is WorkingMemoryEntry.Assistant -> SummaryTurn(
userMessage = "",
assistantMessage = e.content.text(),
createdAt = row.createdAt,
)
else -> null
}
}
val paired = ArrayList<SummaryTurn>()
var pendingUser: SummaryTurn? = null
for (t in turns) {
if (t.userMessage.isNotBlank()) {
if (pendingUser != null) paired.add(pendingUser)
pendingUser = t
} else if (t.assistantMessage.isNotBlank() && pendingUser != null) {
paired.add(pendingUser.copy(assistantMessage = t.assistantMessage))
pendingUser = null
} else if (t.assistantMessage.isNotBlank()) {
paired.add(t)
}
}
if (pendingUser != null) paired.add(pendingUser)
if (paired.isEmpty()) {
log.info { "compactPreTurn: nothing to compact for ${state.id}" }
return false
}
val summaryText = try {
compactor.summarize(paired)
} catch (e: CancellationException) {
throw e
} catch (e: Throwable) {
log.warn(e) { "context summarization failed for ${state.id}: ${e.message}" }
return false
}
if (summaryText.isBlank()) return false
val reviewer = memoryReviewer
val store = memoryStoreForReview
if (reviewer != null && store != null) {
try {
val convTurns = paired.map {
ConversationTurn(
userMessage = it.userMessage,
assistantMessage = it.assistantMessage,
createdAt = it.createdAt,
)
}
val decision = reviewer.reviewPreCompaction(convTurns)
for (n in decision.toSave) {
val note = materializeReviewNote(n, conversationId = null)
runCatching { store.upsert(note) }
.onFailure { log.warn(it) { "pre-compaction upsert failed: ${it.message}" } }
}
for (delId in decision.toDelete) {
runCatching { store.delete(delId) }
.onFailure { log.warn(it) { "pre-compaction delete failed: ${it.message}" } }
}
} catch (e: CancellationException) {
throw e
} catch (e: Throwable) {
log.warn(e) { "pre-compaction review failed for ${state.id}: ${e.message}" }
}
}
val dropFrom = toCompact.first().orderIdx
workingMemory.compact(dropFromOrderIdx = dropFrom, conversationId = state.id, summaryText = summaryText)
state.liteConvRef.getAndSet(null)?.let { runCatching { it.close() } }
val after = estimateTokens(
systemText = systemText,
history = workingMemory.list(state.id).filter { it.entry is WorkingMemoryEntry.User || it.entry is WorkingMemoryEntry.Assistant },
toolsChars = toolsCharsCached,
)
if (after.toDouble() / window >= compressionThreshold) {
log.warn { "compactPreTurn: still over threshold for ${state.id} (estimated=$after, window=$window, threshold=$compressionThreshold). Consider raising contextWindow or lowering threshold." }
}
return true
}
private fun estimateTokens(systemText: String, history: List<WorkingMemoryRow>, toolsChars: Int): Int {
val sysTokens = systemText.length / 4
val toolsTokens = toolsChars / 4
val historyChars = history.sumOf { row ->
when (val e = row.entry) {
is WorkingMemoryEntry.User -> e.content.sumCharLen()
is WorkingMemoryEntry.Assistant -> e.content.sumCharLen()
else -> 0
}
}
return sysTokens + toolsTokens + historyChars / 4
}
private fun List<Content>.text(): String =
filterIsInstance<Content.Text>().joinToString("\n") { it.body }
private fun List<Content>.sumCharLen(): Int = sumOf { c ->
when (c) {
is Content.Text -> c.body.length
is Content.Image -> c.data.size / 4
}
}
suspend fun getOrCreateLiteConversation(
systemPrompt: String,
excludeUserSourceId: String? = null,
): LiteConversation {
state.liteConvRef.get()?.let { return it }
val pastTurns: List<LiteMessage> = if (state.isTemporal) emptyList() else workingMemory.list(state.id)
.filter { row ->
val isRelevant = row.entry is WorkingMemoryEntry.User
|| row.entry is WorkingMemoryEntry.Assistant
|| row.entry is WorkingMemoryEntry.ToolExchange
val isPendingUser = excludeUserSourceId != null && row.sourceMessageId == excludeUserSourceId
isRelevant && !isPendingUser
}
.mapNotNull { row ->
val e: WorkingMemoryEntry = row.entry
val msg: LiteMessage? = when (e) {
is WorkingMemoryEntry.User -> LiteMessage(
LiteRole.USER,
applyContextPrefix(e.content.toLiteContents(), e.context),
)
is WorkingMemoryEntry.Assistant -> LiteMessage(LiteRole.MODEL, e.content.toLiteContents())
is WorkingMemoryEntry.ToolExchange -> LiteMessage(
LiteRole.TOOL,
listOf(
LiteContentPart.ToolResult(
callId = e.sourceMessageId,
name = e.toolName,
response = e.resultText,
),
),
)
else -> null
}
msg
}
val capped = if (pastTurns.size > MAX_SEEDED_MESSAGES) pastTurns.takeLast(MAX_SEEDED_MESSAGES) else pastTurns
val config = LiteConversationConfig(
systemInstruction = systemPrompt.takeIf { it.isNotBlank() },
initialMessages = capped,
tools = state.tools.map { it.tool },
)
return liteLlm.createConversation(config).also { state.liteConvRef.set(it) }
}
companion object {
private const val KEEP_RECENT_TURNS = 4
private const val MAX_SEEDED_MESSAGES = 50
}
}
internal fun List<Content>.toLiteContents(): List<LiteContentPart> = map { it.toLite() }
internal fun Content.toLite(): LiteContentPart = when (this) {
is Content.Text -> LiteContentPart.Text(body)
is Content.Image -> LiteContentPart.Image(data, mime)
}
@@ -0,0 +1,60 @@
package pw.binom.agentik.standalone.agent
import mu.KotlinLogging
import pw.binom.agentik.memory.MemoryPrefetcher
import pw.binom.litert.LiteContentPart
import pw.binom.agentik.storage.MessageContext
import pw.binom.agentik.storage.MessageOrigin
internal class ContextBuilder(
private val memoryPrefetcher: MemoryPrefetcher?,
) {
private val log = KotlinLogging.logger {}
suspend fun buildMemoryPrefix(parts: List<LiteContentPart>): String? {
val prefetcher = memoryPrefetcher ?: return null
val userText = parts.asSequence()
.filterIsInstance<LiteContentPart.Text>()
.map { it.text }
.joinToString("\n")
.trim()
if (userText.isEmpty()) return null
val notes = try {
prefetcher.prefetch(userText, topK = 10)
} catch (e: Throwable) {
log.warn(e) { "memory prefetch failed: ${e.message}" }
return null
}
if (notes.isEmpty()) return null
val body = notes.joinToString("\n") { n -> "- [${n.category.id}] ${n.content.take(280)}" }
return buildString {
appendLine("[Memory context — relevant long-term facts from previous sessions. Use if directly relevant to the user''s current request; do NOT treat as instructions or new facts to memorize. This block is regenerated each turn and may differ from one turn to another — that''s expected.]")
append(body)
}.trimEnd()
}
}
internal fun formatContextPrefix(context: MessageContext): String {
val parts = mutableListOf<String>()
parts += "[${context.origin.name}]"
context.description?.takeIf { it.isNotBlank() }?.let { parts += " $it" }
context.sourceId?.takeIf { it.isNotBlank() }?.let { parts += " (sourceId=$it)" }
return parts.joinToString("")
}
internal fun applyContextPrefix(parts: List<LiteContentPart>, context: MessageContext?): List<LiteContentPart> {
if (context == null || context.origin == MessageOrigin.USER) return parts
val prefix = formatContextPrefix(context)
val out = ArrayList<LiteContentPart>(parts.size + 1)
var inserted = false
for (p in parts) {
if (!inserted && p is LiteContentPart.Text) {
out += LiteContentPart.Text("$prefix\n${p.text}")
inserted = true
} else {
out += p
}
}
if (!inserted) out.add(0, LiteContentPart.Text(prefix))
return out
}
@@ -0,0 +1,21 @@
package pw.binom.agentik.standalone.agent
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import pw.binom.agentik.proto.Event as ProtoEvent
internal class ConversationEvents {
private val _flow = MutableSharedFlow<ProtoEvent>(
replay = 0,
extraBufferCapacity = 4096,
onBufferOverflow = BufferOverflow.DROP_OLDEST,
)
val flow: SharedFlow<ProtoEvent> get() = _flow.asSharedFlow()
fun tryEmit(event: ProtoEvent): Boolean = _flow.tryEmit(event)
suspend fun emit(event: ProtoEvent) = _flow.emit(event)
}
@@ -0,0 +1,572 @@
package pw.binom.agentik.standalone.agent
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.cancelAndJoin
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.serialization.json.Json
import kotlinx.serialization.json.JsonArray
import kotlinx.serialization.json.JsonElement
import kotlinx.serialization.json.JsonNull
import kotlinx.serialization.json.JsonPrimitive
import kotlinx.serialization.json.buildJsonObject
import mu.KotlinLogging
import pw.binom.agentik.memory.MemoryPrefetcher
import pw.binom.agentik.memory.MemoryReviewer
import pw.binom.agentik.memory.MemoryStore
import pw.binom.agentik.proto.Content as ProtoContent
import pw.binom.agentik.proto.Conversation as ProtoConversation
import pw.binom.agentik.proto.Event as ProtoEvent
import pw.binom.agentik.proto.Message as ProtoMessage
import pw.binom.agentik.proto.MessageContext as ProtoMessageContext
import pw.binom.agentik.skills.SkillStore
import pw.binom.agentik.storage.Content
import pw.binom.agentik.storage.ConversationRecord
import pw.binom.agentik.storage.ConversationStore
import pw.binom.agentik.storage.MessageContext
import pw.binom.agentik.storage.MessageOrigin
import pw.binom.agentik.storage.MessageRecord
import pw.binom.agentik.storage.MessageStore
import pw.binom.agentik.storage.ReflectionStore
import pw.binom.agentik.storage.StorageBundle
import pw.binom.agentik.storage.TurnTokens
import pw.binom.agentik.storage.WorkingMemoryEntry
import pw.binom.agentik.storage.WorkingMemoryStore
import pw.binom.agentik.toolsets.ToolsetDispatchPolicy
import pw.binom.litert.LiteContentPart
import pw.binom.litert.LiteConversation
import pw.binom.litert.LiteLlm
import pw.binom.litert.LiteTool
import pw.binom.litert.LiteToolCall
import java.util.concurrent.atomic.AtomicBoolean
import kotlin.time.Instant
class ConversationLoop(
record: ConversationRecord,
private val storage: StorageBundle,
private val llm: LiteLlm,
private val systemPrompt: String,
private val tools: List<NamedTool> = emptyList(),
private val toolsetDispatch: ToolsetDispatchPolicy? = null,
private val memoryPrefetcher: MemoryPrefetcher? = null,
private val memoryReviewer: MemoryReviewer? = null,
private val memoryStoreForReview: MemoryStore? = null,
private val memoryReviewInterval: Int = 0,
private val contextWindow: Int? = null,
private val compressionThreshold: Double = 0.8,
private val contextCompactor: ContextCompactor? = null,
private val reflectionStore: ReflectionStore? = null,
private val reflector: LlmReflector? = null,
private val reflectionInterval: Int = 0,
private val skillMiner: SkillMiner? = null,
private val skillMiningStore: SkillStore? = null,
private val skillMiningInterval: Int = 0,
) : ProtoConversation, AutoCloseable {
private val log = KotlinLogging.logger {}
private val agentScope: CoroutineScope = CoroutineScope(
SupervisorJob() + Dispatchers.IO.limitedParallelism(8),
)
private val state = ConversationState(
initialRecord = record,
tools = tools,
agentScope = agentScope,
)
private val events = ConversationEvents()
private val conversationStore: ConversationStore get() = storage.conversationStore
private val messageStore: MessageStore get() = storage.messageStore
private val workingMemory: WorkingMemoryStore get() = storage.workingMemoryStore
private val toolsByName: MutableMap<String, NamedTool> = tools.associateBy { it.name }.toMutableMap()
private val contextBuilder = ContextBuilder(memoryPrefetcher = memoryPrefetcher)
private val compactor = CompactionCoordinator(
state = state,
contextWindow = contextWindow,
compressionThreshold = compressionThreshold,
contextCompactor = contextCompactor,
memoryReviewer = memoryReviewer,
memoryStoreForReview = memoryStoreForReview,
workingMemory = workingMemory,
liteLlm = llm,
systemPrompt = systemPrompt,
)
private val toolDispatcher = ToolDispatcher(
state = state,
messageStore = messageStore,
events = events,
toolsByName = toolsByName,
toolsetDispatch = toolsetDispatch,
newId = ::newId,
encodeArgsJson = ::encodeArgsJson,
now = ::now,
)
private val backgroundScheduler = BackgroundScheduler(
state = state,
workingMemory = workingMemory,
config = BackgroundConfig(
memoryReviewer = memoryReviewer,
memoryStore = memoryStoreForReview,
memoryReviewInterval = memoryReviewInterval,
reflectionStore = reflectionStore,
reflector = reflector,
reflectionInterval = reflectionInterval,
skillMiner = skillMiner,
skillMiningStore = skillMiningStore,
skillMiningInterval = skillMiningInterval,
),
)
override val id: String get() = state.id
override val isSupportImageInput: Boolean get() = false
override val isSupportImageOutput: Boolean get() = false
override val isTemporal: Boolean get() = state.isTemporal
override val title: String? get() = state.record.title
override val updatedAt: Instant get() = state.record.updatedAt
private val turnLock = Mutex()
@Volatile
private var activeTurn: Job? = null
private val interrupted = AtomicBoolean(false)
internal val isClosed: Boolean get() = state.isClosed
override suspend fun rename(title: String) {
val newRecord = conversationStore.rename(id, title)?.let { ts ->
state.record.copy(title = title, updatedAt = ts)
} ?: state.record.copy(title = title)
state.record = newRecord
}
override suspend fun send(content: List<ProtoContent>, context: ProtoMessageContext?) {
check(!state.isClosed) { "Conversation closed: $id" }
val turnStarted = now()
val userMessageId = newId("msg")
val storageContext = context?.toStorage()
val userRecord = MessageRecord.UserMessage(
id = userMessageId,
conversationId = id,
content = content.map { it.toStorage() },
createdAt = turnStarted,
context = storageContext,
)
if (!state.isTemporal) {
messageStore.append(userRecord)
workingMemory.append(
conversationId = id,
entry = WorkingMemoryEntry.User(
sourceMessageId = userMessageId,
content = userRecord.content,
context = storageContext,
),
now = turnStarted,
)
}
turnLock.withLock {
activeTurn = agentScope.launch {
runTurn(userRecord, turnStarted)
}
activeTurn?.join()
}
}
override suspend fun interrupt() {
if (activeTurn?.isActive != true) {
log.info { "interrupt() no-op: no active turn for $id" }
return
}
interrupted.set(true)
runCatching { state.liteConvRef.get()?.cancel() }
toolDispatcher.currentToolJob?.cancel()
}
override fun events(after: Instant): Flow<ProtoEvent> =
events.flow
override suspend fun getMessages(after: Instant, offset: Int, limit: Int): List<ProtoMessage> =
messageStore.list(conversationId = id, after = after, offset = offset, limit = limit)
.map { it.toProto() }
override fun close() {
if (state.isClosed) return
state.markClosed()
state.liteConvRef.getAndSet(null)?.let { runCatching { it.close() } }
runCatching { runBlocking { activeTurn?.cancelAndJoin() } }
agentScope.cancel()
}
suspend fun forceCompactNow(): Boolean = compactor.forceCompactNow()
internal fun registerToolForTest(name: String, tool: LiteTool) {
toolDispatcher.registerToolForTest(name, tool)
}
private suspend fun runTurn(userRecord: MessageRecord.UserMessage, turnStarted: Instant) {
val wasInterruptedAtEntry = interrupted.get()
if (!state.isTemporal) {
compactor.compactPreTurnIfNeeded()
}
emitEvent(ProtoEvent.StartReasoning(date = turnStarted))
emitEvent(ProtoEvent.StartResponse(date = now(), responseType = ProtoEvent.ResponseType.TEXT))
val parts = userRecord.content.mapNotNull { c ->
when (c) {
is Content.Text -> LiteContentPart.Text(c.body)
is Content.Image -> {
log.warn { "dropping image input (v1 text-only): mime=${c.mime}, ${c.data.size} bytes" }
null
}
}
}.let { baseParts -> applyContextPrefix(baseParts, userRecord.context) }
if (parts.isEmpty()) {
failTurn("Empty user input (no text content)")
return
}
val initialParts = buildList {
val memoryBlock = contextBuilder.buildMemoryPrefix(parts)
if (memoryBlock != null) {
add(LiteContentPart.Text(memoryBlock))
}
addAll(parts)
}
val conv = try {
compactor.getOrCreateLiteConversation(
systemPrompt = systemPrompt,
excludeUserSourceId = if (state.isTemporal) null else userRecord.id,
)
} catch (e: Throwable) {
state.liteConvRef.set(null)
failTurn(e.message ?: "LiteConversation init failed")
return
}
val reply = StringBuilder()
val toolExchanges = mutableListOf<WorkingMemoryEntry.ToolExchange>()
var currentParts: List<LiteContentPart> = initialParts
var loopGuard = 0
val tokensAtTurnStart: Int? = readTokenCount(conv)
var turnTokens: TurnTokens? = null
var pendingParts: List<LiteContentPart>? = currentParts
try {
if (wasInterruptedAtEntry) {
log.info { "runTurn short-circuit on interrupted-flag-at-entry: $id" }
return
}
var pendingPostToolCalls: List<LiteToolCall> = emptyList()
while (loopGuard++ < MAX_TOOL_LOOPS) {
if (interrupted.get() && pendingParts == null) break
val collectedCalls = mutableListOf<LiteToolCall>()
if (pendingParts != null) {
val lc = state.liteConvRef.get() ?: return
try {
lc.sendStreamContents(pendingParts).collect { delta ->
if (delta.text.isNotEmpty()) {
reply.append(delta.text)
emitEvent(ProtoEvent.AppendText(date = now(), body = delta.text))
}
if (delta.toolCalls.isNotEmpty()) {
collectedCalls.addAll(delta.toolCalls)
}
}
} catch (e: CancellationException) {
log.info { "sendStreamContents cancelled for $id" }
break
} catch (e: Throwable) {
state.liteConvRef.set(null)
failTurn(e.message ?: e.javaClass.simpleName)
return
}
pendingParts = null
}
var nextCalls = if (pendingPostToolCalls.isNotEmpty()) pendingPostToolCalls else collectedCalls
pendingPostToolCalls = emptyList()
while (nextCalls.isNotEmpty()) {
val prev = nextCalls
nextCalls = mutableListOf()
for (call in prev) {
val exchange = toolDispatcher.runToolAndPersist(call)
toolExchanges += exchange
val lc = state.liteConvRef.get() ?: return
val delta = try {
lc.addToolResult(callId = exchange.sourceMessageId, name = exchange.toolName, result = exchange.resultText)
} catch (e: CancellationException) {
log.info { "addToolResult cancelled for $id" }
break
} catch (e: Throwable) {
state.liteConvRef.set(null)
failTurn(e.message ?: e.javaClass.simpleName)
return
}
if (delta.text.isNotEmpty()) {
reply.append(delta.text)
emitEvent(ProtoEvent.AppendText(date = now(), body = delta.text))
}
if (delta.toolCalls.isNotEmpty()) {
nextCalls.addAll(delta.toolCalls)
}
if (!interrupted.get()) {
try {
val collectedPostTool = mutableListOf<LiteToolCall>()
val lc = state.liteConvRef.get() ?: return
lc.sendStreamContents(listOf(LiteContentPart.Text(" "))).collect { followUp ->
if (followUp.text.isNotEmpty()) {
reply.append(followUp.text)
emitEvent(ProtoEvent.AppendText(date = now(), body = followUp.text))
}
if (followUp.toolCalls.isNotEmpty()) {
collectedPostTool.addAll(followUp.toolCalls)
}
}
if (collectedPostTool.isNotEmpty()) {
pendingPostToolCalls = collectedPostTool
}
} catch (e: CancellationException) {
log.info { "post-tool sendStreamContents cancelled for $id" }
break
} catch (e: Throwable) {
log.warn(e) { "post-tool sendStreamContents failed for $id" }
break
}
}
}
if (interrupted.get()) break
}
if (nextCalls.isEmpty() && pendingParts == null) break
if (interrupted.get()) break
if (nextCalls.isEmpty()) break
}
if (loopGuard >= MAX_TOOL_LOOPS) {
log.warn { "tool loop hit MAX_TOOL_LOOPS=$MAX_TOOL_LOOPS for $id — bailing" }
}
if (tokensAtTurnStart != null) {
val tokensAtTurnEnd = readTokenCount(conv!!)
if (tokensAtTurnEnd != null) {
val output = (tokensAtTurnEnd - tokensAtTurnStart).coerceAtLeast(0)
turnTokens = TurnTokens(input = tokensAtTurnStart, output = output)
}
}
} finally {
val lc = state.liteConvRef.getAndSet(null)
runCatching { lc?.close() }
val wasInterrupted = interrupted.get()
if (!state.isTemporal) {
if (reply.isNotEmpty() || toolExchanges.isNotEmpty()) {
val assistantId = newId("msg")
val assistantAt = now()
val assistantContent = listOf(Content.Text(reply.toString()))
val assistantRecord = MessageRecord.AssistantMessage(
id = assistantId,
conversationId = id,
content = assistantContent,
createdAt = assistantAt,
tokens = turnTokens,
)
messageStore.append(assistantRecord)
workingMemory.append(
conversationId = id,
entry = WorkingMemoryEntry.Assistant(
sourceMessageId = assistantId,
content = assistantContent,
),
now = assistantAt,
)
for (ex in toolExchanges) {
workingMemory.append(
conversationId = id,
entry = ex,
now = assistantAt,
)
}
state.record = state.record.copy(updatedAt = assistantAt)
conversationStore.touch(id, assistantAt)
backgroundScheduler.maybeScheduleReview(userRecord, assistantContent)
backgroundScheduler.maybeScheduleReflection(userRecord, assistantContent)
backgroundScheduler.maybeScheduleSkillMining(userRecord, assistantContent)
}
}
if (wasInterrupted || interrupted.get()) {
emitEvent(ProtoEvent.Interrupted(date = now()))
}
emitEvent(ProtoEvent.End(date = now()))
interrupted.set(false)
}
}
private fun emitEvent(event: ProtoEvent) {
events.tryEmit(event)
}
private suspend fun failTurn(message: String, code: String? = null) {
val ts = now()
if (!state.isTemporal) {
messageStore.append(
MessageRecord.Error(
id = newId("err"),
conversationId = id,
message = message,
code = code,
createdAt = ts,
),
)
}
emitEvent(ProtoEvent.Error(date = ts, message = message, code = code))
}
private fun now(): Instant =
Instant.fromEpochMilliseconds(System.currentTimeMillis())
private fun newId(prefix: String): String = pw.binom.agentik.storage.Ids.new(prefix)
private fun encodeArgsJson(arguments: Map<String, Any?>): String {
val el = JsonElement.serializer()
val obj = buildJsonObject {
arguments.forEach { (k, v) -> put(k, v.toJsonElement()) }
}
return Json.encodeToString(el, obj)
}
private fun Any?.toJsonElement(): JsonElement = when (this) {
null -> JsonNull
is Boolean -> JsonPrimitive(this)
is Number -> JsonPrimitive(this)
is String -> JsonPrimitive(this)
is Map<*, *> -> buildJsonObject {
this@toJsonElement.forEach { (k, v) ->
put(k.toString(), v.toJsonElement())
}
}
is List<*> -> JsonArray(this.map { it.toJsonElement() })
else -> JsonPrimitive(toString())
}
companion object {
private const val MAX_TOOL_LOOPS = 16
}
}
private fun Content.toProto(): ProtoContent = when (this) {
is Content.Text -> ProtoContent.Text(body = body)
is Content.Image -> ProtoContent.Image(data = data, mime = mime)
}
internal fun ProtoContent.toStorage(): Content = when (this) {
is ProtoContent.Text -> Content.Text(body)
is ProtoContent.Image -> Content.Image(data, mime)
}
internal fun ProtoMessageContext.toStorage(): MessageContext = MessageContext(
origin = when (origin) {
pw.binom.agentik.proto.MessageOrigin.USER -> MessageOrigin.USER
pw.binom.agentik.proto.MessageOrigin.SYSTEM -> MessageOrigin.SYSTEM
pw.binom.agentik.proto.MessageOrigin.EVENT -> MessageOrigin.EVENT
},
description = description,
sourceId = sourceId,
metadata = metadata,
)
internal fun MessageContext.toProto(): ProtoMessageContext {
val protoOrigin = when (origin) {
MessageOrigin.USER -> pw.binom.agentik.proto.MessageOrigin.USER
MessageOrigin.SYSTEM -> pw.binom.agentik.proto.MessageOrigin.SYSTEM
MessageOrigin.EVENT -> pw.binom.agentik.proto.MessageOrigin.EVENT
}
return ProtoMessageContext(
origin = protoOrigin,
description = description,
sourceId = sourceId,
metadata = metadata,
)
}
internal fun MessageRecord.toProto(): ProtoMessage = when (this) {
is MessageRecord.UserMessage -> ProtoMessage.UserMessage(
id = id,
date = createdAt,
content = content.map { it.toProto() },
context = context?.toProto(),
)
is MessageRecord.AssistantMessage -> ProtoMessage.AssistantMessage(
id = id,
date = createdAt,
content = content.map { it.toProto() },
)
is MessageRecord.ToolCall -> ProtoMessage.ToolCall(
id = id,
date = createdAt,
title = toolTitle,
toolName = toolName,
toolArgs = toolArgsJson,
)
is MessageRecord.ToolResult -> ProtoMessage.ToolResult(
id = id,
date = createdAt,
result = result,
)
is MessageRecord.Error -> ProtoMessage.Error(
id = id,
date = createdAt,
message = message,
code = code,
)
is MessageRecord.Summary -> ProtoMessage.AssistantMessage(
id = id,
date = createdAt,
content = listOf(ProtoContent.Text(body = text)),
)
is MessageRecord.System -> ProtoMessage.UserMessage(
id = id,
date = createdAt,
content = listOf(ProtoContent.Text(body = text)),
)
}
private fun readTokenCount(liteConv: LiteConversation): Int? = try {
val n = liteConv.tokenCount()
if (n < 0) null else n
} catch (_: Throwable) {
null
}
@@ -0,0 +1,30 @@
package pw.binom.agentik.standalone.agent
import kotlinx.coroutines.CoroutineScope
import pw.binom.agentik.storage.ConversationRecord
import pw.binom.litert.LiteConversation
import java.util.concurrent.atomic.AtomicReference
internal class ConversationState(
initialRecord: ConversationRecord,
val tools: List<NamedTool>,
val agentScope: CoroutineScope,
) {
@Volatile
var record: ConversationRecord = initialRecord
val id: String get() = record.id
val isTemporal: Boolean get() = record.isTemporal
@Volatile
private var closed = false
val isClosed: Boolean get() = closed
fun markClosed() {
closed = true
}
private val _liteConvRef = AtomicReference<LiteConversation?>(null)
val liteConvRef: AtomicReference<LiteConversation?> get() = _liteConvRef
}
@@ -0,0 +1,113 @@
package pw.binom.agentik.standalone.agent
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.Job
import kotlinx.coroutines.async
import mu.KotlinLogging
import pw.binom.agentik.proto.Event as ProtoEvent
import pw.binom.agentik.storage.MessageRecord
import pw.binom.agentik.storage.MessageStore
import pw.binom.agentik.storage.WorkingMemoryEntry
import pw.binom.agentik.toolsets.ToolsetDispatchPolicy
import pw.binom.litert.LiteToolCall
import pw.binom.litert.LiteTool
import kotlin.time.Instant
internal class ToolDispatcher(
private val state: ConversationState,
private val messageStore: MessageStore,
private val events: ConversationEvents,
private val toolsByName: MutableMap<String, NamedTool>,
private val toolsetDispatch: ToolsetDispatchPolicy?,
private val newId: (String) -> String,
private val encodeArgsJson: (Map<String, Any?>) -> String,
private val now: () -> Instant,
) {
private val log = KotlinLogging.logger {}
@Volatile
private var _currentToolJob: Job? = null
val currentToolJob: Job? get() = _currentToolJob
internal fun registerToolForTest(name: String, tool: LiteTool) {
toolsByName[name] = NamedTool(name = name, tool = tool)
}
suspend fun runToolAndPersist(call: LiteToolCall): WorkingMemoryEntry.ToolExchange {
val callId = newId("tc")
val resultId = newId("tr")
val argsJson = encodeArgsJson(call.arguments)
val nowTs = now()
events.tryEmit(ProtoEvent.ToolCall(date = nowTs, id = callId, title = null, toolName = call.name, toolArgs = argsJson))
if (!state.isTemporal) {
messageStore.append(
MessageRecord.ToolCall(
id = callId,
conversationId = state.id,
toolName = call.name,
toolTitle = null,
toolArgsJson = argsJson,
createdAt = nowTs,
),
)
}
val toolDeferred = state.agentScope.async {
if (toolsetDispatch == null) {
val t = toolsByName[call.name]
if (t == null) {
log.warn { "tool '${call.name}' requested but not registered" }
"[tool not found: ${call.name}]"
} else {
t.tool.invoke(argsJson)
}
} else {
val d = toolsetDispatch
when (val o = d.dispatch(call.name, argsJson)) {
is ToolsetDispatchPolicy.Outcome.Ran -> o.result
is ToolsetDispatchPolicy.Outcome.Unknown -> "[tool not found: ${call.name}]"
}
}
}
_currentToolJob = toolDeferred
val resultText: String = try {
toolDeferred.await()
} catch (e: CancellationException) {
"[cancelled by user]"
} catch (e: InterruptedException) {
"[cancelled by user]"
} catch (e: Throwable) {
log.warn(e) { "tool '${call.name}' threw: ${e.message}" }
"[tool error: ${e.message ?: e.javaClass.simpleName}]"
} finally {
_currentToolJob = null
}
val resultAt = now()
events.tryEmit(ProtoEvent.ToolResult(date = resultAt, id = resultId, result = resultText))
if (!state.isTemporal) {
messageStore.append(
MessageRecord.ToolResult(
id = resultId,
conversationId = state.id,
toolCallId = callId,
result = resultText,
createdAt = resultAt,
),
)
}
return WorkingMemoryEntry.ToolExchange(
sourceMessageId = callId,
toolName = call.name,
toolArgsJson = argsJson,
resultText = resultText,
wasCancelled = resultText == "[cancelled by user]",
)
}
}