徐涛
46de111965
新增 SessionManager::switch_agent / dispatch / dispatch_all / dispatch_stream 四个核心方法，补齐多 Agent 基础系统原语。 交付物: - switch_agent 运行时替换 Arc<dyn Agent>，保留上下文并更新 SessionMeta - dispatch 单任务派发：create_child → inherit_memory → submit_turn - dispatch_all 并行派发：Semaphore 并发控制 + 部分成功语义 - dispatch_stream 流式派发：unbounded_channel + spawn task 消息重建 + finalize_turn - DispatchConfig / SubTaskResult / SubTaskStreamEvent 公开类型 - 4 个端到端示例（agent_switch_demo / sub_agent_dispatch_demo / bridge_keys_demo / dispatch_stream_demo） 辅助变更: - EngineError 新增 DispatchFailed 变体 - CostTracker 实现 From<Usage> 转换 - save_session_meta / load_session_meta 改 pub(crate) 供 switch.rs 使用 测试: +17 个内联测试（4 switch + 5 dispatch + 4 dispatch_all + 4 dispatch_stream），全量 391 passed / 0 failed， clippy 0 警告，doc 0 warning。
122 lines
4.1 KiB
Rust
122 lines
4.1 KiB
Rust
//! dispatch_stream_demo —— 流式子代理调度示例。
|
||
//!
|
||
//! 演示：
|
||
//! 1. 创建父 session
|
||
//! 2. dispatch_stream 单个子 agent
|
||
//! 3. 消费 SubTaskStreamEvent 序列
|
||
//! 4. 验证事件序列：ChildCreated → Stream(...) × N → Completed
|
||
//! 5. 验证：完成时 turn_index 已递增（finalize 副作用）
|
||
//!
|
||
//! 运行：`cargo run --example dispatch_stream_demo`
|
||
|
||
use std::sync::Arc;
|
||
|
||
use agcore::agent::{Agent, AgentBuilder};
|
||
use agcore::engine::{SessionManager, SubTaskStreamEvent};
|
||
use agcore::llm::hooks::HookExecutor;
|
||
use agcore::llm::mock::MockProvider;
|
||
use agcore::llm::types::Usage;
|
||
use agcore::llm::types::message::{ContentBlock, Message};
|
||
use agcore::llm::types::response_v2::{MessageResponse, StopReason};
|
||
use agcore::memory::store::InMemoryStore;
|
||
use agcore::tools::ToolRegistry;
|
||
use futures_util::StreamExt;
|
||
|
||
struct StreamWorkerAgent;
|
||
|
||
impl Agent for StreamWorkerAgent {
|
||
fn name(&self) -> &str {
|
||
"stream_worker"
|
||
}
|
||
fn system_prompt(&self) -> Option<&str> {
|
||
Some("You are a streaming worker.")
|
||
}
|
||
}
|
||
|
||
fn assistant_text(text: &str) -> MessageResponse {
|
||
MessageResponse {
|
||
id: String::new(),
|
||
model: String::new(),
|
||
message: Message::Assistant {
|
||
content: vec![ContentBlock::Text { text: text.into() }],
|
||
},
|
||
usage: Usage::from_input_output(8, 4),
|
||
stop_reason: StopReason::Stop,
|
||
extra: Default::default(),
|
||
}
|
||
}
|
||
|
||
#[tokio::main]
|
||
async fn main() {
|
||
println!("=== Dispatch Stream Demo ===\n");
|
||
|
||
let store: Arc<dyn agcore::memory::store::MemoryStore> = Arc::new(InMemoryStore::new());
|
||
let provider = Arc::new(MockProvider::new(vec![assistant_text("streamed response")]));
|
||
let bundle = Arc::new(
|
||
AgentBuilder::new()
|
||
.provider(provider)
|
||
.tool_registry(Arc::new(ToolRegistry::new()))
|
||
.hook_executor(Arc::new(HookExecutor::new()))
|
||
.session_memory_backend(store.clone())
|
||
.build()
|
||
.expect("RuntimeBundle"),
|
||
);
|
||
|
||
let worker: Arc<dyn Agent> = Arc::new(StreamWorkerAgent);
|
||
|
||
let sm = Arc::new(SessionManager::new(store));
|
||
let parent_id = sm
|
||
.create(worker.clone(), bundle.clone())
|
||
.await
|
||
.expect("create");
|
||
println!("[1] parent session: {parent_id}");
|
||
|
||
// dispatch_stream
|
||
let mut stream = sm
|
||
.dispatch_stream(&parent_id, worker, "do streaming work", Default::default())
|
||
.await
|
||
.expect("dispatch_stream");
|
||
|
||
println!("[2] consuming SubTaskStreamEvent sequence...\n");
|
||
let mut saw_child_created = false;
|
||
let mut saw_stream_count = 0;
|
||
let mut completed = None;
|
||
|
||
while let Some(event) = stream.next().await {
|
||
match event {
|
||
SubTaskStreamEvent::ChildCreated { child_id } => {
|
||
println!(" → ChildCreated({})", &child_id[..20]);
|
||
saw_child_created = true;
|
||
}
|
||
SubTaskStreamEvent::Stream(_) => {
|
||
saw_stream_count += 1;
|
||
}
|
||
SubTaskStreamEvent::Completed(r) => {
|
||
println!(" → Completed(child_id={}, {} tokens)", &r.child_id[..20], r.usage.total().total_tokens);
|
||
completed = Some(r);
|
||
break;
|
||
}
|
||
SubTaskStreamEvent::Error { child_id, error } => {
|
||
panic!("unexpected error: child_id={child_id}, error={error}");
|
||
}
|
||
}
|
||
}
|
||
|
||
let result = completed.expect("Completed should arrive");
|
||
|
||
// 验证事件序列
|
||
assert!(saw_child_created, "ChildCreated should be received");
|
||
assert!(saw_stream_count > 0, "at least one Stream event");
|
||
println!("\n[3] received {} stream events", saw_stream_count);
|
||
|
||
// 验证 finalize 已发生（turn_index 递增）
|
||
let child_session = sm.get(&result.child_id).await.unwrap();
|
||
let child_guard = child_session.lock().await;
|
||
let child_turn_index = child_guard.turn_index();
|
||
drop(child_guard);
|
||
assert_eq!(child_turn_index, 1, "turn_index should increment after finalize");
|
||
println!("[4] child session turn_index = {child_turn_index} (finalize works)");
|
||
|
||
println!("\n✓ dispatch_stream completed successfully");
|
||
}
|