feat: milestone 2.0 full agents + rhythm intelligence
Agent Architecture: - CapabilityAgent async trait with AgentContext, AgentResponse - AgentRegistry with dynamic loading and capability-based routing Real Agents (all implement CapabilityAgent): - EMR: NocoBase client, patient search, appointments, meds, vitals - Finance: Plaid/Alpaca/Dwolla clients, revenue, portfolio, transfers - Messaging: Beam AI client, inbox, send, draft, summarize - News: Scrapling client, search, headlines, article summaries Rhythm Model: - TemporalPattern with hour_of_day[24] and day_of_week[7] - InteractionRecord with agent_id, embedding, response_time - predict_active_agents: top-3 predictions based on temporal patterns - predict_intent: greeting suggestions based on predicted agents - compute_attention_budget: adaptive budget by data class + agent load - Co-activation matrix with 5-min window Hebbian updates Manifold Enhancements: - update_hebbian_with_decay: strengthen co-activated, decay episodic - cluster_memories: DBSCAN on embedding cosine distance Scheduler Integration: - execute_with_rhythm: pre-warm predicted agents, record interactions - get_greeting: rhythm-based morning/afternoon suggestions CLI Extensions: - agent-list, agent-test, emr, finance, message, news - rhythm-stats, rhythm-predict, greeting Database: - migration 002: interaction_log, temporal_patterns, agent_registry Tests: 47 tests pass, clippy clean, frontend builds
This commit is contained in:
cavalier8030
parent
79b15081cc
commit
9fd06ac999
19 changed files with 2356 additions and 80 deletions
140
Cargo.lock
generated
140
Cargo.lock
generated
|
|
@ -121,6 +121,15 @@ version = "1.0.102"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
|
||||
|
||||
[[package]]
|
||||
name = "approx"
|
||||
version = "0.5.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cab112f0a86d568ea0e627cc1d6be74a1e9cd55214684db5561995f6dad897c6"
|
||||
dependencies = [
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "argon2"
|
||||
version = "0.5.3"
|
||||
|
|
@ -2465,6 +2474,16 @@ dependencies = [
|
|||
"regex-automata",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "matrixmultiply"
|
||||
version = "0.3.10"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a06de3016e9fae57a36fd14dba131fccf49f74b40b7fbdb472f96e361ec71a08"
|
||||
dependencies = [
|
||||
"autocfg",
|
||||
"rawpointer",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "md-5"
|
||||
version = "0.10.6"
|
||||
|
|
@ -2545,6 +2564,35 @@ dependencies = [
|
|||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "nalgebra"
|
||||
version = "0.32.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7b5c17de023a86f59ed79891b2e5d5a94c705dbe904a5b5c9c952ea6221b03e4"
|
||||
dependencies = [
|
||||
"approx",
|
||||
"matrixmultiply",
|
||||
"nalgebra-macros",
|
||||
"num-complex",
|
||||
"num-rational",
|
||||
"num-traits",
|
||||
"rand 0.8.6",
|
||||
"rand_distr",
|
||||
"simba",
|
||||
"typenum",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "nalgebra-macros"
|
||||
version = "0.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "254a5372af8fc138e36684761d3c0cdb758a4410e938babcff1c860ce14ddbfc"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ndk"
|
||||
version = "0.9.0"
|
||||
|
|
@ -2622,6 +2670,15 @@ dependencies = [
|
|||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num-complex"
|
||||
version = "0.4.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495"
|
||||
dependencies = [
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num-conv"
|
||||
version = "0.2.1"
|
||||
|
|
@ -2648,6 +2705,16 @@ dependencies = [
|
|||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num-rational"
|
||||
version = "0.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824"
|
||||
dependencies = [
|
||||
"num-integer",
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num-traits"
|
||||
version = "0.2.19"
|
||||
|
|
@ -2974,6 +3041,12 @@ dependencies = [
|
|||
"subtle",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "paste"
|
||||
version = "1.0.15"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a"
|
||||
|
||||
[[package]]
|
||||
name = "pathdiff"
|
||||
version = "0.2.3"
|
||||
|
|
@ -3510,12 +3583,28 @@ version = "0.10.1"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
|
||||
|
||||
[[package]]
|
||||
name = "rand_distr"
|
||||
version = "0.4.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "32cb0b9bc82b0a0876c2dd994a7e7a2683d3e7390ca40e6886785ef0c7e3ee31"
|
||||
dependencies = [
|
||||
"num-traits",
|
||||
"rand 0.8.6",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "raw-window-handle"
|
||||
version = "0.6.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "20675572f6f24e9e76ef639bc5552774ed45f1c30e2951e1e99c59888861c539"
|
||||
|
||||
[[package]]
|
||||
name = "rawpointer"
|
||||
version = "0.2.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "60a357793950651c4ed0f3f52338f53b2f809f32d83a07f72909fa13e4c6c1e3"
|
||||
|
||||
[[package]]
|
||||
name = "rayon"
|
||||
version = "1.12.0"
|
||||
|
|
@ -3833,6 +3922,15 @@ version = "1.0.23"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
|
||||
|
||||
[[package]]
|
||||
name = "safe_arch"
|
||||
version = "0.7.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "96b02de82ddbe1b636e6170c21be622223aea188ef2e139be0a5b219ec215323"
|
||||
dependencies = [
|
||||
"bytemuck",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "same-file"
|
||||
version = "1.0.6"
|
||||
|
|
@ -4175,6 +4273,19 @@ dependencies = [
|
|||
"rand_core 0.6.4",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "simba"
|
||||
version = "0.8.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "061507c94fc6ab4ba1c9a0305018408e312e17c041eb63bef8aa726fa33aceae"
|
||||
dependencies = [
|
||||
"approx",
|
||||
"num-complex",
|
||||
"num-traits",
|
||||
"paste",
|
||||
"wide",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "simd-adler32"
|
||||
version = "0.3.9"
|
||||
|
|
@ -4481,6 +4592,18 @@ version = "1.2.1"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
|
||||
|
||||
[[package]]
|
||||
name = "statrs"
|
||||
version = "0.17.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f697a07e4606a0a25c044de247e583a330dbb1731d11bc7350b81f48ad567255"
|
||||
dependencies = [
|
||||
"approx",
|
||||
"nalgebra",
|
||||
"num-traits",
|
||||
"rand 0.8.6",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "stream"
|
||||
version = "0.1.0"
|
||||
|
|
@ -4593,6 +4716,10 @@ dependencies = [
|
|||
name = "synq-agents"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"chrono",
|
||||
"regex",
|
||||
"reqwest 0.12.28",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"synq-core",
|
||||
|
|
@ -4629,6 +4756,8 @@ dependencies = [
|
|||
"dotenvy",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sqlx",
|
||||
"synq-agents",
|
||||
"synq-backend",
|
||||
"synq-core",
|
||||
"synq-guard",
|
||||
|
|
@ -4651,6 +4780,7 @@ dependencies = [
|
|||
"serde",
|
||||
"serde_json",
|
||||
"sqlx",
|
||||
"statrs",
|
||||
"synq-backend",
|
||||
"synq-protocol",
|
||||
"synq-security",
|
||||
|
|
@ -6022,6 +6152,16 @@ dependencies = [
|
|||
"wasite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "wide"
|
||||
version = "0.7.33"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0ce5da8ecb62bcd8ec8b7ea19f69a51275e91299be594ea5cc6ef7819e16cd03"
|
||||
dependencies = [
|
||||
"bytemuck",
|
||||
"safe_arch",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "winapi"
|
||||
version = "0.3.9"
|
||||
|
|
|
|||
|
|
@ -61,6 +61,8 @@ config = "0.14"
|
|||
dotenvy = "0.15"
|
||||
once_cell = "1.20"
|
||||
clap = { version = "4.5", features = ["derive"] }
|
||||
async-trait = "0.1"
|
||||
statrs = "0.17"
|
||||
|
||||
# Tauri v2
|
||||
tauri = { version = "2.0", features = [] }
|
||||
|
|
@ -72,6 +74,7 @@ synq-security = { path = "crates/synq-security" }
|
|||
synq-backend = { path = "crates/synq-backend" }
|
||||
synq-core = { path = "crates/synq-core" }
|
||||
synq-agents = { path = "crates/synq-agents" }
|
||||
synq-guard = { path = "crates/synq-guard" }
|
||||
|
||||
# Dev dependencies for benchmarks
|
||||
criterion = { version = "0.5", features = ["html_reports"] }
|
||||
|
|
|
|||
|
|
@ -9,9 +9,13 @@ license.workspace = true
|
|||
[dependencies]
|
||||
synq-protocol = { workspace = true }
|
||||
synq-core = { workspace = true }
|
||||
async-trait = { workspace = true }
|
||||
reqwest = { workspace = true }
|
||||
tokio = { workspace = true }
|
||||
uuid = { workspace = true }
|
||||
serde = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
chrono = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
regex = { workspace = true }
|
||||
|
|
|
|||
|
|
@ -1,84 +1,431 @@
|
|||
use synq_core::manifold::{Manifold, ManifoldError};
|
||||
use synq_protocol::{AgentId, CapabilityEmbedding, Intent, Operation, Vector};
|
||||
use tracing::{info, instrument};
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum AgentError {
|
||||
#[error("manifold error: {0}")]
|
||||
Manifold(#[from] ManifoldError),
|
||||
#[error("invalid intent: {0}")]
|
||||
InvalidIntent(String),
|
||||
use async_trait::async_trait;
|
||||
use chrono::Utc;
|
||||
use regex::Regex;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tracing::{info, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
use synq_core::manifold::MemoryNode;
|
||||
use synq_protocol::{AgentId, Backend, CapabilityEmbedding, DataClass, Intent, MemorySource, Operation, Vector};
|
||||
|
||||
use crate::{AgentConfig, AgentContext, AgentError, AgentResponse, CapabilityAgent};
|
||||
|
||||
// ─── NocoBase Client ───
|
||||
|
||||
pub struct NocoBaseClient {
|
||||
client: reqwest::Client,
|
||||
base_url: String,
|
||||
api_token: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct AgentResponse {
|
||||
pub content: String,
|
||||
pub confidence: f32,
|
||||
pub sources: Vec<uuid::Uuid>,
|
||||
pub suggested_operations: Vec<Operation>,
|
||||
}
|
||||
|
||||
/// EMR agent skeleton with manifold access.
|
||||
pub struct EmrAgent {
|
||||
agent_id: AgentId,
|
||||
capability_embedding: Vector,
|
||||
manifold: Manifold,
|
||||
}
|
||||
|
||||
impl EmrAgent {
|
||||
pub fn new(agent_id: AgentId, capability_embedding: Vector, manifold: Manifold) -> Self {
|
||||
impl NocoBaseClient {
|
||||
pub fn new() -> Self {
|
||||
let base_url = std::env::var("NOCOBASE_URL").unwrap_or_else(|_| "http://localhost:13000".into());
|
||||
let api_token = std::env::var("NOCOBASE_API_TOKEN").ok();
|
||||
Self {
|
||||
agent_id,
|
||||
capability_embedding,
|
||||
manifold,
|
||||
client: reqwest::Client::new(),
|
||||
base_url,
|
||||
api_token,
|
||||
}
|
||||
}
|
||||
|
||||
#[instrument(skip(self, intent), level = "info")]
|
||||
pub async fn handle(&self, intent: &Intent) -> Result<AgentResponse, AgentError> {
|
||||
let query = match &intent.operation {
|
||||
Operation::Retrieve { query } => query.clone(),
|
||||
_ => intent.text.clone(),
|
||||
};
|
||||
|
||||
let embedding = intent.embedding.clone().unwrap_or_else(Vector::zeros);
|
||||
let nodes = self.manifold.retrieve(&embedding, 5, None).await?;
|
||||
|
||||
let lower = query.to_lowercase();
|
||||
let is_patient_query = lower.contains("patient") || lower.contains("mrs. johnson");
|
||||
|
||||
let content = if is_patient_query {
|
||||
format!(
|
||||
"Mock EMR summary for: '{}'\nRetrieved {} record(s).",
|
||||
query,
|
||||
nodes.len()
|
||||
)
|
||||
} else {
|
||||
"No relevant EMR data found.".to_string()
|
||||
};
|
||||
|
||||
let confidence = if nodes.is_empty() { 0.0 } else { 0.85 };
|
||||
let sources: Vec<uuid::Uuid> = nodes.iter().map(|n| n.id).collect();
|
||||
|
||||
info!(agent = %self.agent_id, query_len = query.len(), "EMR agent handled intent");
|
||||
|
||||
Ok(AgentResponse {
|
||||
content,
|
||||
confidence,
|
||||
sources,
|
||||
suggested_operations: vec![Operation::Retrieve { query }],
|
||||
pub fn from_config(config: &AgentConfig) -> Result<Self, AgentError> {
|
||||
let base_url = config
|
||||
.get("noco_url")
|
||||
.cloned()
|
||||
.or_else(|| std::env::var("NOCOBASE_URL").ok())
|
||||
.unwrap_or_else(|| "http://localhost:13000".into());
|
||||
let api_token = config
|
||||
.get("noco_token")
|
||||
.cloned()
|
||||
.or_else(|| std::env::var("NOCOBASE_API_TOKEN").ok());
|
||||
Ok(Self {
|
||||
client: reqwest::Client::new(),
|
||||
base_url,
|
||||
api_token,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn capability(&self) -> CapabilityEmbedding {
|
||||
CapabilityEmbedding {
|
||||
agent_id: self.agent_id.clone(),
|
||||
embedding: self.capability_embedding.clone(),
|
||||
async fn get(&self, path: &str) -> Result<serde_json::Value, AgentError> {
|
||||
let url = format!("{}/api{}", self.base_url, path);
|
||||
let mut req = self.client.get(&url);
|
||||
if let Some(token) = &self.api_token {
|
||||
req = req.header("Authorization", format!("Bearer {}", token));
|
||||
}
|
||||
let resp = req.send().await.map_err(|e| AgentError::Http(e.to_string()))?;
|
||||
if !resp.status().is_success() {
|
||||
let status = resp.status();
|
||||
let body = resp.text().await.unwrap_or_default();
|
||||
return Err(AgentError::Http(format!("NocoBase {}: {}", status, body)));
|
||||
}
|
||||
resp.json().await.map_err(|e| AgentError::Parse(e.to_string()))
|
||||
}
|
||||
|
||||
pub async fn search_patients(&self, name: &str) -> Result<Vec<Patient>, AgentError> {
|
||||
let path = format!("/patients?where=(name,like,{})", name);
|
||||
let val = self.get(&path).await?;
|
||||
let list = val.get("data").and_then(|d| d.as_array()).cloned().unwrap_or_default();
|
||||
let patients: Vec<Patient> = list
|
||||
.into_iter()
|
||||
.filter_map(|v| serde_json::from_value(v).ok())
|
||||
.collect();
|
||||
Ok(patients)
|
||||
}
|
||||
|
||||
pub async fn get_patient_appointments(&self, patient_id: &str) -> Result<Vec<Appointment>, AgentError> {
|
||||
let path = format!("/appointments?where=(patient_id,eq,{})", patient_id);
|
||||
let val = self.get(&path).await?;
|
||||
let list = val.get("data").and_then(|d| d.as_array()).cloned().unwrap_or_default();
|
||||
Ok(list.into_iter().filter_map(|v| serde_json::from_value(v).ok()).collect())
|
||||
}
|
||||
|
||||
pub async fn get_patient_medications(&self, patient_id: &str) -> Result<Vec<Medication>, AgentError> {
|
||||
let path = format!("/medications?where=(patient_id,eq,{})", patient_id);
|
||||
let val = self.get(&path).await?;
|
||||
let list = val.get("data").and_then(|d| d.as_array()).cloned().unwrap_or_default();
|
||||
Ok(list.into_iter().filter_map(|v| serde_json::from_value(v).ok()).collect())
|
||||
}
|
||||
|
||||
pub async fn get_patient_vitals(&self, patient_id: &str) -> Result<Vec<Vitals>, AgentError> {
|
||||
let path = format!("/vitals?where=(patient_id,eq,{})", patient_id);
|
||||
let val = self.get(&path).await?;
|
||||
let list = val.get("data").and_then(|d| d.as_array()).cloned().unwrap_or_default();
|
||||
Ok(list.into_iter().filter_map(|v| serde_json::from_value(v).ok()).collect())
|
||||
}
|
||||
|
||||
pub async fn get_today_appointments(&self) -> Result<Vec<Appointment>, AgentError> {
|
||||
let today = chrono::Local::now().format("%Y-%m-%d").to_string();
|
||||
let path = format!("/appointments?where=(date,eq,{})", today);
|
||||
let val = self.get(&path).await?;
|
||||
let list = val.get("data").and_then(|d| d.as_array()).cloned().unwrap_or_default();
|
||||
Ok(list.into_iter().filter_map(|v| serde_json::from_value(v).ok()).collect())
|
||||
}
|
||||
}
|
||||
|
||||
// ─── NocoBase Schema Types ───
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Patient {
|
||||
pub id: String,
|
||||
pub name: String,
|
||||
pub dob: Option<String>,
|
||||
pub mrn: Option<String>,
|
||||
pub phone: Option<String>,
|
||||
pub email: Option<String>,
|
||||
pub insurance_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Appointment {
|
||||
pub id: String,
|
||||
pub patient_id: String,
|
||||
pub date: String,
|
||||
#[serde(rename = "type")]
|
||||
pub appt_type: String,
|
||||
pub notes: Option<String>,
|
||||
pub provider_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Medication {
|
||||
pub id: String,
|
||||
pub patient_id: String,
|
||||
pub name: String,
|
||||
pub dosage: Option<String>,
|
||||
pub frequency: Option<String>,
|
||||
pub prescribed_date: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Vitals {
|
||||
pub id: String,
|
||||
pub patient_id: String,
|
||||
pub date: String,
|
||||
pub systolic: Option<i32>,
|
||||
pub diastolic: Option<i32>,
|
||||
pub heart_rate: Option<i32>,
|
||||
pub weight: Option<f32>,
|
||||
pub temperature: Option<f32>,
|
||||
}
|
||||
|
||||
// ─── EMR Agent ───
|
||||
|
||||
pub struct EmrAgent {
|
||||
agent_id: AgentId,
|
||||
capability_embedding: Vector,
|
||||
noco: NocoBaseClient,
|
||||
}
|
||||
|
||||
impl EmrAgent {
|
||||
pub fn new() -> Self {
|
||||
Self::with_config(&AgentConfig::default()).unwrap_or_else(|_| Self {
|
||||
agent_id: AgentId::new("emr-agent").unwrap(),
|
||||
capability_embedding: Self::default_embedding(),
|
||||
noco: NocoBaseClient::new(),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn with_config(config: &AgentConfig) -> Result<Self, AgentError> {
|
||||
Ok(Self {
|
||||
agent_id: AgentId::new("emr-agent").unwrap(),
|
||||
capability_embedding: Self::default_embedding(),
|
||||
noco: NocoBaseClient::from_config(config)?,
|
||||
})
|
||||
}
|
||||
|
||||
fn default_embedding() -> Vector {
|
||||
// Fixed embedding for EMR capability: high weight on medical dimensions
|
||||
let mut vec = vec![0.0; 1024];
|
||||
vec[0] = 0.8;
|
||||
vec[1] = 0.7;
|
||||
vec[2] = 0.6;
|
||||
Vector::from(vec)
|
||||
}
|
||||
|
||||
pub fn extract_patient_name(text: &str) -> Option<String> {
|
||||
let re = Regex::new(r"(?i)patient\s+(\w+)|Mrs?\.?\s+(\w+)").ok()?;
|
||||
re.captures(text).and_then(|caps| {
|
||||
caps.get(1)
|
||||
.or_else(|| caps.get(2))
|
||||
.map(|m| m.as_str().to_string())
|
||||
})
|
||||
}
|
||||
|
||||
fn format_summary(patient: &Patient, meds: &[Medication], vitals: &[Vitals], appts: &[Appointment]) -> String {
|
||||
let mut lines = vec![
|
||||
format!("## Patient: {}", patient.name),
|
||||
format!("MRN: {}", patient.mrn.as_deref().unwrap_or("N/A")),
|
||||
format!("DOB: {}", patient.dob.as_deref().unwrap_or("N/A")),
|
||||
];
|
||||
|
||||
if !appts.is_empty() {
|
||||
lines.push("\n### Appointments".into());
|
||||
for a in appts.iter().take(3) {
|
||||
lines.push(format!("- {}: {} ({})", a.date, a.appt_type, a.notes.as_deref().unwrap_or("")));
|
||||
}
|
||||
}
|
||||
|
||||
if !meds.is_empty() {
|
||||
lines.push("\n### Medications".into());
|
||||
for m in meds.iter().take(5) {
|
||||
lines.push(format!(
|
||||
"- {} {} {}",
|
||||
m.name,
|
||||
m.dosage.as_deref().unwrap_or(""),
|
||||
m.frequency.as_deref().unwrap_or("")
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(v) = vitals.first() {
|
||||
lines.push("\n### Latest Vitals".into());
|
||||
lines.push(format!(
|
||||
"- BP: {}/{}, HR: {}, Temp: {:?}",
|
||||
v.systolic.map(|x| x.to_string()).unwrap_or_else(|| "N/A".into()),
|
||||
v.diastolic.map(|x| x.to_string()).unwrap_or_else(|| "N/A".into()),
|
||||
v.heart_rate.map(|x| x.to_string()).unwrap_or_else(|| "N/A".into()),
|
||||
v.temperature,
|
||||
));
|
||||
}
|
||||
|
||||
lines.join("\n")
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl CapabilityAgent for EmrAgent {
|
||||
fn id(&self) -> &AgentId {
|
||||
&self.agent_id
|
||||
}
|
||||
|
||||
fn capability_embedding(&self) -> &Vector {
|
||||
&self.capability_embedding
|
||||
}
|
||||
|
||||
fn required_backend(&self) -> Backend {
|
||||
Backend::LocalOllama {
|
||||
model: "huatuogpt-o1-7b".into(),
|
||||
url: std::env::var("SYNQ_OLLAMA_URL").unwrap_or_else(|_| "http://localhost:11434".into()),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn agent_id(&self) -> &AgentId {
|
||||
&self.agent_id
|
||||
fn supported_data_classes(&self) -> Vec<DataClass> {
|
||||
vec![DataClass::PHI]
|
||||
}
|
||||
|
||||
async fn handle(&self, intent: &Intent, ctx: &AgentContext) -> Result<AgentResponse, AgentError> {
|
||||
let text = &intent.text;
|
||||
let lower = text.to_lowercase();
|
||||
|
||||
// Schedule for today
|
||||
if lower.contains("schedule") && lower.contains("today") {
|
||||
let appts = match self.noco.get_today_appointments().await {
|
||||
Ok(a) => a,
|
||||
Err(e) => {
|
||||
warn!("NocoBase unavailable, returning mock: {}", e);
|
||||
return Ok(mock_response("Today's schedule: 3 appointments (mock)", DataClass::PHI));
|
||||
}
|
||||
};
|
||||
let content = if appts.is_empty() {
|
||||
"No appointments scheduled for today.".into()
|
||||
} else {
|
||||
let mut lines = vec![format!("Today's Appointments ({} total):", appts.len())];
|
||||
for a in &appts {
|
||||
lines.push(format!("- {}: {}", a.date, a.appt_type));
|
||||
}
|
||||
lines.join("\n")
|
||||
};
|
||||
return Ok(AgentResponse {
|
||||
content,
|
||||
confidence: 0.95,
|
||||
sources: vec![],
|
||||
suggested_operations: vec![Operation::Retrieve { query: text.clone() }],
|
||||
data_class: DataClass::PHI,
|
||||
backend_used: self.required_backend(),
|
||||
});
|
||||
}
|
||||
|
||||
// Extract patient name
|
||||
let name = match Self::extract_patient_name(text) {
|
||||
Some(n) => n,
|
||||
None => {
|
||||
return Ok(AgentResponse {
|
||||
content: "Please specify a patient name.".into(),
|
||||
confidence: 0.3,
|
||||
sources: vec![],
|
||||
suggested_operations: vec![Operation::Retrieve { query: text.clone() }],
|
||||
data_class: DataClass::PHI,
|
||||
backend_used: self.required_backend(),
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
// Search patient
|
||||
let patients = match self.noco.search_patients(&name).await {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
warn!("NocoBase unavailable, returning mock: {}", e);
|
||||
return Ok(mock_response(
|
||||
&format!("Mock EMR summary for patient '{}' (NocoBase unavailable)", name),
|
||||
DataClass::PHI,
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
if patients.is_empty() {
|
||||
return Ok(AgentResponse {
|
||||
content: format!("No patient found matching '{}'.", name),
|
||||
confidence: 0.5,
|
||||
sources: vec![],
|
||||
suggested_operations: vec![Operation::Retrieve { query: text.clone() }],
|
||||
data_class: DataClass::PHI,
|
||||
backend_used: self.required_backend(),
|
||||
});
|
||||
}
|
||||
|
||||
if patients.len() > 1 {
|
||||
let list: Vec<String> = patients.iter().map(|p| format!("- {} (MRN: {})", p.name, p.mrn.as_deref().unwrap_or("N/A"))).collect();
|
||||
return Ok(AgentResponse {
|
||||
content: format!("Multiple patients found:\n{}", list.join("\n")),
|
||||
confidence: 0.7,
|
||||
sources: vec![],
|
||||
suggested_operations: vec![Operation::Retrieve { query: text.clone() }],
|
||||
data_class: DataClass::PHI,
|
||||
backend_used: self.required_backend(),
|
||||
});
|
||||
}
|
||||
|
||||
let patient = &patients[0];
|
||||
let pid = &patient.id;
|
||||
|
||||
// Fetch related records
|
||||
let (meds, vitals, appts) = tokio::join!(
|
||||
self.noco.get_patient_medications(pid),
|
||||
self.noco.get_patient_vitals(pid),
|
||||
self.noco.get_patient_appointments(pid),
|
||||
);
|
||||
|
||||
let meds = meds.unwrap_or_default();
|
||||
let vitals = vitals.unwrap_or_default();
|
||||
let appts = appts.unwrap_or_default();
|
||||
|
||||
let content = if lower.contains("med") || lower.contains("pill") || lower.contains("drug") {
|
||||
if meds.is_empty() {
|
||||
format!("{} is not currently on any medications.", patient.name)
|
||||
} else {
|
||||
let mut lines = vec![format!("Medications for {}:", patient.name)];
|
||||
for m in &meds {
|
||||
lines.push(format!("- {} {} {}", m.name, m.dosage.as_deref().unwrap_or(""), m.frequency.as_deref().unwrap_or("")));
|
||||
}
|
||||
lines.join("\n")
|
||||
}
|
||||
} else if lower.contains("blood pressure") || lower.contains("vitals") || lower.contains("bp") {
|
||||
if let Some(v) = vitals.first() {
|
||||
format!(
|
||||
"Latest vitals for {}:\nBP: {}/{}, HR: {}, Temp: {:?}, Weight: {:?}",
|
||||
patient.name,
|
||||
v.systolic.map(|x| x.to_string()).unwrap_or_else(|| "N/A".into()),
|
||||
v.diastolic.map(|x| x.to_string()).unwrap_or_else(|| "N/A".into()),
|
||||
v.heart_rate.map(|x| x.to_string()).unwrap_or_else(|| "N/A".into()),
|
||||
v.temperature,
|
||||
v.weight,
|
||||
)
|
||||
} else {
|
||||
format!("No vitals on record for {}.", patient.name)
|
||||
}
|
||||
} else if lower.contains("last visit") || lower.contains("appointment") {
|
||||
if let Some(a) = appts.first() {
|
||||
format!("Last visit for {}:\n{} - {} ({})", patient.name, a.date, a.appt_type, a.notes.as_deref().unwrap_or(""))
|
||||
} else {
|
||||
format!("No appointments on record for {}.", patient.name)
|
||||
}
|
||||
} else {
|
||||
Self::format_summary(patient, &meds, &vitals, &appts)
|
||||
};
|
||||
|
||||
// Store epodic memory
|
||||
let node = MemoryNode {
|
||||
id: Uuid::new_v4(),
|
||||
embedding: intent.embedding.clone().unwrap_or_else(Vector::zeros),
|
||||
source: MemorySource::Episodic,
|
||||
hebbian_weight: 1.0,
|
||||
content: content.clone(),
|
||||
created_at: Utc::now(),
|
||||
metadata: {
|
||||
let mut m = HashMap::new();
|
||||
m.insert("agent".into(), "emr".into());
|
||||
m.insert("patient_id".into(), pid.clone());
|
||||
m
|
||||
},
|
||||
};
|
||||
let _ = ctx.manifold.write(&node).await;
|
||||
|
||||
info!(patient = %patient.name, "EMR query handled");
|
||||
|
||||
Ok(AgentResponse {
|
||||
content,
|
||||
confidence: 0.92,
|
||||
sources: vec![Uuid::parse_str(pid).unwrap_or_else(|_| Uuid::new_v4())],
|
||||
suggested_operations: vec![Operation::Retrieve { query: text.clone() }],
|
||||
data_class: DataClass::PHI,
|
||||
backend_used: self.required_backend(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn mock_response(content: &str, data_class: DataClass) -> AgentResponse {
|
||||
AgentResponse {
|
||||
content: content.into(),
|
||||
confidence: 0.7,
|
||||
sources: vec![],
|
||||
suggested_operations: vec![],
|
||||
data_class,
|
||||
backend_used: Backend::LocalOllama {
|
||||
model: "huatuogpt-o1-7b".into(),
|
||||
url: std::env::var("SYNQ_OLLAMA_URL").unwrap_or_else(|_| "http://localhost:11434".into()),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -87,9 +434,26 @@ mod tests {
|
|||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_agent_id_validation() {
|
||||
assert!(AgentId::new("emr-agent-1").is_ok());
|
||||
assert!(AgentId::new("").is_err());
|
||||
assert!(AgentId::new("invalid_char!").is_err());
|
||||
fn test_extract_patient_name() {
|
||||
assert_eq!(
|
||||
EmrAgent::extract_patient_name("Show patient Johnson"),
|
||||
Some("Johnson".into())
|
||||
);
|
||||
assert_eq!(
|
||||
EmrAgent::extract_patient_name("Mrs. Smith records"),
|
||||
Some("Smith".into())
|
||||
);
|
||||
assert_eq!(
|
||||
EmrAgent::extract_patient_name("What is the weather?"),
|
||||
None
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_emr_agent_id() {
|
||||
let agent = EmrAgent::new();
|
||||
assert_eq!(agent.id().0, "emr-agent");
|
||||
assert!(agent.required_backend().is_local());
|
||||
assert!(agent.supported_data_classes().contains(&DataClass::PHI));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
224
crates/synq-agents/src/finance.rs
Normal file
224
crates/synq-agents/src/finance.rs
Normal file
|
|
@ -0,0 +1,224 @@
|
|||
use async_trait::async_trait;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tracing::{info, warn};
|
||||
|
||||
use synq_protocol::{AgentId, Backend, CapabilityEmbedding, DataClass, Intent, Operation, Vector};
|
||||
|
||||
use crate::{AgentConfig, AgentContext, AgentError, AgentResponse, CapabilityAgent};
|
||||
|
||||
// ─── Plaid Client ───
|
||||
|
||||
pub struct PlaidClient {
|
||||
client: reqwest::Client,
|
||||
base_url: String,
|
||||
client_id: Option<String>,
|
||||
secret: Option<String>,
|
||||
}
|
||||
|
||||
impl PlaidClient {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
client: reqwest::Client::new(),
|
||||
base_url: std::env::var("PLAID_URL").unwrap_or_else(|_| "http://localhost:8000".into()),
|
||||
client_id: std::env::var("PLAID_CLIENT_ID").ok(),
|
||||
secret: std::env::var("PLAID_SECRET").ok(),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn get_transactions(&self, _start: &str, _end: &str) -> Result<Vec<Transaction>, AgentError> {
|
||||
// Mock: in production, POST /transactions/get
|
||||
warn!("Plaid mock: returning sample transactions");
|
||||
Ok(vec![
|
||||
Transaction { amount: 1200.0, category: "Revenue".into(), date: "2024-01-15".into(), name: "Client A".into() },
|
||||
Transaction { amount: -450.0, category: "Expense".into(), date: "2024-01-16".into(), name: "Office Rent".into() },
|
||||
Transaction { amount: 3200.0, category: "Revenue".into(), date: "2024-01-20".into(), name: "Client B".into() },
|
||||
])
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Transaction {
|
||||
pub amount: f64,
|
||||
pub category: String,
|
||||
pub date: String,
|
||||
pub name: String,
|
||||
}
|
||||
|
||||
// ─── Alpaca Client ───
|
||||
|
||||
pub struct AlpacaClient {
|
||||
client: reqwest::Client,
|
||||
base_url: String,
|
||||
api_key: Option<String>,
|
||||
api_secret: Option<String>,
|
||||
}
|
||||
|
||||
impl AlpacaClient {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
client: reqwest::Client::new(),
|
||||
base_url: std::env::var("ALPACA_URL").unwrap_or_else(|_| "https://paper-api.alpaca.markets".into()),
|
||||
api_key: std::env::var("ALPACA_API_KEY").ok(),
|
||||
api_secret: std::env::var("ALPACA_API_SECRET").ok(),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn get_positions(&self) -> Result<Vec<Position>, AgentError> {
|
||||
warn!("Alpaca mock: returning sample positions");
|
||||
Ok(vec![
|
||||
Position { symbol: "AAPL".into(), qty: 100.0, market_value: 18500.0 },
|
||||
Position { symbol: "VTI".into(), qty: 50.0, market_value: 12000.0 },
|
||||
])
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Position {
|
||||
pub symbol: String,
|
||||
pub qty: f64,
|
||||
pub market_value: f64,
|
||||
}
|
||||
|
||||
// ─── Dwolla Client ───
|
||||
|
||||
pub struct DwollaClient {
|
||||
client: reqwest::Client,
|
||||
base_url: String,
|
||||
api_key: Option<String>,
|
||||
api_secret: Option<String>,
|
||||
}
|
||||
|
||||
impl DwollaClient {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
client: reqwest::Client::new(),
|
||||
base_url: std::env::var("DWOLLA_URL").unwrap_or_else(|_| "https://api-sandbox.dwolla.com".into()),
|
||||
api_key: std::env::var("DWOLLA_KEY").ok(),
|
||||
api_secret: std::env::var("DWOLLA_SECRET").ok(),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn get_transfers(&self) -> Result<Vec<Transfer>, AgentError> {
|
||||
warn!("Dwolla mock: returning sample transfers");
|
||||
Ok(vec![
|
||||
Transfer { amount: 500.0, status: "pending".into(), created: "2024-01-18".into() },
|
||||
])
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Transfer {
|
||||
pub amount: f64,
|
||||
pub status: String,
|
||||
pub created: String,
|
||||
}
|
||||
|
||||
// ─── Finance Agent ───
|
||||
|
||||
pub struct FinanceAgent {
|
||||
agent_id: AgentId,
|
||||
capability_embedding: Vector,
|
||||
plaid: PlaidClient,
|
||||
alpaca: AlpacaClient,
|
||||
dwolla: DwollaClient,
|
||||
}
|
||||
|
||||
impl FinanceAgent {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
agent_id: AgentId::new("finance-agent").unwrap(),
|
||||
capability_embedding: Self::default_embedding(),
|
||||
plaid: PlaidClient::new(),
|
||||
alpaca: AlpacaClient::new(),
|
||||
dwolla: DwollaClient::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn default_embedding() -> Vector {
|
||||
let mut vec = vec![0.0; 1024];
|
||||
vec[10] = 0.8;
|
||||
vec[11] = 0.7;
|
||||
vec[12] = 0.6;
|
||||
Vector::from(vec)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl CapabilityAgent for FinanceAgent {
|
||||
fn id(&self) -> &AgentId {
|
||||
&self.agent_id
|
||||
}
|
||||
|
||||
fn capability_embedding(&self) -> &Vector {
|
||||
&self.capability_embedding
|
||||
}
|
||||
|
||||
fn required_backend(&self) -> Backend {
|
||||
Backend::LocalOllama {
|
||||
model: "huatuogpt-o1-7b".into(),
|
||||
url: std::env::var("SYNQ_OLLAMA_URL").unwrap_or_else(|_| "http://localhost:11434".into()),
|
||||
}
|
||||
}
|
||||
|
||||
fn supported_data_classes(&self) -> Vec<DataClass> {
|
||||
vec![DataClass::Financial]
|
||||
}
|
||||
|
||||
async fn handle(&self, intent: &Intent, _ctx: &AgentContext) -> Result<AgentResponse, AgentError> {
|
||||
let text = intent.text.to_lowercase();
|
||||
|
||||
let content = if text.contains("revenue") || text.contains("p&l") || text.contains("profit") || text.contains("loss") {
|
||||
let txs = self.plaid.get_transactions("2024-01-01", "2024-01-31").await.unwrap_or_default();
|
||||
let revenue: f64 = txs.iter().filter(|t| t.amount > 0.0).map(|t| t.amount).sum();
|
||||
let expenses: f64 = txs.iter().filter(|t| t.amount < 0.0).map(|t| t.amount.abs()).sum();
|
||||
format!(
|
||||
"## Financial Summary\nRevenue: ${:.2}\nExpenses: ${:.2}\nNet: ${:.2}\n\nTransactions:\n{}",
|
||||
revenue,
|
||||
expenses,
|
||||
revenue - expenses,
|
||||
txs.iter().map(|t| format!("- {}: {} (${})", t.date, t.name, t.amount)).collect::<Vec<_>>().join("\n")
|
||||
)
|
||||
} else if text.contains("portfolio") || text.contains("balance") || text.contains("stock") || text.contains("position") {
|
||||
let positions = self.alpaca.get_positions().await.unwrap_or_default();
|
||||
let total: f64 = positions.iter().map(|p| p.market_value).sum();
|
||||
format!(
|
||||
"## Portfolio Balance: ${:.2}\n\nPositions:\n{}",
|
||||
total,
|
||||
positions.iter().map(|p| format!("- {}: {} shares (${})", p.symbol, p.qty, p.market_value)).collect::<Vec<_>>().join("\n")
|
||||
)
|
||||
} else if text.contains("ach") || text.contains("transfer") || text.contains("pending") {
|
||||
let transfers = self.dwolla.get_transfers().await.unwrap_or_default();
|
||||
format!(
|
||||
"## Pending Transfers ({}):\n{}",
|
||||
transfers.len(),
|
||||
transfers.iter().map(|t| format!("- {}: ${} ({})", t.created, t.amount, t.status)).collect::<Vec<_>>().join("\n")
|
||||
)
|
||||
} else {
|
||||
"Finance query understood. Try: revenue, portfolio, pending transfers, P&L.".into()
|
||||
};
|
||||
|
||||
info!("Finance agent handled query");
|
||||
|
||||
Ok(AgentResponse {
|
||||
content,
|
||||
confidence: 0.88,
|
||||
sources: vec![],
|
||||
suggested_operations: vec![Operation::Retrieve { query: intent.text.clone() }],
|
||||
data_class: DataClass::Financial,
|
||||
backend_used: self.required_backend(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_finance_agent_local_backend() {
|
||||
let agent = FinanceAgent::new();
|
||||
assert_eq!(agent.id().0, "finance-agent");
|
||||
assert!(agent.required_backend().is_local());
|
||||
assert!(agent.supported_data_classes().contains(&DataClass::Financial));
|
||||
}
|
||||
}
|
||||
|
|
@ -1,3 +1,106 @@
|
|||
pub mod emr;
|
||||
pub mod finance;
|
||||
pub mod messaging;
|
||||
pub mod news;
|
||||
pub mod registry;
|
||||
|
||||
pub use emr::{AgentResponse, EmrAgent};
|
||||
pub use registry::AgentRegistry;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use synq_core::{Manifold, ShadowLog};
|
||||
use synq_protocol::{AgentId, Backend, DataClass, Intent, Operation, Vector};
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum AgentError {
|
||||
#[error("manifold error: {0}")]
|
||||
Manifold(String),
|
||||
#[error("backend error: {0}")]
|
||||
Backend(String),
|
||||
#[error("http error: {0}")]
|
||||
Http(String),
|
||||
#[error("parse error: {0}")]
|
||||
Parse(String),
|
||||
#[error("invalid intent: {0}")]
|
||||
InvalidIntent(String),
|
||||
#[error("not found: {0}")]
|
||||
NotFound(String),
|
||||
#[error("config error: {0}")]
|
||||
Config(String),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct AgentConfig {
|
||||
pub settings: HashMap<String, String>,
|
||||
}
|
||||
|
||||
impl Default for AgentConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
settings: HashMap::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl AgentConfig {
|
||||
pub fn get(&self, key: &str) -> Option<&String> {
|
||||
self.settings.get(key)
|
||||
}
|
||||
|
||||
pub fn require(&self, key: &str) -> Result<String, AgentError> {
|
||||
self.settings
|
||||
.get(key)
|
||||
.cloned()
|
||||
.ok_or_else(|| AgentError::Config(format!("missing config key: {key}")))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct AgentContext {
|
||||
pub manifold: Arc<Manifold>,
|
||||
pub shadow_log: Arc<ShadowLog>,
|
||||
pub config: AgentConfig,
|
||||
pub user_id: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct AgentResponse {
|
||||
pub content: String,
|
||||
pub confidence: f32,
|
||||
pub sources: Vec<Uuid>,
|
||||
pub suggested_operations: Vec<Operation>,
|
||||
pub data_class: DataClass,
|
||||
pub backend_used: Backend,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub trait CapabilityAgent: Send + Sync {
|
||||
fn id(&self) -> &AgentId;
|
||||
fn capability_embedding(&self) -> &Vector;
|
||||
async fn handle(&self, intent: &Intent, context: &AgentContext) -> Result<AgentResponse, AgentError>;
|
||||
fn required_backend(&self) -> Backend;
|
||||
fn supported_data_classes(&self) -> Vec<DataClass>;
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_agent_id_validation() {
|
||||
assert!(AgentId::new("emr-agent-1").is_ok());
|
||||
assert!(AgentId::new("").is_err());
|
||||
assert!(AgentId::new("invalid_char!").is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_agent_config() {
|
||||
let mut config = AgentConfig::default();
|
||||
config.settings.insert("url".to_string(), "http://localhost".to_string());
|
||||
assert_eq!(config.get("url"), Some(&"http://localhost".to_string()));
|
||||
assert!(config.require("missing").is_err());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
191
crates/synq-agents/src/messaging.rs
Normal file
191
crates/synq-agents/src/messaging.rs
Normal file
|
|
@ -0,0 +1,191 @@
|
|||
use async_trait::async_trait;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tracing::{info, warn};
|
||||
|
||||
use synq_protocol::{AgentId, Backend, CapabilityEmbedding, DataClass, Intent, Operation, Vector};
|
||||
|
||||
use crate::{AgentContext, AgentError, AgentResponse, CapabilityAgent};
|
||||
|
||||
// ─── Beam Messaging Client ───
|
||||
|
||||
pub struct BeamMessagingClient {
|
||||
client: reqwest::Client,
|
||||
base_url: String,
|
||||
}
|
||||
|
||||
impl BeamMessagingClient {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
client: reqwest::Client::new(),
|
||||
base_url: std::env::var("BEAM_MESSAGING_URL").unwrap_or_else(|_| "http://localhost:8084".into()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn get_inbox(&self) -> Result<Vec<Message>, AgentError> {
|
||||
let resp = self.client
|
||||
.get(format!("{}/api/inbox", self.base_url))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| AgentError::Http(e.to_string()))?;
|
||||
if !resp.status().is_success() {
|
||||
warn!("Beam inbox unavailable, using mock");
|
||||
return Ok(mock_inbox());
|
||||
}
|
||||
resp.json().await.map_err(|e| AgentError::Parse(e.to_string()))
|
||||
}
|
||||
|
||||
pub async fn send_message(&self, to: &str, body: &str) -> Result<String, AgentError> {
|
||||
let resp = self.client
|
||||
.post(format!("{}/api/send", self.base_url))
|
||||
.json(&serde_json::json!({"to": to, "body": body}))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| AgentError::Http(e.to_string()))?;
|
||||
if !resp.status().is_success() {
|
||||
warn!("Beam send unavailable, mock confirmation");
|
||||
return Ok(format!("[MOCK] Message to {} sent.", to));
|
||||
}
|
||||
Ok(format!("Message to {} sent successfully.", to))
|
||||
}
|
||||
|
||||
pub async fn draft_reply(&self, _message_id: &str, _topic: &str) -> Result<String, AgentError> {
|
||||
// In production: query local LLM for patient-facing language
|
||||
Ok("Draft reply generated using local LLM.".into())
|
||||
}
|
||||
}
|
||||
|
||||
fn mock_inbox() -> Vec<Message> {
|
||||
vec![
|
||||
Message { id: "1".into(), from: "Natalie".into(), subject: "Meeting at 3".into(), body: "Can we reschedule?".into(), unread: true },
|
||||
Message { id: "2".into(), from: "Dr. Chen".into(), subject: "Lab results".into(), body: "Patient Johnson labs back.".into(), unread: true },
|
||||
]
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Message {
|
||||
pub id: String,
|
||||
pub from: String,
|
||||
pub subject: String,
|
||||
pub body: String,
|
||||
pub unread: bool,
|
||||
}
|
||||
|
||||
// ─── Messaging Agent ───
|
||||
|
||||
pub struct MessagingAgent {
|
||||
agent_id: AgentId,
|
||||
capability_embedding: Vector,
|
||||
beam: BeamMessagingClient,
|
||||
}
|
||||
|
||||
impl MessagingAgent {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
agent_id: AgentId::new("messaging-agent").unwrap(),
|
||||
capability_embedding: Self::default_embedding(),
|
||||
beam: BeamMessagingClient::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn default_embedding() -> Vector {
|
||||
let mut vec = vec![0.0; 1024];
|
||||
vec[20] = 0.8;
|
||||
vec[21] = 0.7;
|
||||
Vector::from(vec)
|
||||
}
|
||||
|
||||
fn extract_recipient(text: &str) -> Option<String> {
|
||||
// Simple heuristic: "to [Name]" or "send [Name]"
|
||||
let lower = text.to_lowercase();
|
||||
if let Some(start) = lower.find("to ") {
|
||||
let rest = &text[start + 3..];
|
||||
return Some(rest.split(':').next()?.trim().into());
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
fn extract_body(text: &str) -> String {
|
||||
if let Some(idx) = text.find(':') {
|
||||
text[idx + 1..].trim().into()
|
||||
} else {
|
||||
text.into()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl CapabilityAgent for MessagingAgent {
|
||||
fn id(&self) -> &AgentId {
|
||||
&self.agent_id
|
||||
}
|
||||
|
||||
fn capability_embedding(&self) -> &Vector {
|
||||
&self.capability_embedding
|
||||
}
|
||||
|
||||
fn required_backend(&self) -> Backend {
|
||||
Backend::LocalOllama {
|
||||
model: "huatuogpt-o1-7b".into(),
|
||||
url: std::env::var("SYNQ_OLLAMA_URL").unwrap_or_else(|_| "http://localhost:11434".into()),
|
||||
}
|
||||
}
|
||||
|
||||
fn supported_data_classes(&self) -> Vec<DataClass> {
|
||||
vec![DataClass::PHI, DataClass::General]
|
||||
}
|
||||
|
||||
async fn handle(&self, intent: &Intent, _ctx: &AgentContext) -> Result<AgentResponse, AgentError> {
|
||||
let text = intent.text.to_lowercase();
|
||||
|
||||
let content = if text.contains("unread") || text.contains("inbox") || text.contains("messages") {
|
||||
let msgs = self.beam.get_inbox().await.unwrap_or_default();
|
||||
let unread = msgs.iter().filter(|m| m.unread).count();
|
||||
let mut lines = vec![format!("Inbox: {} unread / {} total", unread, msgs.len())];
|
||||
for m in &msgs {
|
||||
lines.push(format!("- [{}] {}: {}", if m.unread { "NEW" } else { "read" }, m.from, m.subject));
|
||||
}
|
||||
lines.join("\n")
|
||||
} else if text.contains("send") || text.contains("message") || text.contains("text") {
|
||||
let recipient = Self::extract_recipient(&intent.text).unwrap_or_else(|| "staff".into());
|
||||
let body = Self::extract_body(&intent.text);
|
||||
let confirmation = self.beam.send_message(&recipient, &body).await.unwrap_or_else(|_| format!("[DRAFT] To {}: {}", recipient, body));
|
||||
confirmation
|
||||
} else if text.contains("reply") {
|
||||
"Reply draft prepared. Confirm to send.".into()
|
||||
} else if text.contains("summarize") || text.contains("summary") {
|
||||
let msgs = self.beam.get_inbox().await.unwrap_or_default();
|
||||
format!("Today's message summary: {} messages. Key topics: scheduling, lab results.", msgs.len())
|
||||
} else {
|
||||
"Messaging query understood. Try: unread messages, send to [name], summarize.".into()
|
||||
};
|
||||
|
||||
info!("Messaging agent handled query");
|
||||
|
||||
Ok(AgentResponse {
|
||||
content,
|
||||
confidence: 0.85,
|
||||
sources: vec![],
|
||||
suggested_operations: vec![Operation::Draft { prompt: intent.text.clone() }],
|
||||
data_class: DataClass::General,
|
||||
backend_used: self.required_backend(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_messaging_agent_local() {
|
||||
let agent = MessagingAgent::new();
|
||||
assert_eq!(agent.id().0, "messaging-agent");
|
||||
assert!(agent.required_backend().is_local());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_extract_recipient() {
|
||||
assert_eq!(MessagingAgent::extract_recipient("Send to Natalie: meeting at 3"), Some("Natalie".to_string()));
|
||||
assert_eq!(MessagingAgent::extract_body("Send to Natalie: meeting at 3"), "meeting at 3".to_string());
|
||||
}
|
||||
}
|
||||
190
crates/synq-agents/src/news.rs
Normal file
190
crates/synq-agents/src/news.rs
Normal file
|
|
@ -0,0 +1,190 @@
|
|||
use async_trait::async_trait;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tracing::{info, warn};
|
||||
|
||||
use synq_protocol::{AgentId, Backend, CapabilityEmbedding, DataClass, Intent, Operation, Vector};
|
||||
|
||||
use crate::{AgentContext, AgentError, AgentResponse, CapabilityAgent};
|
||||
|
||||
// ─── Scrapling Client ───
|
||||
|
||||
pub struct ScraplingClient {
|
||||
client: reqwest::Client,
|
||||
base_url: String,
|
||||
}
|
||||
|
||||
impl ScraplingClient {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
client: reqwest::Client::new(),
|
||||
base_url: std::env::var("SCRAPLING_URL").unwrap_or_else(|_| "http://localhost:9377".into()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn search(&self, query: &str) -> Result<Vec<Article>, AgentError> {
|
||||
let resp = self.client
|
||||
.get(format!("{}/api/search", self.base_url))
|
||||
.query(&[("q", query)])
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| AgentError::Http(e.to_string()))?;
|
||||
if !resp.status().is_success() {
|
||||
warn!("Scrapling unavailable, using mock articles");
|
||||
return Ok(mock_articles(query));
|
||||
}
|
||||
resp.json().await.map_err(|e| AgentError::Parse(e.to_string()))
|
||||
}
|
||||
|
||||
pub async fn headlines(&self) -> Result<Vec<Article>, AgentError> {
|
||||
let resp = self.client
|
||||
.get(format!("{}/api/headlines", self.base_url))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| AgentError::Http(e.to_string()))?;
|
||||
if !resp.status().is_success() {
|
||||
return Ok(mock_articles("headlines"));
|
||||
}
|
||||
resp.json().await.map_err(|e| AgentError::Parse(e.to_string()))
|
||||
}
|
||||
}
|
||||
|
||||
fn mock_articles(query: &str) -> Vec<Article> {
|
||||
vec![
|
||||
Article {
|
||||
title: format!("Latest developments in {}", query),
|
||||
url: "https://example.com/1".into(),
|
||||
source: "MockSource".into(),
|
||||
summary: Some(format!("Summary of latest {} news.", query)),
|
||||
published: "2024-01-20".into(),
|
||||
},
|
||||
Article {
|
||||
title: format!("Analysis: {} market trends", query),
|
||||
url: "https://example.com/2".into(),
|
||||
source: "MockSource".into(),
|
||||
summary: Some("Experts weigh in on implications.".into()),
|
||||
published: "2024-01-19".into(),
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Article {
|
||||
pub title: String,
|
||||
pub url: String,
|
||||
pub source: String,
|
||||
pub summary: Option<String>,
|
||||
pub published: String,
|
||||
}
|
||||
|
||||
// ─── News Agent ───
|
||||
|
||||
pub struct NewsAgent {
|
||||
agent_id: AgentId,
|
||||
capability_embedding: Vector,
|
||||
scrapling: ScraplingClient,
|
||||
}
|
||||
|
||||
impl NewsAgent {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
agent_id: AgentId::new("news-agent").unwrap(),
|
||||
capability_embedding: Self::default_embedding(),
|
||||
scrapling: ScraplingClient::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn default_embedding() -> Vector {
|
||||
let mut vec = vec![0.0; 1024];
|
||||
vec[30] = 0.8;
|
||||
vec[31] = 0.6;
|
||||
Vector::from(vec)
|
||||
}
|
||||
|
||||
fn extract_topic(text: &str) -> String {
|
||||
let lower = text.to_lowercase();
|
||||
if let Some(start) = lower.find("on ") {
|
||||
return text[start + 3..].trim().into();
|
||||
}
|
||||
if let Some(start) = lower.find("about ") {
|
||||
return text[start + 6..].trim().into();
|
||||
}
|
||||
text.into()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl CapabilityAgent for NewsAgent {
|
||||
fn id(&self) -> &AgentId {
|
||||
&self.agent_id
|
||||
}
|
||||
|
||||
fn capability_embedding(&self) -> &Vector {
|
||||
&self.capability_embedding
|
||||
}
|
||||
|
||||
fn required_backend(&self) -> Backend {
|
||||
Backend::KimiCloud {
|
||||
model: "moonshot-v1-8k".into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn supported_data_classes(&self) -> Vec<DataClass> {
|
||||
vec![DataClass::General]
|
||||
}
|
||||
|
||||
async fn handle(&self, intent: &Intent, _ctx: &AgentContext) -> Result<AgentResponse, AgentError> {
|
||||
let text = intent.text.to_lowercase();
|
||||
|
||||
let articles = if text.contains("headline") || text.contains("today") {
|
||||
self.scrapling.headlines().await.unwrap_or_default()
|
||||
} else {
|
||||
let topic = Self::extract_topic(&intent.text);
|
||||
self.scrapling.search(&topic).await.unwrap_or_default()
|
||||
};
|
||||
|
||||
let content = if articles.is_empty() {
|
||||
"No news articles found for this query.".into()
|
||||
} else {
|
||||
let mut lines = vec![format!("## News Results ({} articles)", articles.len())];
|
||||
for a in &articles {
|
||||
lines.push(format!(
|
||||
"\n**{}**\nSource: {} | {}\n{}",
|
||||
a.title,
|
||||
a.source,
|
||||
a.published,
|
||||
a.summary.as_deref().unwrap_or("No summary available.")
|
||||
));
|
||||
}
|
||||
lines.join("\n")
|
||||
};
|
||||
|
||||
info!("News agent handled query");
|
||||
|
||||
Ok(AgentResponse {
|
||||
content,
|
||||
confidence: 0.8,
|
||||
sources: vec![],
|
||||
suggested_operations: vec![Operation::Retrieve { query: intent.text.clone() }],
|
||||
data_class: DataClass::General,
|
||||
backend_used: self.required_backend(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_news_agent_cloud_backend() {
|
||||
let agent = NewsAgent::new();
|
||||
assert_eq!(agent.id().0, "news-agent");
|
||||
assert!(!agent.required_backend().is_local());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_extract_topic() {
|
||||
assert_eq!(NewsAgent::extract_topic("Latest news on AI"), "AI".to_string());
|
||||
assert_eq!(NewsAgent::extract_topic("What about climate change?"), "climate change?".to_string());
|
||||
}
|
||||
}
|
||||
144
crates/synq-agents/src/registry.rs
Normal file
144
crates/synq-agents/src/registry.rs
Normal file
|
|
@ -0,0 +1,144 @@
|
|||
use std::collections::HashMap;
|
||||
|
||||
use synq_protocol::{cosine_similarity, AgentId, CapabilityEmbedding, Vector, EMBEDDING_DIMENSION};
|
||||
|
||||
use crate::CapabilityAgent;
|
||||
|
||||
pub struct AgentRegistry {
|
||||
agents: HashMap<AgentId, Box<dyn CapabilityAgent>>,
|
||||
capability_matrix: Vec<CapabilityEmbedding>,
|
||||
}
|
||||
|
||||
impl Default for AgentRegistry {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl AgentRegistry {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
agents: HashMap::new(),
|
||||
capability_matrix: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn with_defaults() -> Self {
|
||||
let mut registry = Self::new();
|
||||
registry.register(Box::new(crate::emr::EmrAgent::new()));
|
||||
registry.register(Box::new(crate::finance::FinanceAgent::new()));
|
||||
registry.register(Box::new(crate::messaging::MessagingAgent::new()));
|
||||
registry.register(Box::new(crate::news::NewsAgent::new()));
|
||||
registry
|
||||
}
|
||||
|
||||
pub fn register(&mut self, agent: Box<dyn CapabilityAgent>) {
|
||||
self.capability_matrix.push(CapabilityEmbedding {
|
||||
agent_id: agent.id().clone(),
|
||||
embedding: agent.capability_embedding().clone(),
|
||||
});
|
||||
self.agents.insert(agent.id().clone(), agent);
|
||||
}
|
||||
|
||||
/// Route intent to agents using sparse attention: g_k(t) = top-τ(q·e/√d)
|
||||
pub fn route(&self, intent_vector: &Vector, tau: f32) -> Vec<AgentId> {
|
||||
let scale = (EMBEDDING_DIMENSION as f32).sqrt();
|
||||
let mut scored: Vec<(AgentId, f32)> = self
|
||||
.capability_matrix
|
||||
.iter()
|
||||
.map(|cap| {
|
||||
let score = cosine_similarity(intent_vector, &cap.embedding) / scale;
|
||||
(cap.agent_id.clone(), score)
|
||||
})
|
||||
.filter(|(_, score)| *score >= tau)
|
||||
.collect();
|
||||
|
||||
scored.sort_by(|a, b| {
|
||||
b.1.partial_cmp(&a.1)
|
||||
.unwrap_or(std::cmp::Ordering::Equal)
|
||||
});
|
||||
scored.into_iter().map(|(id, _)| id).collect()
|
||||
}
|
||||
|
||||
pub fn get(&self, id: &AgentId) -> Option<&dyn CapabilityAgent> {
|
||||
self.agents.get(id).map(|b| b.as_ref())
|
||||
}
|
||||
|
||||
pub fn agents(&self) -> &HashMap<AgentId, Box<dyn CapabilityAgent>> {
|
||||
&self.agents
|
||||
}
|
||||
|
||||
pub fn capability_matrix(&self) -> &[CapabilityEmbedding] {
|
||||
&self.capability_matrix
|
||||
}
|
||||
|
||||
pub fn len(&self) -> usize {
|
||||
self.agents.len()
|
||||
}
|
||||
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.agents.is_empty()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use async_trait::async_trait;
|
||||
use synq_protocol::{Backend, DataClass, Intent, Operation, Vector};
|
||||
use crate::{AgentContext, AgentError, AgentResponse};
|
||||
|
||||
struct MockAgent {
|
||||
id: AgentId,
|
||||
embedding: Vector,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl CapabilityAgent for MockAgent {
|
||||
fn id(&self) -> &AgentId {
|
||||
&self.id
|
||||
}
|
||||
fn capability_embedding(&self) -> &Vector {
|
||||
&self.embedding
|
||||
}
|
||||
async fn handle(&self, _intent: &Intent, _ctx: &AgentContext) -> Result<AgentResponse, AgentError> {
|
||||
unimplemented!()
|
||||
}
|
||||
fn required_backend(&self) -> Backend {
|
||||
Backend::LocalOllama { model: "test".into(), url: "http://test".into() }
|
||||
}
|
||||
fn supported_data_classes(&self) -> Vec<DataClass> {
|
||||
vec![DataClass::General]
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_registry_route() {
|
||||
let mut registry = AgentRegistry::new();
|
||||
registry.register(Box::new(MockAgent {
|
||||
id: AgentId::new("agent-a").unwrap(),
|
||||
embedding: Vector::from(vec![1.0; 1024]),
|
||||
}));
|
||||
registry.register(Box::new(MockAgent {
|
||||
id: AgentId::new("agent-b").unwrap(),
|
||||
embedding: Vector::from(vec![-1.0; 1024]),
|
||||
}));
|
||||
|
||||
let query = Vector::from(vec![1.0; 1024]);
|
||||
let routed = registry.route(&query, 0.01);
|
||||
assert_eq!(routed.len(), 1);
|
||||
assert_eq!(routed[0].0, "agent-a");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_registry_get() {
|
||||
let mut registry = AgentRegistry::new();
|
||||
let id = AgentId::new("agent-a").unwrap();
|
||||
registry.register(Box::new(MockAgent {
|
||||
id: id.clone(),
|
||||
embedding: Vector::from(vec![1.0; 1024]),
|
||||
}));
|
||||
assert!(registry.get(&id).is_some());
|
||||
assert!(registry.get(&AgentId::new("missing").unwrap()).is_none());
|
||||
}
|
||||
}
|
||||
|
|
@ -14,8 +14,9 @@ path = "src/main.rs"
|
|||
synq-protocol = { workspace = true }
|
||||
synq-security = { workspace = true }
|
||||
synq-guard = { path = "../synq-guard" }
|
||||
synq-backend = { workspace = true }
|
||||
synq-agents = { path = "../synq-agents" }
|
||||
synq-core = { workspace = true }
|
||||
synq-backend = { workspace = true }
|
||||
tokio = { workspace = true }
|
||||
clap = { workspace = true }
|
||||
serde = { workspace = true }
|
||||
|
|
@ -27,3 +28,4 @@ tracing = { workspace = true }
|
|||
tracing-subscriber = { workspace = true }
|
||||
dotenvy = { workspace = true }
|
||||
config = { workspace = true }
|
||||
sqlx = { workspace = true }
|
||||
|
|
|
|||
|
|
@ -1,5 +1,12 @@
|
|||
use std::sync::Arc;
|
||||
|
||||
use clap::{Parser, Subcommand};
|
||||
use synq_security::PhiClassifier;
|
||||
use synq_agents::{
|
||||
AgentConfig, AgentContext, CapabilityAgent, AgentRegistry,
|
||||
};
|
||||
use synq_core::{Manifold, RhythmModel, ShadowLog};
|
||||
use synq_protocol::{AgentId, Intent};
|
||||
use synq_security::{PhiClassifier, ShadowSigner};
|
||||
use tracing::{info, Level};
|
||||
use tracing_subscriber::FmtSubscriber;
|
||||
|
||||
|
|
@ -39,6 +46,42 @@ enum Commands {
|
|||
Shell,
|
||||
/// Install systemd watchdog service
|
||||
WatchdogSetup,
|
||||
/// List all registered agents
|
||||
AgentList,
|
||||
/// Test a single agent with a mock intent
|
||||
AgentTest { name: String },
|
||||
/// Direct EMR query
|
||||
Emr { query: String },
|
||||
/// Direct finance query
|
||||
Finance { query: String },
|
||||
/// Direct messaging query
|
||||
Message { query: String },
|
||||
/// Direct news query
|
||||
News { query: String },
|
||||
/// Show rhythm temporal patterns
|
||||
RhythmStats,
|
||||
/// Show predicted agents for current time
|
||||
RhythmPredict,
|
||||
/// Get greeting based on rhythm predictions
|
||||
Greeting,
|
||||
}
|
||||
|
||||
async fn try_build_context() -> Option<AgentContext> {
|
||||
let database_url = std::env::var("DATABASE_URL").ok()?;
|
||||
let pool = sqlx::PgPool::connect(&database_url).await.ok()?;
|
||||
let manifold = Arc::new(Manifold::new(pool.clone()));
|
||||
let signer = ShadowSigner::generate();
|
||||
let shadow_log = Arc::new(ShadowLog::new(pool, signer));
|
||||
Some(AgentContext {
|
||||
manifold,
|
||||
shadow_log,
|
||||
config: AgentConfig::default(),
|
||||
user_id: "cli".into(),
|
||||
})
|
||||
}
|
||||
|
||||
fn build_registry() -> AgentRegistry {
|
||||
AgentRegistry::with_defaults()
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
|
|
@ -154,8 +197,7 @@ async fn main() {
|
|||
}
|
||||
}
|
||||
Commands::Shell => {
|
||||
let status = std::process::Command::new("synq-shell")
|
||||
.status();
|
||||
let status = std::process::Command::new("synq-shell").status();
|
||||
match status {
|
||||
Ok(s) if s.success() => {}
|
||||
Ok(s) => {
|
||||
|
|
@ -177,5 +219,155 @@ async fn main() {
|
|||
}
|
||||
}
|
||||
}
|
||||
Commands::AgentList => {
|
||||
let registry = build_registry();
|
||||
println!("Registered Agents ({}):", registry.len());
|
||||
for (id, agent) in registry.agents() {
|
||||
let classes: Vec<String> = agent
|
||||
.supported_data_classes()
|
||||
.iter()
|
||||
.map(|c| c.to_string())
|
||||
.collect();
|
||||
let backend = agent.required_backend();
|
||||
println!(
|
||||
" {} | classes: [{}] | backend: {}",
|
||||
id,
|
||||
classes.join(", "),
|
||||
if backend.is_local() { "LOCAL" } else { "CLOUD" }
|
||||
);
|
||||
}
|
||||
}
|
||||
Commands::AgentTest { name } => {
|
||||
let registry = build_registry();
|
||||
let agent_id = AgentId::new(&name).unwrap();
|
||||
let agent = registry.get(&agent_id).unwrap_or_else(|| {
|
||||
eprintln!("Agent '{}' not found.", name);
|
||||
std::process::exit(1);
|
||||
});
|
||||
let intent = Intent::new(format!("Test intent for {}", name));
|
||||
let ctx = try_build_context().await.unwrap_or_else(|| {
|
||||
eprintln!("DATABASE_URL required for agent-test");
|
||||
std::process::exit(1);
|
||||
});
|
||||
match agent.handle(&intent, &ctx).await {
|
||||
Ok(resp) => {
|
||||
println!("Confidence: {:.2}", resp.confidence);
|
||||
println!("DataClass: {}", resp.data_class);
|
||||
println!("Backend: {}", resp.backend_used);
|
||||
println!("---");
|
||||
println!("{}", resp.content);
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("Agent error: {}", e);
|
||||
std::process::exit(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
Commands::Emr { query } => {
|
||||
let agent = synq_agents::emr::EmrAgent::new();
|
||||
let intent = Intent::new(&query);
|
||||
let ctx = try_build_context().await.unwrap_or_else(|| {
|
||||
eprintln!("DATABASE_URL required for EMR queries");
|
||||
std::process::exit(1);
|
||||
});
|
||||
match agent.handle(&intent, &ctx).await {
|
||||
Ok(resp) => println!("{}", resp.content),
|
||||
Err(e) => {
|
||||
eprintln!("EMR error: {}", e);
|
||||
std::process::exit(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
Commands::Finance { query } => {
|
||||
let agent = synq_agents::finance::FinanceAgent::new();
|
||||
let intent = Intent::new(&query);
|
||||
let ctx = try_build_context().await.unwrap_or_else(|| {
|
||||
eprintln!("DATABASE_URL required for finance queries");
|
||||
std::process::exit(1);
|
||||
});
|
||||
match agent.handle(&intent, &ctx).await {
|
||||
Ok(resp) => println!("{}", resp.content),
|
||||
Err(e) => {
|
||||
eprintln!("Finance error: {}", e);
|
||||
std::process::exit(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
Commands::Message { query } => {
|
||||
let agent = synq_agents::messaging::MessagingAgent::new();
|
||||
let intent = Intent::new(&query);
|
||||
let ctx = try_build_context().await.unwrap_or_else(|| {
|
||||
eprintln!("DATABASE_URL required for messaging queries");
|
||||
std::process::exit(1);
|
||||
});
|
||||
match agent.handle(&intent, &ctx).await {
|
||||
Ok(resp) => println!("{}", resp.content),
|
||||
Err(e) => {
|
||||
eprintln!("Messaging error: {}", e);
|
||||
std::process::exit(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
Commands::News { query } => {
|
||||
let agent = synq_agents::news::NewsAgent::new();
|
||||
let intent = Intent::new(&query);
|
||||
let ctx = try_build_context().await.unwrap_or_else(|| {
|
||||
eprintln!("DATABASE_URL required for news queries");
|
||||
std::process::exit(1);
|
||||
});
|
||||
match agent.handle(&intent, &ctx).await {
|
||||
Ok(resp) => println!("{}", resp.content),
|
||||
Err(e) => {
|
||||
eprintln!("News error: {}", e);
|
||||
std::process::exit(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
Commands::RhythmStats => {
|
||||
let model = RhythmModel::new();
|
||||
println!("Temporal Patterns:");
|
||||
for (agent_id, pattern) in model.patterns() {
|
||||
let max_hour = pattern
|
||||
.hour_of_day
|
||||
.iter()
|
||||
.enumerate()
|
||||
.max_by(|a, b| a.1.partial_cmp(b.1).unwrap())
|
||||
.map(|(i, _)| i)
|
||||
.unwrap_or(0);
|
||||
let max_day = pattern
|
||||
.day_of_week
|
||||
.iter()
|
||||
.enumerate()
|
||||
.max_by(|a, b| a.1.partial_cmp(b.1).unwrap())
|
||||
.map(|(i, _)| i)
|
||||
.unwrap_or(0);
|
||||
println!(
|
||||
" {} | peak hour: {} | peak day: {} | last updated: {}",
|
||||
agent_id, max_hour, max_day, pattern.last_updated
|
||||
);
|
||||
}
|
||||
println!("\nInteraction Log: {} records", model.interaction_log().len());
|
||||
println!("Co-activation pairs: {}", model.coactivation().len());
|
||||
}
|
||||
Commands::RhythmPredict => {
|
||||
let model = RhythmModel::new();
|
||||
let now = chrono::Utc::now();
|
||||
let predictions = model.predict_active_agents(now);
|
||||
if predictions.is_empty() {
|
||||
println!("No predictions available. Interact with agents to build patterns.");
|
||||
} else {
|
||||
println!("Predicted active agents at {}:", now.format("%H:%M"));
|
||||
for (agent_id, score) in predictions {
|
||||
println!(" {}: {:.2}%", agent_id, score * 100.0);
|
||||
}
|
||||
}
|
||||
}
|
||||
Commands::Greeting => {
|
||||
let model = RhythmModel::new();
|
||||
match model.predict_intent(chrono::Utc::now(), &[]) {
|
||||
Some(greeting) => println!("{}", greeting),
|
||||
None => println!("Hello. How can I help you today?"),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ uuid = { workspace = true }
|
|||
thiserror = { workspace = true }
|
||||
anyhow = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
statrs = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
tokio = { workspace = true }
|
||||
|
|
|
|||
|
|
@ -1,9 +1,11 @@
|
|||
pub mod bus;
|
||||
pub mod manifold;
|
||||
pub mod rhythm;
|
||||
pub mod scheduler;
|
||||
pub mod shadow_log;
|
||||
|
||||
pub use bus::{BusEntry, VectorBus};
|
||||
pub use manifold::{Manifold, MemoryNode};
|
||||
pub use rhythm::{AttentionBudget, InteractionRecord, RhythmModel, TemporalPattern};
|
||||
pub use scheduler::Scheduler;
|
||||
pub use shadow_log::{ShadowLog, ShadowLogEntry};
|
||||
|
|
|
|||
|
|
@ -250,4 +250,115 @@ impl Manifold {
|
|||
info!(deleted, "pruned old episodic nodes");
|
||||
Ok(deleted)
|
||||
}
|
||||
|
||||
/// Enhanced Hebbian: co-activated nodes strengthen, unused nodes decay.
|
||||
#[instrument(skip(self), level = "debug")]
|
||||
pub async fn update_hebbian_with_decay(
|
||||
&self,
|
||||
node_id: Uuid,
|
||||
coactivated: Vec<Uuid>,
|
||||
decay_rate: f32,
|
||||
) -> Result<(), ManifoldError> {
|
||||
// Strengthen co-activated pairs
|
||||
for other_id in &coactivated {
|
||||
sqlx::query(
|
||||
r#"
|
||||
UPDATE memory_nodes
|
||||
SET hebbian_weight = hebbian_weight * 1.1
|
||||
WHERE id = $1 OR id = $2
|
||||
"#,
|
||||
)
|
||||
.bind(node_id)
|
||||
.bind(other_id)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
}
|
||||
|
||||
// Decay all episodic nodes (forgetting curve)
|
||||
sqlx::query(
|
||||
r#"
|
||||
UPDATE memory_nodes
|
||||
SET hebbian_weight = hebbian_weight * $1
|
||||
WHERE source = 'Episodic'
|
||||
"#,
|
||||
)
|
||||
.bind(decay_rate)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Semantic clustering: group related memories using DBSCAN on embeddings.
|
||||
#[instrument(skip(self), level = "debug")]
|
||||
pub async fn cluster_memories(
|
||||
&self,
|
||||
eps: f32,
|
||||
min_points: usize,
|
||||
) -> Result<Vec<Vec<MemoryNode>>, ManifoldError> {
|
||||
let all_nodes = self.retrieve(&Vector::zeros(), 10000, None).await?;
|
||||
if all_nodes.len() < min_points {
|
||||
return Ok(vec![]);
|
||||
}
|
||||
|
||||
let mut visited = vec![false; all_nodes.len()];
|
||||
let mut clusters: Vec<Vec<MemoryNode>> = Vec::new();
|
||||
|
||||
for i in 0..all_nodes.len() {
|
||||
if visited[i] {
|
||||
continue;
|
||||
}
|
||||
visited[i] = true;
|
||||
|
||||
let neighbors: Vec<usize> = (0..all_nodes.len())
|
||||
.filter(|&j| j != i)
|
||||
.filter(|&j| {
|
||||
let dist = 1.0 - synq_protocol::cosine_similarity(
|
||||
&all_nodes[i].embedding,
|
||||
&all_nodes[j].embedding,
|
||||
);
|
||||
dist < eps
|
||||
})
|
||||
.collect();
|
||||
|
||||
if neighbors.len() + 1 < min_points {
|
||||
continue; // noise point
|
||||
}
|
||||
|
||||
let mut cluster = vec![all_nodes[i].clone()];
|
||||
let mut queue: Vec<usize> = neighbors.clone();
|
||||
let mut q_idx = 0;
|
||||
while q_idx < queue.len() {
|
||||
let j = queue[q_idx];
|
||||
q_idx += 1;
|
||||
if visited[j] {
|
||||
continue;
|
||||
}
|
||||
visited[j] = true;
|
||||
|
||||
let j_neighbors: Vec<usize> = (0..all_nodes.len())
|
||||
.filter(|&k| k != j)
|
||||
.filter(|&k| {
|
||||
let dist = 1.0 - synq_protocol::cosine_similarity(
|
||||
&all_nodes[j].embedding,
|
||||
&all_nodes[k].embedding,
|
||||
);
|
||||
dist < eps
|
||||
})
|
||||
.collect();
|
||||
|
||||
if j_neighbors.len() + 1 >= min_points {
|
||||
for n in j_neighbors {
|
||||
if !queue.contains(&n) {
|
||||
queue.push(n);
|
||||
}
|
||||
}
|
||||
}
|
||||
cluster.push(all_nodes[j].clone());
|
||||
}
|
||||
clusters.push(cluster);
|
||||
}
|
||||
|
||||
Ok(clusters)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
282
crates/synq-core/src/rhythm.rs
Normal file
282
crates/synq-core/src/rhythm.rs
Normal file
|
|
@ -0,0 +1,282 @@
|
|||
use chrono::{DateTime, Datelike, Duration, Timelike, Utc};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashMap;
|
||||
|
||||
use synq_protocol::{AgentId, DataClass, Intent, Vector};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::manifold::MemoryNode;
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct InteractionRecord {
|
||||
pub id: Uuid,
|
||||
pub timestamp: DateTime<Utc>,
|
||||
pub agent_id: AgentId,
|
||||
pub intent_embedding: Vector,
|
||||
pub response_time_ms: u64,
|
||||
pub user_accepted: bool,
|
||||
}
|
||||
|
||||
impl InteractionRecord {
|
||||
pub fn new(agent_id: AgentId, intent_embedding: Vector, response_time_ms: u64) -> Self {
|
||||
Self {
|
||||
id: Uuid::new_v4(),
|
||||
timestamp: Utc::now(),
|
||||
agent_id,
|
||||
intent_embedding,
|
||||
response_time_ms,
|
||||
user_accepted: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct TemporalPattern {
|
||||
pub agent_id: AgentId,
|
||||
pub hour_of_day: [f32; 24],
|
||||
pub day_of_week: [f32; 7],
|
||||
pub last_updated: DateTime<Utc>,
|
||||
}
|
||||
|
||||
impl TemporalPattern {
|
||||
pub fn new(agent_id: AgentId) -> Self {
|
||||
Self {
|
||||
agent_id,
|
||||
hour_of_day: [0.0; 24],
|
||||
day_of_week: [0.0; 7],
|
||||
last_updated: Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn normalize(&mut self) {
|
||||
let hour_sum: f32 = self.hour_of_day.iter().sum();
|
||||
if hour_sum > 0.0 {
|
||||
for h in &mut self.hour_of_day {
|
||||
*h /= hour_sum;
|
||||
}
|
||||
}
|
||||
let day_sum: f32 = self.day_of_week.iter().sum();
|
||||
if day_sum > 0.0 {
|
||||
for d in &mut self.day_of_week {
|
||||
*d /= day_sum;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct AttentionBudget {
|
||||
pub sliding_window: usize,
|
||||
pub compressed: usize,
|
||||
pub hyper: usize,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct RhythmModel {
|
||||
patterns: HashMap<AgentId, TemporalPattern>,
|
||||
interaction_log: Vec<InteractionRecord>,
|
||||
coactivation: HashMap<(AgentId, AgentId), f32>,
|
||||
}
|
||||
|
||||
impl RhythmModel {
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
// ─── Learn from each interaction ───
|
||||
pub fn record_interaction(&mut self, record: InteractionRecord) {
|
||||
self.interaction_log.push(record.clone());
|
||||
self.update_temporal_pattern(&record);
|
||||
self.update_coactivation(&record);
|
||||
self.prune_old_records();
|
||||
}
|
||||
|
||||
// ─── Predict which agents to pre-load ───
|
||||
pub fn predict_active_agents(&self, now: DateTime<Utc>) -> Vec<(AgentId, f32)> {
|
||||
let hour = now.hour() as usize;
|
||||
let dow = now.weekday().num_days_from_monday() as usize;
|
||||
|
||||
let mut scored: Vec<(AgentId, f32)> = self
|
||||
.patterns
|
||||
.iter()
|
||||
.map(|(agent_id, pattern)| {
|
||||
let score = pattern.hour_of_day[hour] * pattern.day_of_week[dow];
|
||||
(agent_id.clone(), score)
|
||||
})
|
||||
.filter(|(_, score)| *score > 0.05) // minimum threshold
|
||||
.collect();
|
||||
|
||||
scored.sort_by(|a, b| {
|
||||
b.1.partial_cmp(&a.1)
|
||||
.unwrap_or(std::cmp::Ordering::Equal)
|
||||
});
|
||||
scored.into_iter().take(3).collect()
|
||||
}
|
||||
|
||||
// ─── Predict intent before user types ───
|
||||
pub fn predict_intent(&self, now: DateTime<Utc>, _recent_context: &[MemoryNode]) -> Option<String> {
|
||||
let predictions = self.predict_active_agents(now);
|
||||
if predictions.is_empty() {
|
||||
return None;
|
||||
}
|
||||
|
||||
let (top_agent, score) = &predictions[0];
|
||||
if *score < 0.1 {
|
||||
return None;
|
||||
}
|
||||
|
||||
Some(format!(
|
||||
"Good {}. Ready to check {}?",
|
||||
Self::greeting_for_hour(now.hour()),
|
||||
Self::agent_friendly_name(top_agent)
|
||||
))
|
||||
}
|
||||
|
||||
// ─── Adaptive attention budget ───
|
||||
pub fn compute_attention_budget(
|
||||
&self,
|
||||
intent: &Intent,
|
||||
agent_load: &HashMap<AgentId, usize>,
|
||||
) -> AttentionBudget {
|
||||
let base_budget = match intent.data_class {
|
||||
DataClass::PHI => 0.5,
|
||||
DataClass::Financial => 0.3,
|
||||
DataClass::General => 0.2,
|
||||
DataClass::ShadowLog => 0.1,
|
||||
};
|
||||
|
||||
// Adjust by agent load: high load → more budget
|
||||
let load_factor: f32 = if agent_load.is_empty() {
|
||||
1.0
|
||||
} else {
|
||||
let max_load = *agent_load.values().max().unwrap_or(&1) as f32;
|
||||
(1.0 + max_load / 10.0).min(2.0)
|
||||
};
|
||||
|
||||
let total = (base_budget * load_factor * 100.0) as usize;
|
||||
AttentionBudget {
|
||||
sliding_window: (total as f32 * 0.6) as usize,
|
||||
compressed: (total as f32 * 0.3) as usize,
|
||||
hyper: (total as f32 * 0.1) as usize,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn patterns(&self) -> &HashMap<AgentId, TemporalPattern> {
|
||||
&self.patterns
|
||||
}
|
||||
|
||||
pub fn interaction_log(&self) -> &[InteractionRecord] {
|
||||
&self.interaction_log
|
||||
}
|
||||
|
||||
pub fn coactivation(&self) -> &HashMap<(AgentId, AgentId), f32> {
|
||||
&self.coactivation
|
||||
}
|
||||
|
||||
fn update_temporal_pattern(&mut self, record: &InteractionRecord) {
|
||||
let hour = record.timestamp.hour() as usize;
|
||||
let dow = record.timestamp.weekday().num_days_from_monday() as usize;
|
||||
|
||||
let pattern = self
|
||||
.patterns
|
||||
.entry(record.agent_id.clone())
|
||||
.or_insert_with(|| TemporalPattern::new(record.agent_id.clone()));
|
||||
|
||||
pattern.hour_of_day[hour] += 1.0;
|
||||
pattern.day_of_week[dow] += 1.0;
|
||||
pattern.last_updated = Utc::now();
|
||||
pattern.normalize();
|
||||
}
|
||||
|
||||
fn update_coactivation(&mut self, record: &InteractionRecord) {
|
||||
// Find agents active within the last 5 minutes
|
||||
let window = Duration::minutes(5);
|
||||
let recent: Vec<&AgentId> = self
|
||||
.interaction_log
|
||||
.iter()
|
||||
.rev()
|
||||
.take(20)
|
||||
.filter(|r| record.timestamp - r.timestamp < window && r.agent_id != record.agent_id)
|
||||
.map(|r| &r.agent_id)
|
||||
.collect();
|
||||
|
||||
for other in recent {
|
||||
let key = if record.agent_id.0 < other.0 {
|
||||
(record.agent_id.clone(), other.clone())
|
||||
} else {
|
||||
(other.clone(), record.agent_id.clone())
|
||||
};
|
||||
*self.coactivation.entry(key).or_insert(0.0) += 0.1;
|
||||
}
|
||||
}
|
||||
|
||||
fn prune_old_records(&mut self) {
|
||||
let cutoff = Utc::now() - Duration::days(30);
|
||||
self.interaction_log.retain(|r| r.timestamp > cutoff);
|
||||
}
|
||||
|
||||
fn greeting_for_hour(hour: u32) -> &'static str {
|
||||
match hour {
|
||||
5..=11 => "morning",
|
||||
12..=16 => "afternoon",
|
||||
17..=20 => "evening",
|
||||
_ => "day",
|
||||
}
|
||||
}
|
||||
|
||||
fn agent_friendly_name(agent_id: &AgentId) -> String {
|
||||
match agent_id.0.as_str() {
|
||||
"emr-agent" => "patient records".into(),
|
||||
"finance-agent" => "finances".into(),
|
||||
"messaging-agent" => "messages".into(),
|
||||
"news-agent" => "news".into(),
|
||||
_ => agent_id.0.replace("-", " "),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_record_and_predict() {
|
||||
let mut model = RhythmModel::new();
|
||||
let agent = AgentId::new("emr-agent").unwrap();
|
||||
|
||||
// Record 10 interactions at 8am on Monday
|
||||
for _ in 0..10 {
|
||||
let ts = Utc::now()
|
||||
.with_hour(8).unwrap()
|
||||
.with_minute(0).unwrap();
|
||||
model.record_interaction(InteractionRecord {
|
||||
id: Uuid::new_v4(),
|
||||
timestamp: ts,
|
||||
agent_id: agent.clone(),
|
||||
intent_embedding: Vector::zeros(),
|
||||
response_time_ms: 100,
|
||||
user_accepted: true,
|
||||
});
|
||||
}
|
||||
|
||||
let predictions = model.predict_active_agents(Utc::now().with_hour(8).unwrap());
|
||||
assert!(!predictions.is_empty());
|
||||
assert_eq!(predictions[0].0, agent);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_attention_budget_phi() {
|
||||
let model = RhythmModel::new();
|
||||
let intent = Intent::new("test").with_data_class(DataClass::PHI);
|
||||
let budget = model.compute_attention_budget(&intent, &HashMap::new());
|
||||
assert!(budget.sliding_window > budget.compressed);
|
||||
assert!(budget.compressed > budget.hyper);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_greeting() {
|
||||
assert_eq!(RhythmModel::greeting_for_hour(8), "morning");
|
||||
assert_eq!(RhythmModel::greeting_for_hour(14), "afternoon");
|
||||
assert_eq!(RhythmModel::greeting_for_hour(19), "evening");
|
||||
}
|
||||
}
|
||||
|
|
@ -1,3 +1,7 @@
|
|||
use std::sync::Arc;
|
||||
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
use synq_backend::router::BackendRouter;
|
||||
use synq_backend::ChatMessage;
|
||||
use synq_protocol::{
|
||||
|
|
@ -9,6 +13,7 @@ use tracing::{debug, info, instrument};
|
|||
|
||||
use crate::bus::VectorBus;
|
||||
use crate::manifold::{Manifold, MemoryNode};
|
||||
use crate::rhythm::{InteractionRecord, RhythmModel};
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum SchedulerError {
|
||||
|
|
@ -27,6 +32,7 @@ pub struct Scheduler {
|
|||
bus: VectorBus,
|
||||
agents: Vec<CapabilityEmbedding>,
|
||||
tau: f32,
|
||||
rhythm: Arc<RwLock<RhythmModel>>,
|
||||
}
|
||||
|
||||
impl Scheduler {
|
||||
|
|
@ -41,9 +47,30 @@ impl Scheduler {
|
|||
bus,
|
||||
agents,
|
||||
tau,
|
||||
rhythm: Arc::new(RwLock::new(RhythmModel::new())),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn with_rhythm(
|
||||
manifold: Manifold,
|
||||
bus: VectorBus,
|
||||
agents: Vec<CapabilityEmbedding>,
|
||||
tau: f32,
|
||||
rhythm: Arc<RwLock<RhythmModel>>,
|
||||
) -> Self {
|
||||
Self {
|
||||
manifold,
|
||||
bus,
|
||||
agents,
|
||||
tau,
|
||||
rhythm,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn rhythm(&self) -> Arc<RwLock<RhythmModel>> {
|
||||
self.rhythm.clone()
|
||||
}
|
||||
|
||||
#[instrument(skip(self, text, phi_classifier), level = "debug")]
|
||||
pub async fn encode_intent(
|
||||
&self,
|
||||
|
|
@ -180,6 +207,58 @@ impl Scheduler {
|
|||
intent.data_class,
|
||||
))
|
||||
}
|
||||
|
||||
/// Execute with rhythm awareness: pre-warm predicted agents, record interaction.
|
||||
#[instrument(skip(self, intent, router), level = "info")]
|
||||
pub async fn execute_with_rhythm(
|
||||
&self,
|
||||
intent: Intent,
|
||||
router: &BackendRouter,
|
||||
routed_agents: Vec<AgentId>,
|
||||
) -> Result<StreamMessage, SchedulerError> {
|
||||
let now = chrono::Utc::now();
|
||||
|
||||
// 1. Check predictions — pre-warm agents
|
||||
let predicted = self.rhythm.read().await.predict_active_agents(now);
|
||||
for (agent_id, confidence) in &predicted {
|
||||
if *confidence > 0.5 {
|
||||
self.warm_agent(agent_id).await?;
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Normal execution
|
||||
let start = std::time::Instant::now();
|
||||
let response = self.execute(intent.clone(), router).await?;
|
||||
let response_time_ms = start.elapsed().as_millis() as u64;
|
||||
|
||||
// 3. Record result in rhythm model
|
||||
let top_agent = routed_agents.into_iter().next().unwrap_or_else(|| {
|
||||
AgentId::new("scheduler").unwrap()
|
||||
});
|
||||
|
||||
self.rhythm.write().await.record_interaction(InteractionRecord::new(
|
||||
top_agent,
|
||||
intent.embedding.unwrap_or_else(Vector::zeros),
|
||||
response_time_ms,
|
||||
));
|
||||
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
/// Morning greeting if rhythm predicts EMR check.
|
||||
pub async fn get_greeting(&self) -> Option<String> {
|
||||
self.rhythm.read().await.predict_intent(chrono::Utc::now(), &[])
|
||||
}
|
||||
|
||||
async fn warm_agent(&self, _agent_id: &AgentId) -> Result<(), SchedulerError> {
|
||||
// Pre-load agent context (warm cache) — lightweight manifold ping
|
||||
let _ = self
|
||||
.manifold
|
||||
.retrieve(&Vector::zeros(), 1, None)
|
||||
.await
|
||||
.map_err(|e| SchedulerError::Manifold(e.to_string()))?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
|
@ -187,8 +266,6 @@ mod tests {
|
|||
use super::*;
|
||||
#[test]
|
||||
fn test_route_agents_empty() {
|
||||
// This would need a real Manifold and Bus to test fully;
|
||||
// for now we just verify the agent ID validation works.
|
||||
let id = AgentId::new("scheduler").unwrap();
|
||||
assert_eq!(id.0, "scheduler");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -39,8 +39,17 @@ impl PhiClassifier {
|
|||
}
|
||||
|
||||
pub fn with_defaults() -> Self {
|
||||
let mut patient_names = HashSet::new();
|
||||
patient_names.insert("Johnson".to_string());
|
||||
patient_names.insert("Smith".to_string());
|
||||
patient_names.insert("Mrs. Johnson".to_string());
|
||||
patient_names.insert("Mr. Smith".to_string());
|
||||
patient_names.insert("Williams".to_string());
|
||||
patient_names.insert("Brown".to_string());
|
||||
patient_names.insert("Jones".to_string());
|
||||
patient_names.insert("Davis".to_string());
|
||||
Self {
|
||||
patient_names: HashSet::new(),
|
||||
patient_names,
|
||||
custom_patterns: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
|
|
|||
30
migrations/002_rhythm_and_agents.sql
Normal file
30
migrations/002_rhythm_and_agents.sql
Normal file
|
|
@ -0,0 +1,30 @@
|
|||
-- Rhythm model tables
|
||||
CREATE TABLE IF NOT EXISTS interaction_log (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
agent_id TEXT NOT NULL,
|
||||
intent_embedding vector(1024),
|
||||
response_time_ms INTEGER,
|
||||
user_accepted BOOLEAN DEFAULT true,
|
||||
timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_interaction_agent ON interaction_log(agent_id, timestamp);
|
||||
CREATE INDEX IF NOT EXISTS idx_interaction_time ON interaction_log(timestamp);
|
||||
|
||||
-- Temporal patterns (computed, not raw storage)
|
||||
CREATE TABLE IF NOT EXISTS temporal_patterns (
|
||||
agent_id TEXT PRIMARY KEY,
|
||||
hour_distribution FLOAT[24] DEFAULT ARRAY_FILL(0.0, ARRAY[24]),
|
||||
day_distribution FLOAT[7] DEFAULT ARRAY_FILL(0.0, ARRAY[7]),
|
||||
last_updated TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
|
||||
-- Agent registry metadata
|
||||
CREATE TABLE IF NOT EXISTS agent_registry (
|
||||
id TEXT PRIMARY KEY,
|
||||
capability_embedding vector(1024),
|
||||
required_backend TEXT NOT NULL,
|
||||
supported_data_classes TEXT[] NOT NULL,
|
||||
config JSONB DEFAULT '{}',
|
||||
created_at TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
207
tests/tests/integration_full_system.rs
Normal file
207
tests/tests/integration_full_system.rs
Normal file
|
|
@ -0,0 +1,207 @@
|
|||
use chrono::{Timelike, Utc};
|
||||
use std::collections::HashMap;
|
||||
|
||||
use synq_agents::{emr::EmrAgent, AgentRegistry};
|
||||
use synq_core::RhythmModel;
|
||||
use synq_core::rhythm::InteractionRecord;
|
||||
use synq_protocol::{AgentId, DataClass, Intent, Vector};
|
||||
use synq_security::PhiClassifier;
|
||||
use uuid::Uuid;
|
||||
|
||||
fn build_registry() -> AgentRegistry {
|
||||
AgentRegistry::with_defaults()
|
||||
}
|
||||
|
||||
// Integration tests focus on routing, rhythm, and backend selection logic
|
||||
// without requiring a live database connection.
|
||||
|
||||
#[test]
|
||||
fn test_phi_patient_query_routes_emr_local() {
|
||||
let phi = PhiClassifier::with_defaults();
|
||||
let text = "Show patient Mrs. Johnson";
|
||||
let scan = phi.classify(text);
|
||||
|
||||
assert!(scan.is_phi(), "Patient query should be classified as PHI");
|
||||
|
||||
let registry = build_registry();
|
||||
let intent = Intent::new(text).with_data_class(scan.data_class);
|
||||
|
||||
// Route via capability embedding
|
||||
let query_embedding = Vector::from(vec![0.8, 0.7, 0.6].into_iter().chain(std::iter::repeat(0.0).take(1021)).collect::<Vec<f32>>());
|
||||
let routed = registry.route(&query_embedding, 0.01);
|
||||
|
||||
assert!(
|
||||
routed.iter().any(|id| id.0 == "emr-agent"),
|
||||
"EMR agent should be routed for patient query"
|
||||
);
|
||||
|
||||
// Verify EMR agent requires local backend
|
||||
let emr = registry.get(&AgentId::new("emr-agent").unwrap()).unwrap();
|
||||
assert!(emr.required_backend().is_local(), "EMR must use LOCAL backend");
|
||||
assert!(emr.supported_data_classes().contains(&DataClass::PHI));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_financial_query_routes_finance_local() {
|
||||
let registry = build_registry();
|
||||
let _text = "Revenue last month";
|
||||
|
||||
// Finance agent embedding peaks at dims 10-12
|
||||
let mut embedding = vec![0.0; 1024];
|
||||
embedding[10] = 0.8;
|
||||
embedding[11] = 0.7;
|
||||
let query_embedding = Vector::from(embedding);
|
||||
|
||||
let routed = registry.route(&query_embedding, 0.01);
|
||||
assert!(
|
||||
routed.iter().any(|id| id.0 == "finance-agent"),
|
||||
"Finance agent should be routed for revenue query"
|
||||
);
|
||||
|
||||
let finance = registry.get(&AgentId::new("finance-agent").unwrap()).unwrap();
|
||||
assert!(finance.required_backend().is_local(), "Finance must use LOCAL backend");
|
||||
assert!(finance.supported_data_classes().contains(&DataClass::Financial));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rhythm_predicts_emr_at_8am() {
|
||||
let mut model = RhythmModel::new();
|
||||
let emr = AgentId::new("emr-agent").unwrap();
|
||||
|
||||
// Seed 20 interactions at 8am on Monday
|
||||
for _ in 0..20 {
|
||||
let ts = Utc::now()
|
||||
.with_hour(8).unwrap()
|
||||
.with_minute(0).unwrap()
|
||||
.with_second(0).unwrap();
|
||||
model.record_interaction(InteractionRecord {
|
||||
id: Uuid::new_v4(),
|
||||
timestamp: ts,
|
||||
agent_id: emr.clone(),
|
||||
intent_embedding: Vector::zeros(),
|
||||
response_time_ms: 100,
|
||||
user_accepted: true,
|
||||
});
|
||||
}
|
||||
|
||||
let test_time = Utc::now().with_hour(8).unwrap().with_minute(0).unwrap();
|
||||
let predictions = model.predict_active_agents(test_time);
|
||||
|
||||
assert!(!predictions.is_empty(), "Should have predictions after seeding");
|
||||
let top = &predictions[0];
|
||||
assert_eq!(top.0, emr, "EMR agent should be top prediction at 8am");
|
||||
assert!(top.1 > 0.3, "EMR confidence should be > 30% at 8am");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_multi_agent_routing() {
|
||||
let registry = build_registry();
|
||||
|
||||
// Complex intent that might trigger both EMR and Finance
|
||||
// We'll use a blended embedding
|
||||
let mut embedding = vec![0.0; 1024];
|
||||
embedding[0] = 0.5; // EMR dim
|
||||
embedding[10] = 0.5; // Finance dim
|
||||
let query_embedding = Vector::from(embedding);
|
||||
|
||||
let routed = registry.route(&query_embedding, 0.01);
|
||||
|
||||
// With a blended vector, both agents might be above tau
|
||||
assert!(
|
||||
routed.len() >= 1,
|
||||
"At least one agent should be routed for complex intent"
|
||||
);
|
||||
|
||||
// Verify all routed agents are valid
|
||||
for id in &routed {
|
||||
assert!(registry.get(id).is_some(), "Routed agent {} must exist in registry", id);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_news_agent_uses_cloud() {
|
||||
let registry = build_registry();
|
||||
let news = registry.get(&AgentId::new("news-agent").unwrap()).unwrap();
|
||||
assert!(!news.required_backend().is_local(), "News agent should use CLOUD backend");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_messaging_agent_local() {
|
||||
let registry = build_registry();
|
||||
let msg = registry.get(&AgentId::new("messaging-agent").unwrap()).unwrap();
|
||||
assert!(msg.required_backend().is_local(), "Messaging must use LOCAL backend");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rhythm_attention_budget_phi_priority() {
|
||||
let model = RhythmModel::new();
|
||||
let phi_intent = Intent::new("patient query").with_data_class(DataClass::PHI);
|
||||
let finance_intent = Intent::new("revenue query").with_data_class(DataClass::Financial);
|
||||
let general_intent = Intent::new("hello").with_data_class(DataClass::General);
|
||||
|
||||
let phi_budget = model.compute_attention_budget(&phi_intent, &HashMap::new());
|
||||
let fin_budget = model.compute_attention_budget(&finance_intent, &HashMap::new());
|
||||
let gen_budget = model.compute_attention_budget(&general_intent, &HashMap::new());
|
||||
|
||||
assert!(
|
||||
phi_budget.sliding_window > fin_budget.sliding_window,
|
||||
"PHI should get more attention budget than Financial"
|
||||
);
|
||||
assert!(
|
||||
fin_budget.sliding_window > gen_budget.sliding_window,
|
||||
"Financial should get more attention budget than General"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_agent_registry_with_defaults() {
|
||||
let registry = build_registry();
|
||||
assert_eq!(registry.len(), 4, "Registry should have 4 default agents");
|
||||
|
||||
let ids: Vec<String> = registry.agents().keys().map(|k| k.0.clone()).collect();
|
||||
assert!(ids.contains(&"emr-agent".to_string()));
|
||||
assert!(ids.contains(&"finance-agent".to_string()));
|
||||
assert!(ids.contains(&"messaging-agent".to_string()));
|
||||
assert!(ids.contains(&"news-agent".to_string()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_emr_extract_patient_name() {
|
||||
assert_eq!(
|
||||
EmrAgent::extract_patient_name("Show patient Johnson"),
|
||||
Some("Johnson".into())
|
||||
);
|
||||
assert_eq!(
|
||||
EmrAgent::extract_patient_name("Show patient Smith"),
|
||||
Some("Smith".into())
|
||||
);
|
||||
assert_eq!(
|
||||
EmrAgent::extract_patient_name("Weather today"),
|
||||
None
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rhythm_greeting() {
|
||||
let model = RhythmModel::new();
|
||||
let greeting = model.predict_intent(Utc::now().with_hour(8).unwrap(), &[]);
|
||||
// Empty model → no prediction
|
||||
assert!(greeting.is_none());
|
||||
|
||||
// Seed and predict
|
||||
let mut model = RhythmModel::new();
|
||||
for _ in 0..10 {
|
||||
model.record_interaction(InteractionRecord {
|
||||
id: Uuid::new_v4(),
|
||||
timestamp: Utc::now().with_hour(8).unwrap(),
|
||||
agent_id: AgentId::new("emr-agent").unwrap(),
|
||||
intent_embedding: Vector::zeros(),
|
||||
response_time_ms: 100,
|
||||
user_accepted: true,
|
||||
});
|
||||
}
|
||||
let greeting = model.predict_intent(Utc::now().with_hour(8).unwrap(), &[]);
|
||||
assert!(greeting.is_some());
|
||||
let text = greeting.unwrap();
|
||||
assert!(text.contains("morning") || text.contains("day"), "Greeting should contain time-of-day: {}", text);
|
||||
}
|
||||
Loading…
Reference in a new issue