ooodc/PicoBot
1
0
Fork 0
Code Pull Requests Activity

feat: 重构技能事件处理逻辑，移除 SkillEventSink，添加 SkillActivateTool 模块以优化技能激活流程

Browse Source 
Co-authored-by: Copilot <copilot@github.com>
This commit is contained in:
ooodc 2026-04-28 15:31:56 +08:00
parent 396504dffb
commit 90e44950cb
10 changed files with 174 additions and 294 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

158
src/agent/agent_loop.rs
View File

@ -2,7 +2,6 @@ use crate::bus::ChatMessage;
use crate::bus::message::ToolMessageState;
use crate::config::LLMProviderConfig;
use crate::domain::messages::{ContentBlock, ToolCall};
use crate::domain::tools::Tool;
use crate::observability::{
Observer, ObserverEvent, ToolExecutionOutcome, ToolExecutionState, truncate_args,
};
@ -297,7 +296,6 @@ pub struct AgentLoop {
provider: Box<dyn LLMProvider>,
tools: Arc<ToolRegistry>,
skills: Arc<dyn SkillProvider>,
skill_event_sink: Option<Arc<dyn SkillEventSink>>,
tool_context: ToolContext,
observer: Option<Arc<dyn Observer>>,
emitted_message_handler: Option<Arc<dyn EmittedMessageHandler>>,
@ -315,25 +313,8 @@ pub trait EmittedMessageHandler: Send + Sync + 'static {
async fn handle(&self, message: ChatMessage);
}
#[derive(Debug, Clone)]
pub struct SkillEvent {
pub event_type: String,
pub skill_name: Option<String>,
pub payload: serde_json::Value,
}
pub trait SkillEventSink: Send + Sync + 'static {
fn record_skill_event(&self, event: SkillEvent);
}
pub trait SkillProvider: Send + Sync + 'static {
fn system_index_prompt(&self) -> Option<String>;
fn skill_tool_definition(&self) -> Option<Tool>;
fn activation_payload(&self, name: &str) -> Result<String, String>;
fn activation_event_payload(&self, name: &str) -> Result<serde_json::Value, String>;
}
#[derive(Default)]
@ -343,18 +324,6 @@ impl SkillProvider for EmptySkillProvider {
fn system_index_prompt(&self) -> Option<String> {
None
}
fn skill_tool_definition(&self) -> Option<Tool> {
None
}
fn activation_payload(&self, name: &str) -> Result<String, String> {
Err(format!("skill '{}' not found", name))
}
fn activation_event_payload(&self, name: &str) -> Result<serde_json::Value, String> {
Err(format!("skill '{}' not found", name))
}
}
impl AgentLoop {
@ -368,7 +337,6 @@ impl AgentLoop {
provider,
tools: Arc::new(ToolRegistry::new()),
skills: Arc::new(EmptySkillProvider),
skill_event_sink: None,
tool_context: ToolContext::default(),
observer: None,
emitted_message_handler: None,
@ -389,7 +357,6 @@ impl AgentLoop {
provider,
tools,
skills: Arc::new(EmptySkillProvider),
skill_event_sink: None,
tool_context: ToolContext::default(),
observer: None,
emitted_message_handler: None,
@ -411,7 +378,6 @@ impl AgentLoop {
provider,
tools,
skills,
skill_event_sink: None,
tool_context: ToolContext::default(),
observer: None,
emitted_message_handler: None,
@ -419,11 +385,6 @@ impl AgentLoop {
})
}
pub fn with_skill_event_sink(mut self, sink: Arc<dyn SkillEventSink>) -> Self {
self.skill_event_sink = Some(sink);
self
}
pub fn with_tool_context(mut self, context: ToolContext) -> Self {
self.tool_context = context;
self
@ -479,10 +440,7 @@ impl AgentLoop {
messages_for_llm.extend(messages.iter().map(chat_message_to_llm_message));
// Build request
let mut tool_defs = self.tools.get_definitions();
if let Some(skill_tool) = self.skills.skill_tool_definition() {
tool_defs.push(skill_tool);
}
let tool_defs = self.tools.get_definitions();
let tools = if tool_defs.is_empty() {
None
} else {
@ -818,46 +776,6 @@ impl AgentLoop {
async fn execute_tool_internal(&self, tool_call: &ToolCall) -> ToolExecutionOutcome {
let normalized_arguments = normalize_tool_arguments(&tool_call.arguments);
if tool_call.name == "skill_activate" {
let skill_name = match normalized_arguments.get("name").and_then(|v| v.as_str()) {
Some(name) if !name.trim().is_empty() => name,
_ => {
self.record_skill_event(
"activation_failed",
None,
serde_json::json!({
"reason": "missing_name",
"arguments": normalized_arguments,
}),
);
return ToolExecutionOutcome::failure(
"Error: Missing required parameter: name".to_string(),
Some("Missing required parameter: name".to_string()),
);
}
};
return match self.skills.activation_payload(skill_name) {
Ok(output) => {
if let Ok(payload) = self.skills.activation_event_payload(skill_name) {
self.record_skill_event("activated", Some(skill_name), payload);
}
ToolExecutionOutcome::success(output)
}
Err(err) => {
self.record_skill_event(
"activation_failed",
Some(skill_name),
serde_json::json!({
"reason": err,
"arguments": normalized_arguments,
}),
);
ToolExecutionOutcome::failure(format!("Error: {}", err), Some(err))
}
};
}
let tool = match self.tools.get(&tool_call.name) {
Some(t) => t,
None => {
@ -906,23 +824,6 @@ impl AgentLoop {
}
}
}
fn record_skill_event(
&self,
event_type: &str,
skill_name: Option<&str>,
payload: serde_json::Value,
) {
let Some(sink) = self.skill_event_sink.as_ref() else {
return;
};
sink.record_skill_event(SkillEvent {
event_type: event_type.to_string(),
skill_name: skill_name.map(str::to_string),
payload,
});
}
}
#[cfg(test)]
@ -931,25 +832,6 @@ mod tests {
use crate::observability::{MultiObserver, Observer};
use tempfile::tempdir;
fn test_provider_config() -> LLMProviderConfig {
LLMProviderConfig {
provider_type: "openai".to_string(),
name: "test".to_string(),
base_url: "http://localhost".to_string(),
api_key: "test-key".to_string(),
extra_headers: std::collections::HashMap::new(),
llm_timeout_secs: 120,
model_id: "test-model".to_string(),
temperature: Some(0.0),
max_tokens: Some(32),
context_window_tokens: None,
model_extra: std::collections::HashMap::new(),
max_tool_iterations: 1,
tool_result_max_chars: 20_000,
context_tool_result_trim_chars: 20_000,
}
}
struct TestObserver {
events: std::sync::Mutex<Vec<ObserverEvent>>,
}
@ -962,24 +844,6 @@ mod tests {
}
}
struct TestSkillEventSink {
events: std::sync::Mutex<Vec<SkillEvent>>,
}
impl TestSkillEventSink {
fn new() -> Self {
Self {
events: std::sync::Mutex::new(Vec::new()),
}
}
}
impl SkillEventSink for TestSkillEventSink {
fn record_skill_event(&self, event: SkillEvent) {
self.events.lock().unwrap().push(event);
}
}
impl Observer for TestObserver {
fn record_event(&self, event: &ObserverEvent) {
self.events.lock().unwrap().push(event.clone());
@ -1006,26 +870,6 @@ mod tests {
assert_eq!(multi.len(), 1);
}
#[test]
fn test_skill_events_are_emitted_to_sink() {
let sink = Arc::new(TestSkillEventSink::new());
let agent = AgentLoop::new(test_provider_config())
.unwrap()
.with_skill_event_sink(sink.clone());
agent.record_skill_event(
"activated",
Some("demo"),
serde_json::json!({ "source": "test" }),
);
let events = sink.events.lock().unwrap();
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_type, "activated");
assert_eq!(events[0].skill_name.as_deref(), Some("demo"));
assert_eq!(events[0].payload, serde_json::json!({ "source": "test" }));
}
#[test]
fn test_should_execute_in_parallel_single_tool() {
// Would need a proper setup with AgentLoop to test fully

3
src/agent/mod.rs
View File

@ -2,7 +2,6 @@ pub mod agent_loop;
pub mod context_compressor;
pub use agent_loop::{
AgentError, AgentLoop, AgentProcessResult, EmittedMessageHandler, SkillEvent, SkillEventSink,
SkillProvider,
AgentError, AgentLoop, AgentProcessResult, EmittedMessageHandler, SkillProvider,
};
pub use context_compressor::ContextCompressor;

40
src/gateway/agent_factory.rs
View File

@ -2,16 +2,13 @@ use std::sync::Arc;
use crate::agent::{AgentError, AgentLoop, SkillProvider};
use crate::config::LLMProviderConfig;
use crate::storage::{SessionStore, persistent_session_id};
use crate::storage::persistent_session_id;
use crate::tools::{ToolContext, ToolRegistry};
use super::skill_event_sink::PersistentSkillEventSink;
#[derive(Clone)]
pub(crate) struct AgentFactory {
tools: Arc<ToolRegistry>,
skills: Arc<dyn SkillProvider>,
store: Arc<SessionStore>,
}
pub(crate) struct AgentBuildRequest<'a> {
@ -23,16 +20,8 @@ pub(crate) struct AgentBuildRequest<'a> {
}
impl AgentFactory {
pub(crate) fn new(
tools: Arc<ToolRegistry>,
skills: Arc<dyn SkillProvider>,
store: Arc<SessionStore>,
) -> Self {
Self {
tools,
skills,
store,
}
pub(crate) fn new(tools: Arc<ToolRegistry>, skills: Arc<dyn SkillProvider>) -> Self {
Self { tools, skills }
}
pub(crate) fn create(&self, request: AgentBuildRequest<'_>) -> Result<AgentLoop, AgentError> {
@ -43,21 +32,14 @@ impl AgentFactory {
self.skills.clone(),
)
.map(|agent| {
let skill_event_sink = Arc::new(PersistentSkillEventSink::new(
self.store.clone(),
session_id.clone(),
));
agent
.with_skill_event_sink(skill_event_sink)
.with_tool_context(ToolContext {
channel_name: Some(request.channel_name.to_string()),
sender_id: request.sender_id.map(str::to_string),
chat_id: Some(request.chat_id.to_string()),
session_id: Some(session_id),
message_id: request.message_id.map(str::to_string),
message_seq: None,
})
agent.with_tool_context(ToolContext {
channel_name: Some(request.channel_name.to_string()),
sender_id: request.sender_id.map(str::to_string),
chat_id: Some(request.chat_id.to_string()),
session_id: Some(session_id),
message_id: request.message_id.map(str::to_string),
message_seq: None,
})
})
}
}

1
src/gateway/mod.rs
View File

@ -20,7 +20,6 @@ pub mod session_history;
pub mod session_lifecycle;
pub mod session_message_service;
pub mod session_pool;
pub mod skill_event_sink;
pub mod tool_registry_factory;
pub mod ws;

4
src/gateway/session.rs
View File

@ -104,7 +104,7 @@ impl Session {
store: Arc<SessionStore>,
agent_prompt_reinject_every: u64,
) -> Result<Self, AgentError> {
let agent_factory = AgentFactory::new(tools, skills.clone(), store.clone());
let agent_factory = AgentFactory::new(tools, skills.clone());
let prompt_injector = PromptInjector::new(store.clone(), agent_prompt_reinject_every);
Self::with_factories(
channel_name,
@ -370,7 +370,7 @@ impl SessionManager {
)
.build(),
);
let agent_factory = AgentFactory::new(tools.clone(), skills.clone(), store.clone());
let agent_factory = AgentFactory::new(tools.clone(), skills.clone());
let prompt_injector = PromptInjector::new(store.clone(), agent_prompt_reinject_every);
let session_factory = SessionFactory::new(
provider_config.clone(),

32
src/gateway/skill_event_sink.rs
View File

@ -1,32 +0,0 @@
use std::sync::Arc;
use crate::agent::{SkillEvent, SkillEventSink};
use crate::storage::SkillEventRepository;
pub(crate) struct PersistentSkillEventSink {
events: Arc<dyn SkillEventRepository>,
session_id: String,
}
impl PersistentSkillEventSink {
pub(crate) fn new(events: Arc<dyn SkillEventRepository>, session_id: String) -> Self {
Self { events, session_id }
}
}
impl SkillEventSink for PersistentSkillEventSink {
fn record_skill_event(&self, event: SkillEvent) {
if let Err(err) = self.events.append_skill_event(
Some(&self.session_id),
&event.event_type,
event.skill_name.as_deref(),
&event.payload,
) {
tracing::warn!(
error = %err,
event_type = %event.event_type,
"Failed to record skill event"
);
}
}
}

8
src/gateway/tool_registry_factory.rs
View File

@ -5,8 +5,8 @@ use crate::skills::SkillRuntime;
use crate::storage::SessionStore;
use crate::tools::{
BashTool, CalculatorTool, FileEditTool, FileReadTool, FileWriteTool, HttpRequestTool,
MemoryManageTool, MemorySearchTool, SchedulerManageTool, SkillListTool, SkillManageTool,
TimeTool, ToolRegistry, WebFetchTool,
MemoryManageTool, MemorySearchTool, SchedulerManageTool, SkillActivateTool, SkillListTool,
SkillManageTool, TimeTool, ToolRegistry, WebFetchTool,
};
pub(crate) struct ToolRegistryFactory {
@ -44,6 +44,10 @@ impl ToolRegistryFactory {
self.store.clone(),
self.known_agents.clone(),
));
registry.register(SkillActivateTool::new(
self.skills.clone(),
self.store.clone(),
));
registry.register(SkillListTool::new(self.skills.clone()));
registry.register(SkillManageTool::new(self.skills.clone()));
registry.register(BashTool::new());

69
src/skills/mod.rs
View File

@ -6,7 +6,6 @@ use std::path::{Path, PathBuf};
use std::sync::RwLock;
use crate::config::SkillsConfig;
use crate::domain::tools::{Tool, ToolFunction};
#[derive(Debug, Clone)]
pub struct Skill {
@ -120,13 +119,6 @@ impl SkillRuntime {
.offered_event_payload()
}
pub fn skill_tool_definition(&self) -> Option<Tool> {
self.catalog
.read()
.expect("skills rwlock poisoned")
.skill_tool_definition()
}
pub fn activation_payload(&self, name: &str) -> Result<String, String> {
self.catalog
.read()
@ -234,18 +226,6 @@ impl crate::agent::SkillProvider for SkillRuntime {
fn system_index_prompt(&self) -> Option<String> {
SkillRuntime::system_index_prompt(self)
}
fn skill_tool_definition(&self) -> Option<Tool> {
SkillRuntime::skill_tool_definition(self)
}
fn activation_payload(&self, name: &str) -> Result<String, String> {
SkillRuntime::activation_payload(self, name)
}
fn activation_event_payload(&self, name: &str) -> Result<serde_json::Value, String> {
SkillRuntime::activation_event_payload(self, name)
}
}
impl SkillSource {
@ -362,30 +342,6 @@ impl SkillCatalog {
self.catalog_event_payload()
}
pub fn skill_tool_definition(&self) -> Option<Tool> {
if self.skills.is_empty() {
return None;
}
Some(Tool {
tool_type: "function".to_string(),
function: ToolFunction {
name: "skill_activate".to_string(),
description: "Load detailed instructions for a named skill discovered from SKILL.md files. Use when a task matches a listed skill description.".to_string(),
parameters: json!({
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Skill name from the available skills list"
}
},
"required": ["name"]
}),
},
})
}
pub fn activation_payload(&self, name: &str) -> Result<String, String> {
let skill = self
.find_skill(name)
@ -697,31 +653,6 @@ mod tests {
assert!(err.contains("description"));
}
#[test]
fn test_skill_tool_definition_exists_when_skills_present() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join(".picobot").join("skills").join("demo");
fs::create_dir_all(&root).unwrap();
fs::write(
root.join("SKILL.md"),
"---\ndescription: demo skill\n---\nDo demo",
)
.unwrap();
let skills = load_skills_from_root(
&dir.path().join(".picobot").join("skills"),
SkillSource::Project,
);
let catalog = SkillCatalog {
skills,
max_index_chars: 4000,
max_listed_skills: 10,
};
let tool = catalog.skill_tool_definition().unwrap();
assert_eq!(tool.function.name, "skill_activate");
}
#[test]
fn test_activation_payload_contains_body() {
let dir = tempfile::tempdir().unwrap();

2
src/tools/mod.rs
View File

@ -9,6 +9,7 @@ pub mod memory_search;
pub mod registry;
pub mod scheduler_manage;
pub mod schema;
pub mod skill_activate;
pub mod skill_manage;
pub mod time;
pub mod traits;
@ -25,6 +26,7 @@ pub use memory_search::MemorySearchTool;
pub use registry::ToolRegistry;
pub use scheduler_manage::SchedulerManageTool;
pub use schema::{CleaningStrategy, SchemaCleanr};
pub use skill_activate::SkillActivateTool;
pub use skill_manage::{SkillListTool, SkillManageTool};
pub use time::TimeTool;
pub use traits::{Tool, ToolContext, ToolResult};

151
src/tools/skill_activate.rs Normal file
View File

@ -0,0 +1,151 @@
use std::sync::Arc;
use async_trait::async_trait;
use serde_json::json;
use crate::skills::SkillRuntime;
use crate::storage::SkillEventRepository;
use crate::tools::traits::{Tool, ToolContext, ToolResult};
pub struct SkillActivateTool {
skills: Arc<SkillRuntime>,
events: Arc<dyn SkillEventRepository>,
}
impl SkillActivateTool {
pub fn new(skills: Arc<SkillRuntime>, events: Arc<dyn SkillEventRepository>) -> Self {
Self { skills, events }
}
fn record_event(
&self,
context: &ToolContext,
event_type: &str,
skill_name: Option<&str>,
payload: &serde_json::Value,
) {
if let Err(err) = self.events.append_skill_event(
context.session_id.as_deref(),
event_type,
skill_name,
payload,
) {
tracing::warn!(error = %err, event_type, skill_name, "Failed to record skill activation event");
}
}
}
#[async_trait]
impl Tool for SkillActivateTool {
fn name(&self) -> &str {
"skill_activate"
}
fn description(&self) -> &str {
"Load detailed instructions for a named skill discovered from SKILL.md files. Use when a task matches a listed skill description."
}
fn parameters_schema(&self) -> serde_json::Value {
json!({
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Skill name from the available skills list"
}
},
"required": ["name"]
})
}
async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
self.execute_with_context(&ToolContext::default(), args)
.await
}
async fn execute_with_context(
&self,
context: &ToolContext,
args: serde_json::Value,
) -> anyhow::Result<ToolResult> {
let skill_name = match args.get("name").and_then(|value| value.as_str()) {
Some(name) if !name.trim().is_empty() => name,
_ => {
self.record_event(
context,
"activation_failed",
None,
&json!({
"reason": "missing_name",
"arguments": args,
}),
);
return Ok(error_result("Missing required parameter: name"));
}
};
match self.skills.activation_payload(skill_name) {
Ok(output) => {
if let Ok(payload) = self.skills.activation_event_payload(skill_name) {
self.record_event(context, "activated", Some(skill_name), &payload);
}
Ok(ToolResult {
success: true,
output,
error: None,
})
}
Err(err) => {
self.record_event(
context,
"activation_failed",
Some(skill_name),
&json!({
"reason": err,
"arguments": args,
}),
);
Ok(error_result(&err))
}
}
}
}
fn error_result(message: &str) -> ToolResult {
ToolResult {
success: false,
output: String::new(),
error: Some(message.to_string()),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::storage::SessionStore;
#[tokio::test]
async fn test_skill_activate_records_failed_activation_event() {
let skills = Arc::new(SkillRuntime::default());
let store = Arc::new(SessionStore::in_memory().unwrap());
store.ensure_channel_session("feishu", "chat-1").unwrap();
let tool = SkillActivateTool::new(skills, store.clone());
let context = ToolContext {
session_id: Some("feishu:chat-1".to_string()),
..ToolContext::default()
};
let result = tool
.execute_with_context(&context, json!({ "name": "demo" }))
.await
.unwrap();
assert!(!result.success);
assert!(result.error.unwrap().contains("not found"));
let events = store.list_skill_events(Some("feishu:chat-1")).unwrap();
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_type, "activation_failed");
assert_eq!(events[0].skill_name.as_deref(), Some("demo"));
}
}