dc693fa80b
- 主代理解析链新增 topic 级选择（最高优先级），session 级首命中时物化固化为话题选择 - 子代理解析链共享 helper：def frontmatter > 话题级 > session 级 > 全局基础配置 - TaskToolResult 携带子代理实际使用的 provider/model，供前端差异展示
354 lines
16 KiB
Rust
354 lines
16 KiB
Rust
use std::sync::Arc;
|
||
|
||
use tokio::sync::mpsc;
|
||
|
||
use crate::agent::context_compressor::ContextCompressor;
|
||
use crate::agent::{AgentError, AgentLoop, AgentRuntimeConfig, CompositeSystemPromptProvider, SystemPromptProvider};
|
||
use crate::config::{CompactionConfig, LLMProviderConfig, ModelResolver};
|
||
use crate::domain::CapabilityPolicy;
|
||
use crate::experts::ExpertPromptProvider;
|
||
use crate::experts::ExpertRuntime;
|
||
use crate::gateway::agent_prompt_provider::AgentPromptProvider;
|
||
use crate::gateway::model_selection::ModelSelectionStore;
|
||
use crate::gateway::tool_prompt_provider::ToolPromptProvider;
|
||
use crate::observability::Observer;
|
||
use crate::skills::{SkillPromptProvider, SkillRuntime};
|
||
use crate::storage::PromptInjectionRepository;
|
||
use crate::storage::persistent_session_id;
|
||
use crate::storage::SessionStore;
|
||
use crate::tools::task::runtime::{SubagentPromptProvider, SubagentRuntime};
|
||
use crate::tools::task::SubagentResult;
|
||
use crate::tools::{ToolContext, ToolRegistry, WaitCoordinator};
|
||
|
||
/// 构建与 Agent 实际使用的完全一致的组合系统提示词 Provider。
|
||
///
|
||
/// 单一来源：AgentFactory::create 与命令侧（/save、/save-session、/current）
|
||
/// 都调用此函数，确保保存到文件的系统提示词与 LLM 实际接收的提示词一致。
|
||
///
|
||
/// Provider 顺序：AgentPrompt → SkillPrompt → ExpertPrompt → SubagentPrompt → TodoPrompt
|
||
pub(crate) fn build_system_prompt_provider(
|
||
reinject_every: usize,
|
||
provider_config: LLMProviderConfig,
|
||
prompt_repository: Arc<dyn PromptInjectionRepository>,
|
||
skills: Arc<SkillRuntime>,
|
||
experts: Arc<ExpertRuntime>,
|
||
subagent_runtime: Arc<SubagentRuntime>,
|
||
) -> Arc<dyn SystemPromptProvider> {
|
||
Arc::new(CompositeSystemPromptProvider::new(vec![
|
||
Box::new(AgentPromptProvider::new(
|
||
reinject_every,
|
||
provider_config,
|
||
prompt_repository,
|
||
)),
|
||
Box::new(SkillPromptProvider::new(skills, experts.clone())),
|
||
Box::new(ExpertPromptProvider::new(experts.clone())),
|
||
Box::new(SubagentPromptProvider::new(subagent_runtime, experts)),
|
||
Box::new(ToolPromptProvider::new()),
|
||
]))
|
||
}
|
||
|
||
#[derive(Clone)]
|
||
pub(crate) struct AgentFactory {
|
||
tools: Arc<ToolRegistry>,
|
||
skills: Arc<SkillRuntime>,
|
||
experts: Arc<ExpertRuntime>,
|
||
subagent_runtime: Arc<SubagentRuntime>,
|
||
reinject_every: usize,
|
||
prompt_repository: Arc<dyn PromptInjectionRepository>,
|
||
/// Provider/Model 解析器：按专家 frontmatter 中的 provider/model 字段覆盖基础配置
|
||
model_resolver: Arc<ModelResolver>,
|
||
/// per-session 的用户模型选择（覆盖专家配置）
|
||
model_selections: Arc<ModelSelectionStore>,
|
||
/// per-topic 的用户模型选择（最高优先级；物化后话题模型不再随 session 选择漂移）
|
||
topic_model_selections: Arc<ModelSelectionStore>,
|
||
/// 持久化存储：session 级选择首次被话题命中时物化回写 topics 行
|
||
store: Arc<SessionStore>,
|
||
/// 上下文压缩算法配置（所有 agent 共享）
|
||
compaction_config: CompactionConfig,
|
||
/// 可观测性 Observer（依赖注入到 AgentLoop，业务层不感知具体实现）
|
||
observer: Option<Arc<dyn Observer>>,
|
||
/// 实例创建时间戳（用于区分新旧 AgentFactory 实例）
|
||
instance_id: u64,
|
||
}
|
||
|
||
pub(crate) struct AgentBuildRequest<'a> {
|
||
pub(crate) channel_name: &'a str,
|
||
pub(crate) session_chat_id: &'a str,
|
||
pub(crate) notification_chat_id: Option<&'a str>,
|
||
pub(crate) sender_id: Option<&'a str>,
|
||
pub(crate) message_id: Option<&'a str>,
|
||
pub(crate) provider_config: LLMProviderConfig,
|
||
/// 当前话题 ID（可选）：用于 todo 等按 topic 隔离的工具
|
||
pub(crate) topic_id: Option<String>,
|
||
/// 取消信号接收端（可选）：Agent 在每次迭代时检查是否被取消
|
||
pub(crate) cancel_token: Option<tokio::sync::watch::Receiver<()>>,
|
||
/// 端到端追踪 ID（从 InboundMessage 继承，注入 ToolContext 供 tool 执行路径日志关联）
|
||
pub(crate) trace_id: Option<String>,
|
||
/// 异步子代理完成队列的 sender（按 topic 隔离）。
|
||
/// 仅主 agent 有值：TaskTool 据此在子代理完成时发送 SubagentResult。
|
||
pub(crate) sub_done_sender: Option<mpsc::Sender<SubagentResult>>,
|
||
/// wait_for_subagents 工具的协调器（仅主 agent 有值）。
|
||
pub(crate) wait_coordinator: Option<Arc<dyn WaitCoordinator>>,
|
||
}
|
||
|
||
impl AgentFactory {
|
||
pub(crate) fn new(
|
||
tools: Arc<ToolRegistry>,
|
||
skills: Arc<SkillRuntime>,
|
||
experts: Arc<ExpertRuntime>,
|
||
subagent_runtime: Arc<SubagentRuntime>,
|
||
reinject_every: usize,
|
||
prompt_repository: Arc<dyn PromptInjectionRepository>,
|
||
model_resolver: Arc<ModelResolver>,
|
||
model_selections: Arc<ModelSelectionStore>,
|
||
topic_model_selections: Arc<ModelSelectionStore>,
|
||
store: Arc<SessionStore>,
|
||
compaction_config: CompactionConfig,
|
||
observer: Option<Arc<dyn Observer>>,
|
||
) -> Self {
|
||
// 使用 Arc 指针地址作为实例标识符，用于区分新旧 AgentFactory 实例
|
||
let instance_id = Arc::as_ptr(&tools) as u64;
|
||
tracing::info!(
|
||
instance_id = instance_id,
|
||
tool_count = tools.tool_names().len(),
|
||
"AgentFactory::new created"
|
||
);
|
||
Self {
|
||
tools,
|
||
skills,
|
||
experts,
|
||
subagent_runtime,
|
||
reinject_every,
|
||
prompt_repository,
|
||
model_resolver,
|
||
model_selections,
|
||
topic_model_selections,
|
||
store,
|
||
compaction_config,
|
||
observer,
|
||
instance_id,
|
||
}
|
||
}
|
||
|
||
/// 构造 ContextCompressor（参数内聚到 ContextCompressor，CompactionConfig 注入）。
|
||
/// AgentLoop（in-loop 压缩）和 Session（sync 兜底压缩）共用此方法，
|
||
/// 确保两条压缩路径使用同一套用户配置的压缩参数。
|
||
pub(crate) fn build_compressor(&self, runtime_config: &AgentRuntimeConfig) -> ContextCompressor {
|
||
ContextCompressor::with_compaction_config(
|
||
runtime_config.context_window_tokens,
|
||
runtime_config.context_summary_char_budget,
|
||
&self.compaction_config,
|
||
)
|
||
}
|
||
|
||
pub(crate) fn create(&self, request: AgentBuildRequest<'_>) -> Result<AgentLoop, AgentError> {
|
||
let session_id = persistent_session_id(request.channel_name, request.session_chat_id);
|
||
|
||
// 读取所选专家（用于工具过滤 + 子代理策略 + 模型覆盖）
|
||
let expert = self.experts.selected_expert_for(&session_id);
|
||
let expert_capability = expert.as_ref().map(|e| e.capability.clone());
|
||
|
||
// 按专家 frontmatter 中的 provider/model 字段解析覆盖基础 provider_config。
|
||
// 引用不存在的 provider/model 名时报错并阻止会话（用户主动选择的角色，配置错误应明确反馈）。
|
||
let expert_provider_config = match &expert {
|
||
Some(e) if e.provider.is_some() || e.model.is_some() => {
|
||
let resolved = self
|
||
.model_resolver
|
||
.resolve(
|
||
e.provider.as_deref(),
|
||
e.model.as_deref(),
|
||
&request.provider_config,
|
||
)
|
||
.map_err(|e| AgentError::Other(e.to_string()))?;
|
||
tracing::info!(
|
||
instance_id = self.instance_id,
|
||
session_id = %session_id,
|
||
expert = %e.name,
|
||
provider = %resolved.name,
|
||
model_id = %resolved.model_id,
|
||
"AgentFactory: applied expert model override"
|
||
);
|
||
resolved
|
||
}
|
||
_ => request.provider_config.clone(),
|
||
};
|
||
|
||
// 用户手动选择的 provider/model 覆盖（最高优先级，覆盖专家配置）。
|
||
// 优先级：topic 级选择 > session 级选择；均未设置时保持专家/基础配置。
|
||
// 物化规则：session 级选择首次被话题命中时回写 topics 行固化——此后该话题的
|
||
// 模型只能被"在该话题内显式改选"改变，不再随 session 级选择漂移；
|
||
// 专家/config 默认不物化（保持继承活性）。
|
||
// 引用不存在的 provider/model 名时报错并阻止会话（用户主动选择，配置错误应明确反馈）。
|
||
let topic_selection = request
|
||
.topic_id
|
||
.as_deref()
|
||
.and_then(|tid| self.topic_model_selections.get(tid));
|
||
let from_topic = topic_selection.is_some();
|
||
let user_selection = topic_selection.or_else(|| self.model_selections.get(&session_id));
|
||
|
||
let effective_provider_config = match user_selection {
|
||
Some((user_provider, user_model))
|
||
if user_provider.is_some() || user_model.is_some() =>
|
||
{
|
||
let resolved = self
|
||
.model_resolver
|
||
.resolve(
|
||
user_provider.as_deref(),
|
||
user_model.as_deref(),
|
||
&expert_provider_config,
|
||
)
|
||
.map_err(|e| AgentError::Other(e.to_string()))?;
|
||
|
||
// 物化：命中 session 级选择且话题无固化值时，将解析后的具体
|
||
// (provider, model) 写入 topics 行（持久化 + 内存缓存）
|
||
if !from_topic {
|
||
if let Some(tid) = request.topic_id.as_deref() {
|
||
let provider = resolved.name.clone();
|
||
let model = resolved.model_id.clone();
|
||
self.topic_model_selections
|
||
.set(tid, Some(provider.clone()), Some(model.clone()));
|
||
if let Err(err) =
|
||
self.store.update_topic_model(tid, Some(&provider), Some(&model))
|
||
{
|
||
tracing::warn!(
|
||
error = %err,
|
||
topic_id = %tid,
|
||
"AgentFactory: failed to materialize topic model selection"
|
||
);
|
||
}
|
||
}
|
||
}
|
||
|
||
tracing::info!(
|
||
instance_id = self.instance_id,
|
||
session_id = %session_id,
|
||
topic_id = request.topic_id.as_deref().unwrap_or(""),
|
||
source = if from_topic { "topic" } else { "session" },
|
||
provider = %resolved.name,
|
||
model_id = %resolved.model_id,
|
||
"AgentFactory: applied user model override"
|
||
);
|
||
resolved
|
||
}
|
||
_ => expert_provider_config,
|
||
};
|
||
|
||
// 诊断日志：记录 agent 实际使用的配置和实例 ID
|
||
tracing::info!(
|
||
instance_id = self.instance_id,
|
||
channel = %request.channel_name,
|
||
session_id = %session_id,
|
||
provider = %effective_provider_config.name,
|
||
model_id = %effective_provider_config.model_id,
|
||
tool_count = self.tools.tool_names().len(),
|
||
"AgentFactory: creating agent with config"
|
||
);
|
||
|
||
// 创建组合的系统提示词提供者（与命令侧 /save 等共享同一构建逻辑）
|
||
let system_prompt_provider = build_system_prompt_provider(
|
||
self.reinject_every,
|
||
effective_provider_config.clone(),
|
||
self.prompt_repository.clone(),
|
||
self.skills.clone(),
|
||
self.experts.clone(),
|
||
self.subagent_runtime.clone(),
|
||
);
|
||
|
||
// 按所选专家的工具策略过滤工具集（含内置 + MCP 工具）。
|
||
// 无专家或专家未声明工具策略时，复用共享的 Arc<ToolRegistry>（零拷贝）。
|
||
let base_tool_count = self.tools.tool_names().len();
|
||
let tools: Arc<ToolRegistry> = match &expert_capability {
|
||
Some(cap) if cap.has_tool_policy() => {
|
||
let filtered = self.build_filtered_registry(cap);
|
||
let filtered_count = filtered.tool_names().len();
|
||
tracing::info!(
|
||
instance_id = self.instance_id,
|
||
session_id = %session_id,
|
||
base_tool_count,
|
||
filtered_tool_count = filtered_count,
|
||
"AgentFactory: applied expert tool policy"
|
||
);
|
||
Arc::new(filtered)
|
||
}
|
||
_ => self.tools.clone(),
|
||
};
|
||
|
||
AgentLoop::with_tools_and_system_prompt_provider(
|
||
effective_provider_config.clone(),
|
||
tools,
|
||
system_prompt_provider,
|
||
Some(self.skills.clone()),
|
||
)
|
||
.map(|agent| {
|
||
// notification_chat_id 优先，否则使用 session_chat_id
|
||
let tool_chat_id = request
|
||
.notification_chat_id
|
||
.unwrap_or(request.session_chat_id);
|
||
// 构建上下文压缩器（参数内聚到 ContextCompressor，CompactionConfig 注入）
|
||
// 注入取消信号 receiver 的 clone 到 ToolContext，
|
||
// 供 wait_for_subagents 工具传递给 coordinator.wait() 的 select!。
|
||
// watch::Receiver::clone() 创建共享同一 sender 的新 receiver，
|
||
// 各 receiver 的 has_changed()/changed() 状态独立，互不影响。
|
||
let cancel_rx_for_context = request.cancel_token.as_ref().map(|rx| rx.clone());
|
||
|
||
let runtime_config = AgentRuntimeConfig::from(effective_provider_config.clone());
|
||
let compressor = Arc::new(self.build_compressor(&runtime_config));
|
||
let mut agent = 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(tool_chat_id.to_string()),
|
||
session_id: Some(session_id),
|
||
topic_id: request.topic_id.clone(),
|
||
message_id: request.message_id.map(str::to_string),
|
||
message_seq: None,
|
||
subagent_description: None,
|
||
nesting_depth: 0,
|
||
task_id: None,
|
||
parent_task_id: None,
|
||
tool_call_id: None,
|
||
// 注入专家 capability，TaskTool 据此强制校验子代理白/黑名单
|
||
parent_capability: expert_capability.clone(),
|
||
trace_id: request.trace_id.clone(),
|
||
// 注入异步子代理完成队列 sender（按 topic 隔离）
|
||
sub_done_sender: request.sub_done_sender.clone(),
|
||
// 注入 wait 协调器（封装释放/重获取 serial_lock 逻辑）
|
||
wait_coordinator: request.wait_coordinator.clone(),
|
||
// 注入取消信号 receiver clone（供 wait 工具的 cancel 检查）
|
||
cancel_rx: cancel_rx_for_context,
|
||
})
|
||
.with_compressor(Some(compressor));
|
||
// 注入观测器（依赖注入，agent_loop 只认 Observer trait）
|
||
if let Some(ref observer) = self.observer {
|
||
agent = agent.with_observer(observer.clone());
|
||
}
|
||
// 如果有取消信号接收端，注入 Agent
|
||
if let Some(token) = request.cancel_token {
|
||
agent = agent.with_cancel_token(token);
|
||
}
|
||
agent
|
||
})
|
||
}
|
||
|
||
/// 按专家 CapabilityPolicy 构建过滤后的 ToolRegistry 副本。
|
||
/// 生效顺序：先白名单取交集，再黑名单扣除（与 subagent filter_tool_registry 语义一致）。
|
||
/// 底层工具为 Arc<dyn ToolTrait>，克隆廉价。
|
||
fn build_filtered_registry(&self, policy: &CapabilityPolicy) -> ToolRegistry {
|
||
// 1. 白名单（取交集）；None 表示不限，复制一份以便后续黑名单过滤
|
||
let after_allow: ToolRegistry = match &policy.allowed_tools {
|
||
Some(allowed) => {
|
||
let refs: Vec<&str> = allowed.iter().map(|s| s.as_str()).collect();
|
||
self.tools.only(&refs)
|
||
}
|
||
None => self.tools.without(&[]),
|
||
};
|
||
// 2. 黑名单（扣除）
|
||
if policy.denied_tools.is_empty() {
|
||
after_allow
|
||
} else {
|
||
let refs: Vec<&str> = policy.denied_tools.iter().map(|s| s.as_str()).collect();
|
||
after_allow.without(&refs)
|
||
}
|
||
}
|
||
}
|