feat(observability): 端到端可观测性整改，修复 trace_id 断链与指标配对
- 传播 trace_id：BusToolCallEmitter/SubAgentEmitter/processor 全链路设置 - AgentEnd 配对：补发 5 个 cancel/defensive 路径，闭合 AgentStart 指标 - LLM 计时修正：attempt_start 移入 retry 循环，排除退避等待时间 - /metrics auth：非 loopback 部署时纳入 Bearer token 校验 - recorder 复用：OnceLock 缓存 PrometheusHandle，热重启后不再返回 503 - 结构化日志：新增 tracing_ctx + JSON 日志格式支持
This commit is contained in:
oudecheng
parent
7cb170e0c2
commit
29060315a3
1
.gitignore
vendored
1
.gitignore
vendored
@ -40,3 +40,4 @@ node_modules
|
||||
logs
|
||||
dist
|
||||
.trae
|
||||
.opencode/
|
||||
|
||||
131
Cargo.lock
generated
131
Cargo.lock
generated
@ -466,6 +466,15 @@ dependencies = [
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-epoch"
|
||||
version = "0.9.20"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-utils"
|
||||
version = "0.8.21"
|
||||
@ -879,6 +888,25 @@ dependencies = [
|
||||
"yaml-rust2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "h2"
|
||||
version = "0.4.15"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155"
|
||||
dependencies = [
|
||||
"atomic-waker",
|
||||
"bytes",
|
||||
"fnv",
|
||||
"futures-core",
|
||||
"futures-sink",
|
||||
"http",
|
||||
"indexmap",
|
||||
"slab",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hashbrown"
|
||||
version = "0.14.5"
|
||||
@ -937,6 +965,12 @@ version = "0.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
|
||||
|
||||
[[package]]
|
||||
name = "hermit-abi"
|
||||
version = "0.5.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c"
|
||||
|
||||
[[package]]
|
||||
name = "hex"
|
||||
version = "0.4.3"
|
||||
@ -1004,6 +1038,7 @@ dependencies = [
|
||||
"bytes",
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
"h2",
|
||||
"http",
|
||||
"http-body",
|
||||
"httparse",
|
||||
@ -1024,7 +1059,9 @@ dependencies = [
|
||||
"http",
|
||||
"hyper",
|
||||
"hyper-util",
|
||||
"log",
|
||||
"rustls",
|
||||
"rustls-native-certs",
|
||||
"tokio",
|
||||
"tokio-rustls",
|
||||
"tower-service",
|
||||
@ -1441,6 +1478,52 @@ version = "2.8.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79"
|
||||
|
||||
[[package]]
|
||||
name = "metrics"
|
||||
version = "0.23.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3045b4193fbdc5b5681f32f11070da9be3609f189a79f3390706d42587f46bb5"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"portable-atomic",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "metrics-exporter-prometheus"
|
||||
version = "0.15.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b4f0c8427b39666bf970460908b213ec09b3b350f20c0c2eabcbba51704a08e6"
|
||||
dependencies = [
|
||||
"base64",
|
||||
"http-body-util",
|
||||
"hyper",
|
||||
"hyper-rustls",
|
||||
"hyper-util",
|
||||
"indexmap",
|
||||
"ipnet",
|
||||
"metrics",
|
||||
"metrics-util",
|
||||
"quanta",
|
||||
"thiserror 1.0.69",
|
||||
"tokio",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "metrics-util"
|
||||
version = "0.17.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4259040465c955f9f2f1a4a8a16dc46726169bca0f88e8fb2dbeced487c3e828"
|
||||
dependencies = [
|
||||
"crossbeam-epoch",
|
||||
"crossbeam-utils",
|
||||
"hashbrown 0.14.5",
|
||||
"metrics",
|
||||
"num_cpus",
|
||||
"quanta",
|
||||
"sketches-ddsketch",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "meval"
|
||||
version = "0.2.0"
|
||||
@ -1556,6 +1639,16 @@ dependencies = [
|
||||
"autocfg",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num_cpus"
|
||||
version = "1.17.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b"
|
||||
dependencies = [
|
||||
"hermit-abi",
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "once_cell"
|
||||
version = "1.21.4"
|
||||
@ -1654,6 +1747,8 @@ dependencies = [
|
||||
"iana-time-zone",
|
||||
"image",
|
||||
"libc",
|
||||
"metrics",
|
||||
"metrics-exporter-prometheus",
|
||||
"meval",
|
||||
"mime_guess",
|
||||
"parking_lot",
|
||||
@ -1710,6 +1805,12 @@ dependencies = [
|
||||
"miniz_oxide",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "portable-atomic"
|
||||
version = "1.15.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85"
|
||||
|
||||
[[package]]
|
||||
name = "potential_utf"
|
||||
version = "0.1.5"
|
||||
@ -1796,6 +1897,21 @@ version = "0.1.29"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e0c5ccf5294c6ccd63a74f1565028353830a9c2f5eb0c682c355c471726a6e3f"
|
||||
|
||||
[[package]]
|
||||
name = "quanta"
|
||||
version = "0.12.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f3ab5a9d756f0d97bdc89019bd2e4ea098cf9cde50ee7564dde6b81ccc8f06c7"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
"libc",
|
||||
"once_cell",
|
||||
"raw-cpuid",
|
||||
"wasi",
|
||||
"web-sys",
|
||||
"winapi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quick-error"
|
||||
version = "2.0.1"
|
||||
@ -1947,6 +2063,15 @@ version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
|
||||
|
||||
[[package]]
|
||||
name = "raw-cpuid"
|
||||
version = "11.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "498cd0dc59d73224351ee52a95fee0f1a617a2eae0e7d9d720cc622c73a54186"
|
||||
dependencies = [
|
||||
"bitflags",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "redox_syscall"
|
||||
version = "0.5.18"
|
||||
@ -2562,6 +2687,12 @@ version = "1.0.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649"
|
||||
|
||||
[[package]]
|
||||
name = "sketches-ddsketch"
|
||||
version = "0.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "85636c14b73d81f541e525f585c0a2109e6744e1565b5c1668e31c70c10ed65c"
|
||||
|
||||
[[package]]
|
||||
name = "slab"
|
||||
version = "0.4.12"
|
||||
|
||||
@ -41,6 +41,8 @@ prost = "0.14"
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
|
||||
tracing-appender = "0.2"
|
||||
metrics = "0.23"
|
||||
metrics-exporter-prometheus = "0.15"
|
||||
anyhow = "1.0"
|
||||
chrono = { version = "0.4", features = ["serde"] }
|
||||
chrono-tz = "0.10"
|
||||
|
||||
@ -1068,13 +1068,13 @@ impl AgentLoop {
|
||||
/// - `compaction_sink`: 压缩结果回写端（可选）。配置 `compressor` 后，
|
||||
/// 当 LLM 压缩被触发时通过此 sink 把压缩结果持久化到 DB。
|
||||
/// 传 None 则即使配置了 compressor 也只改内存不回写。
|
||||
#[tracing::instrument(skip(self, messages, system_prompt_context, compaction_sink), fields(history_len = messages.len(), max_iterations = self.max_iterations))]
|
||||
pub async fn process(
|
||||
&self,
|
||||
mut messages: Vec<ChatMessage>,
|
||||
system_prompt_context: Option<&SystemPromptContext>,
|
||||
compaction_sink: Option<&dyn CompactionSink>,
|
||||
) -> Result<AgentProcessResult, AgentError> {
|
||||
#[cfg(debug_assertions)]
|
||||
tracing::debug!(
|
||||
history_len = messages.len(),
|
||||
max_iterations = self.max_iterations,
|
||||
@ -1179,6 +1179,14 @@ impl AgentLoop {
|
||||
)
|
||||
.await;
|
||||
|
||||
// Emit AgentStart event for metrics (LLM 请求耗时/token 指标)
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentStart {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
// Set up streaming delta consumer
|
||||
// Pre-generate the message ID so stream deltas and the final assistant
|
||||
// message share the same ID — this lets the front-end replace the
|
||||
@ -1190,9 +1198,12 @@ impl AgentLoop {
|
||||
|
||||
let max_retries = self.runtime_config.max_retries as usize;
|
||||
let mut response: Option<crate::providers::ChatCompletionResponse> = None;
|
||||
// 记录最后一次尝试的耗时（不含重试退避），用于 AgentEnd 指标
|
||||
let mut last_attempt_duration = std::time::Duration::ZERO;
|
||||
|
||||
'retry: for attempt in 0..=max_retries {
|
||||
// 每次重试重建 channel + consumer：上次失败的 channel 可能已关闭。
|
||||
let attempt_start = std::time::Instant::now();
|
||||
let (delta_tx, mut delta_rx) = tokio::sync::mpsc::channel::<StreamDelta>(256);
|
||||
let consumer_handler = self.emitted_message_handler.clone();
|
||||
let consumer_task = tokio::spawn(async move {
|
||||
@ -1222,6 +1233,16 @@ impl AgentLoop {
|
||||
_ = self.cancel_signal() => {
|
||||
drop(stream_callback);
|
||||
let _ = consumer_task.await;
|
||||
// cancel 路径补发 AgentEnd，保证指标配对闭合
|
||||
last_attempt_duration = attempt_start.elapsed();
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentEnd {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
duration: last_attempt_duration,
|
||||
tokens_used: None,
|
||||
});
|
||||
}
|
||||
let cancel = Self::build_cancel_result(iteration, emitted_messages);
|
||||
self.emit_live_tool_call_message(cancel.final_response.clone()).await;
|
||||
return Ok(cancel);
|
||||
@ -1242,10 +1263,12 @@ impl AgentLoop {
|
||||
|
||||
match llm_result {
|
||||
Ok(resp) => {
|
||||
last_attempt_duration = attempt_start.elapsed();
|
||||
response = Some(resp);
|
||||
break 'retry;
|
||||
}
|
||||
Err(e) => {
|
||||
last_attempt_duration = attempt_start.elapsed();
|
||||
let error_text = e.to_string();
|
||||
let can_retry = attempt < max_retries
|
||||
&& !emitted.load(Ordering::SeqCst)
|
||||
@ -1265,6 +1288,16 @@ impl AgentLoop {
|
||||
if self.cancel_token.is_some() {
|
||||
tokio::select! {
|
||||
_ = self.cancel_signal() => {
|
||||
// cancel 路径补发 AgentEnd，保证指标配对闭合
|
||||
last_attempt_duration = attempt_start.elapsed();
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentEnd {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
duration: last_attempt_duration,
|
||||
tokens_used: None,
|
||||
});
|
||||
}
|
||||
let cancel = Self::build_cancel_result(iteration, emitted_messages);
|
||||
self.emit_live_tool_call_message(cancel.final_response.clone()).await;
|
||||
return Ok(cancel);
|
||||
@ -1283,6 +1316,15 @@ impl AgentLoop {
|
||||
error_details = %format_error_chain(e.as_ref()),
|
||||
"LLM request failed"
|
||||
);
|
||||
// 错误分支补发 AgentEnd，保证指标配对闭合
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentEnd {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
duration: last_attempt_duration,
|
||||
tokens_used: None,
|
||||
});
|
||||
}
|
||||
let assistant_message =
|
||||
ChatMessage::assistant(recoverable_llm_message(&error_text));
|
||||
emitted_messages.push(assistant_message.clone());
|
||||
@ -1298,11 +1340,34 @@ impl AgentLoop {
|
||||
}
|
||||
}
|
||||
|
||||
let response = response.ok_or_else(|| {
|
||||
AgentError::Other(
|
||||
"retry loop exited without setting response or returning".to_string(),
|
||||
)
|
||||
})?;
|
||||
// 防御性兜底：retry 循环异常退出（正常不应发生）。
|
||||
// 补发 AgentEnd 保证指标配对闭合，再返回错误。
|
||||
let response = match response {
|
||||
Some(resp) => resp,
|
||||
None => {
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentEnd {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
duration: last_attempt_duration,
|
||||
tokens_used: None,
|
||||
});
|
||||
}
|
||||
return Err(AgentError::Other(
|
||||
"retry loop exited without setting response or returning".to_string(),
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
// Emit AgentEnd event for metrics (LLM 请求耗时/token 指标)
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentEnd {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
duration: last_attempt_duration,
|
||||
tokens_used: Some(response.usage.total_tokens as u64),
|
||||
});
|
||||
}
|
||||
|
||||
// Signal stream end if handler exists
|
||||
let had_streaming = self.emitted_message_handler.is_some();
|
||||
@ -1715,8 +1780,20 @@ impl AgentLoop {
|
||||
.await;
|
||||
let max_retries = self.runtime_config.max_retries as usize;
|
||||
|
||||
// Emit AgentStart for the summary LLM call
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentStart {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
// 记录最后一次尝试的耗时（不含重试退避）
|
||||
let mut summary_last_attempt_duration = std::time::Duration::ZERO;
|
||||
|
||||
for attempt in 0..=max_retries {
|
||||
// 最终 summary 调用也与取消信号竞速
|
||||
let attempt_start = std::time::Instant::now();
|
||||
let final_result: Result<
|
||||
crate::providers::ChatCompletionResponse,
|
||||
Box<dyn std::error::Error + Send + Sync>,
|
||||
@ -1724,6 +1801,16 @@ impl AgentLoop {
|
||||
if self.cancel_token.is_some() {
|
||||
tokio::select! {
|
||||
_ = self.cancel_signal() => {
|
||||
// cancel 路径补发 AgentEnd，保证指标配对闭合
|
||||
summary_last_attempt_duration = attempt_start.elapsed();
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentEnd {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
duration: summary_last_attempt_duration,
|
||||
tokens_used: None,
|
||||
});
|
||||
}
|
||||
let cancel = Self::build_cancel_result(self.max_iterations, std::mem::take(emitted_messages));
|
||||
self.emit_live_tool_call_message(cancel.final_response.clone()).await;
|
||||
return cancel;
|
||||
@ -1738,6 +1825,16 @@ impl AgentLoop {
|
||||
|
||||
match final_result {
|
||||
Ok(response) => {
|
||||
summary_last_attempt_duration = attempt_start.elapsed();
|
||||
// Emit AgentEnd for the summary LLM call
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentEnd {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
duration: summary_last_attempt_duration,
|
||||
tokens_used: Some(response.usage.total_tokens as u64),
|
||||
});
|
||||
}
|
||||
let mut assistant_message = if let Some(reasoning_content) =
|
||||
response.reasoning_content
|
||||
{
|
||||
@ -1761,6 +1858,7 @@ impl AgentLoop {
|
||||
};
|
||||
}
|
||||
Err(e) => {
|
||||
summary_last_attempt_duration = attempt_start.elapsed();
|
||||
let error_text = e.to_string();
|
||||
let can_retry = attempt < max_retries && is_recoverable_llm_error(&error_text);
|
||||
if can_retry {
|
||||
@ -1776,6 +1874,16 @@ impl AgentLoop {
|
||||
if self.cancel_token.is_some() {
|
||||
tokio::select! {
|
||||
_ = self.cancel_signal() => {
|
||||
// cancel 路径补发 AgentEnd，保证指标配对闭合
|
||||
summary_last_attempt_duration = attempt_start.elapsed();
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentEnd {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
duration: summary_last_attempt_duration,
|
||||
tokens_used: None,
|
||||
});
|
||||
}
|
||||
let cancel = Self::build_cancel_result(self.max_iterations, std::mem::take(emitted_messages));
|
||||
self.emit_live_tool_call_message(cancel.final_response.clone()).await;
|
||||
return cancel;
|
||||
@ -1794,6 +1902,15 @@ impl AgentLoop {
|
||||
error_details = %format_error_chain(e.as_ref()),
|
||||
"Failed to get summary from LLM"
|
||||
);
|
||||
// 错误分支补发 AgentEnd，保证指标配对闭合
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentEnd {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
duration: summary_last_attempt_duration,
|
||||
tokens_used: None,
|
||||
});
|
||||
}
|
||||
let final_message =
|
||||
ChatMessage::assistant(recoverable_llm_message(&error_text));
|
||||
emitted_messages.push(final_message.clone());
|
||||
@ -1818,6 +1935,15 @@ impl AgentLoop {
|
||||
model = %self.provider.model_id(),
|
||||
"run_final_summary retry loop exited without returning"
|
||||
);
|
||||
// 兜底也补发 AgentEnd
|
||||
if let Some(ref observer) = self.observer {
|
||||
observer.record_event(&ObserverEvent::AgentEnd {
|
||||
provider: self.provider.name().to_string(),
|
||||
model: self.provider.model_id().to_string(),
|
||||
duration: summary_last_attempt_duration,
|
||||
tokens_used: None,
|
||||
});
|
||||
}
|
||||
let final_message = ChatMessage::assistant(
|
||||
"Failed to generate final summary: retry loop exited unexpectedly.",
|
||||
);
|
||||
@ -1935,6 +2061,7 @@ impl AgentLoop {
|
||||
}
|
||||
|
||||
/// Execute a single tool and return the outcome with event tracking.
|
||||
#[tracing::instrument(skip(self, tool_call), fields(tool = %tool_call.name))]
|
||||
async fn execute_one_tool(&self, tool_call: &ToolCall) -> ToolExecutionOutcome {
|
||||
let start = Instant::now();
|
||||
let tool_name = tool_call.name.clone();
|
||||
|
||||
@ -502,6 +502,9 @@ pub struct InboundMessage {
|
||||
pub metadata: HashMap<String, String>,
|
||||
/// Data forwarded from inbound to outbound (copied to OutboundMessage.metadata by gateway).
|
||||
pub forwarded_metadata: HashMap<String, String>,
|
||||
/// 端到端追踪 ID（由 channel 在构造消息时生成，贯穿 bus→processor→agent_loop→provider→tool 全链路）。
|
||||
/// 基础设施层元数据，不进入 domain 层。
|
||||
pub trace_id: String,
|
||||
}
|
||||
|
||||
impl InboundMessage {
|
||||
@ -537,6 +540,9 @@ pub struct OutboundMessage {
|
||||
/// instead of generating a random UUID. Critical for stream delta → assistant_response
|
||||
/// ID matching on the front-end.
|
||||
pub message_id: Option<String>,
|
||||
/// 端到端追踪 ID（从 InboundMessage 继承，用于 outbound dispatcher 日志关联）。
|
||||
/// 非 agent 执行路径产生的消息（如 scheduler 通知）此字段为空。
|
||||
pub trace_id: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
@ -564,6 +570,12 @@ impl OutboundMessage {
|
||||
)
|
||||
}
|
||||
|
||||
/// 设置 trace_id（builder 模式，用于 agent 执行路径中从 InboundMessage 继承）。
|
||||
pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
|
||||
self.trace_id = trace_id.into();
|
||||
self
|
||||
}
|
||||
|
||||
pub fn assistant(
|
||||
channel: impl Into<String>,
|
||||
chat_id: impl Into<String>,
|
||||
@ -587,6 +599,7 @@ impl OutboundMessage {
|
||||
tool_arguments: None,
|
||||
reasoning_content: None,
|
||||
message_id: None,
|
||||
trace_id: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@ -645,6 +658,7 @@ impl OutboundMessage {
|
||||
tool_arguments: Some(tool_arguments),
|
||||
reasoning_content: None,
|
||||
message_id: None,
|
||||
trace_id: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@ -676,6 +690,7 @@ impl OutboundMessage {
|
||||
tool_arguments: None,
|
||||
reasoning_content: None,
|
||||
message_id: None,
|
||||
trace_id: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@ -707,6 +722,7 @@ impl OutboundMessage {
|
||||
tool_arguments: None,
|
||||
reasoning_content: None,
|
||||
message_id: None,
|
||||
trace_id: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@ -735,6 +751,7 @@ impl OutboundMessage {
|
||||
tool_arguments: None,
|
||||
reasoning_content: reasoning_delta,
|
||||
message_id: None,
|
||||
trace_id: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@ -761,6 +778,7 @@ impl OutboundMessage {
|
||||
tool_arguments: None,
|
||||
reasoning_content: None,
|
||||
message_id: None,
|
||||
trace_id: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@ -786,6 +804,7 @@ impl OutboundMessage {
|
||||
tool_arguments: None,
|
||||
reasoning_content: None,
|
||||
message_id: None,
|
||||
trace_id: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -35,8 +35,15 @@ impl MessageBus {
|
||||
|
||||
/// Publish a message to the inbound queue
|
||||
pub async fn publish_inbound(&self, msg: InboundMessage) -> Result<(), BusError> {
|
||||
#[cfg(debug_assertions)]
|
||||
tracing::debug!(channel = %msg.channel, sender = %msg.sender_id, chat = %msg.chat_id, content_len = %msg.content.len(), media_count = %msg.media.len(), "Bus: publishing inbound message");
|
||||
tracing::debug!(
|
||||
channel = %msg.channel,
|
||||
sender = %msg.sender_id,
|
||||
chat_id = %msg.chat_id,
|
||||
trace_id = %msg.trace_id,
|
||||
content_len = %msg.content.len(),
|
||||
media_count = %msg.media.len(),
|
||||
"Bus: publishing inbound message"
|
||||
);
|
||||
self.inbound_tx
|
||||
.send(msg)
|
||||
.await
|
||||
@ -47,8 +54,13 @@ impl MessageBus {
|
||||
/// Returns `None` when the channel is closed (all senders dropped).
|
||||
pub async fn consume_inbound(&self) -> Option<InboundMessage> {
|
||||
let msg = self.inbound_rx.lock().await.recv().await?;
|
||||
#[cfg(debug_assertions)]
|
||||
tracing::debug!(channel = %msg.channel, sender = %msg.sender_id, chat = %msg.chat_id, "Bus: consuming inbound message");
|
||||
tracing::debug!(
|
||||
channel = %msg.channel,
|
||||
sender = %msg.sender_id,
|
||||
chat_id = %msg.chat_id,
|
||||
trace_id = %msg.trace_id,
|
||||
"Bus: consuming inbound message"
|
||||
);
|
||||
Some(msg)
|
||||
}
|
||||
|
||||
@ -59,13 +71,20 @@ impl MessageBus {
|
||||
/// blocked by slow or disconnected display consumers. Persistent state is
|
||||
/// unaffected — messages are stored in SQLite independently.
|
||||
pub async fn publish_outbound(&self, msg: OutboundMessage) -> Result<(), BusError> {
|
||||
#[cfg(debug_assertions)]
|
||||
tracing::debug!(channel = %msg.channel, chat_id = %msg.chat_id, content_len = %msg.content.len(), "Bus: publishing outbound message");
|
||||
tracing::debug!(
|
||||
channel = %msg.channel,
|
||||
chat_id = %msg.chat_id,
|
||||
trace_id = %msg.trace_id,
|
||||
content_len = %msg.content.len(),
|
||||
"Bus: publishing outbound message"
|
||||
);
|
||||
match self.outbound_tx.try_send(msg) {
|
||||
Ok(()) => Ok(()),
|
||||
Err(tokio::sync::mpsc::error::TrySendError::Full(msg)) => {
|
||||
tracing::warn!(
|
||||
channel = %msg.channel,
|
||||
chat_id = %msg.chat_id,
|
||||
trace_id = %msg.trace_id,
|
||||
"Outbound bus full, dropping message"
|
||||
);
|
||||
Err(BusError::Dropped)
|
||||
|
||||
@ -1327,6 +1327,7 @@ impl FeishuChannel {
|
||||
media: parsed.media.map(|m| vec![m]).unwrap_or_default(),
|
||||
metadata: std::collections::HashMap::new(),
|
||||
forwarded_metadata,
|
||||
trace_id: crate::observability::tracing_ctx::new_trace_id(),
|
||||
};
|
||||
if let Err(e) = channel.handle_and_publish(&bus, &msg).await {
|
||||
tracing::error!(error = %e, open_id = %parsed.open_id, chat_id = %parsed.chat_id, "Failed to publish Feishu message to bus");
|
||||
|
||||
@ -247,6 +247,7 @@ impl Channel for WechatChannel {
|
||||
media,
|
||||
metadata,
|
||||
forwarded_metadata: HashMap::new(),
|
||||
trace_id: crate::observability::tracing_ctx::new_trace_id(),
|
||||
};
|
||||
|
||||
if let Err(error) = bus.publish_inbound(inbound).await {
|
||||
|
||||
@ -79,6 +79,7 @@ impl InitWizard {
|
||||
memory_maintenance: crate::config::MemoryMaintenanceConfig::default(),
|
||||
mcp_servers: HashMap::new(),
|
||||
mcp_tool_timeout_secs: 300,
|
||||
observability: crate::config::ObservabilityConfig::default(),
|
||||
image_context: crate::config::ImageContextConfig::default(),
|
||||
subagents: crate::config::SubagentsConfig::default(),
|
||||
experts: crate::config::ExpertsConfig::default(),
|
||||
@ -845,6 +846,7 @@ impl InitWizard {
|
||||
memory_maintenance: existing.memory_maintenance.clone(),
|
||||
mcp_servers: existing.mcp_servers.clone(),
|
||||
mcp_tool_timeout_secs: existing.mcp_tool_timeout_secs,
|
||||
observability: existing.observability.clone(),
|
||||
image_context: existing.image_context.clone(),
|
||||
subagents: existing.subagents.clone(),
|
||||
experts: existing.experts.clone(),
|
||||
|
||||
@ -45,6 +45,8 @@ pub struct Config {
|
||||
pub experts: ExpertsConfig,
|
||||
#[serde(default)]
|
||||
pub compaction: CompactionConfig,
|
||||
#[serde(default)]
|
||||
pub observability: ObservabilityConfig,
|
||||
}
|
||||
|
||||
/// 图片上下文限制配置
|
||||
@ -125,6 +127,34 @@ impl Default for CompactionConfig {
|
||||
}
|
||||
}
|
||||
|
||||
/// 可观测性配置（日志格式、metrics 开关等）
|
||||
#[derive(Debug, Clone, Deserialize, Serialize)]
|
||||
pub struct ObservabilityConfig {
|
||||
/// 日志输出格式：text（默认）或 json。
|
||||
/// json 格式便于接入 ELK/Loki 等日志聚合系统。
|
||||
#[serde(default)]
|
||||
pub log_format: LogFormat,
|
||||
}
|
||||
|
||||
impl Default for ObservabilityConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
log_format: LogFormat::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 日志输出格式
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, Default, PartialEq, Eq)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum LogFormat {
|
||||
/// 纯文本格式（默认，便于人读）
|
||||
#[default]
|
||||
Text,
|
||||
/// JSON 格式（便于机器解析和日志聚合）
|
||||
Json,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize)]
|
||||
pub struct TimeConfig {
|
||||
#[serde(default = "default_timezone")]
|
||||
|
||||
@ -9,6 +9,7 @@ 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;
|
||||
@ -56,6 +57,8 @@ pub(crate) struct AgentFactory {
|
||||
model_selections: Arc<ModelSelectionStore>,
|
||||
/// 上下文压缩算法配置（所有 agent 共享）
|
||||
compaction_config: CompactionConfig,
|
||||
/// 可观测性 Observer（依赖注入到 AgentLoop，业务层不感知具体实现）
|
||||
observer: Option<Arc<dyn Observer>>,
|
||||
/// 实例创建时间戳（用于区分新旧 AgentFactory 实例）
|
||||
instance_id: u64,
|
||||
}
|
||||
@ -71,6 +74,8 @@ pub(crate) struct AgentBuildRequest<'a> {
|
||||
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>,
|
||||
}
|
||||
|
||||
impl AgentFactory {
|
||||
@ -84,6 +89,7 @@ impl AgentFactory {
|
||||
model_resolver: Arc<ModelResolver>,
|
||||
model_selections: Arc<ModelSelectionStore>,
|
||||
compaction_config: CompactionConfig,
|
||||
observer: Option<Arc<dyn Observer>>,
|
||||
) -> Self {
|
||||
// 使用 Arc 指针地址作为实例标识符，用于区分新旧 AgentFactory 实例
|
||||
let instance_id = Arc::as_ptr(&tools) as u64;
|
||||
@ -102,6 +108,7 @@ impl AgentFactory {
|
||||
model_resolver,
|
||||
model_selections,
|
||||
compaction_config,
|
||||
observer,
|
||||
instance_id,
|
||||
}
|
||||
}
|
||||
@ -245,8 +252,13 @@ impl AgentFactory {
|
||||
tool_call_id: None,
|
||||
// 注入专家 capability，TaskTool 据此强制校验子代理白/黑名单
|
||||
parent_capability: expert_capability.clone(),
|
||||
trace_id: request.trace_id.clone(),
|
||||
})
|
||||
.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);
|
||||
|
||||
@ -84,9 +84,10 @@ pub fn extract_bearer_token(headers: &HeaderMap) -> Option<&str> {
|
||||
})
|
||||
}
|
||||
|
||||
/// axum 中间件：对 `/api/*` 路由强制 Bearer token 校验。
|
||||
/// axum 中间件：对 `/api/*` 和 `/metrics` 路由强制 Bearer token 校验。
|
||||
/// 仅在 `requires_auth` 为 true 时挂载。
|
||||
/// `/health`、`/ws`、静态资源放行；`/ws` 的 token 校验在 ws_handler 内完成。
|
||||
/// `/metrics` 包含运行时指标（provider/model/耗时/token 用量），远程部署时需保护。
|
||||
pub async fn require_bearer_auth(
|
||||
headers: HeaderMap,
|
||||
request: Request,
|
||||
@ -94,8 +95,9 @@ pub async fn require_bearer_auth(
|
||||
) -> Response {
|
||||
let path = request.uri().path();
|
||||
|
||||
// 仅对 /api/ 前缀的请求强制认证
|
||||
if !path.starts_with("/api/") {
|
||||
// /api/* 和 /metrics 需要认证；其余放行
|
||||
let needs_auth = path.starts_with("/api/") || path == "/metrics";
|
||||
if !needs_auth {
|
||||
return next.run(request).await;
|
||||
}
|
||||
|
||||
|
||||
@ -114,6 +114,8 @@ pub(crate) struct MessageExecutionRequest<'a> {
|
||||
pub(crate) live_emitter: Option<Arc<dyn EmittedMessageHandler>>,
|
||||
/// 消息接收时捕获的 topic_id，全程显式传递避免从共享状态重复读取竞态
|
||||
pub(crate) topic_id: Option<String>,
|
||||
/// 端到端追踪 ID（从 InboundMessage 透传，贯穿 agent → tool → outbound）
|
||||
pub(crate) trace_id: &'a str,
|
||||
}
|
||||
|
||||
pub(crate) struct ScheduledExecutionRequest<'a> {
|
||||
@ -127,6 +129,8 @@ pub(crate) struct ScheduledExecutionRequest<'a> {
|
||||
pub(crate) system_prompt: Option<&'a str>,
|
||||
pub(crate) metadata: &'a HashMap<String, String>,
|
||||
pub(crate) fresh_session: bool,
|
||||
/// 端到端追踪 ID（由 ScheduledAgentTaskService 生成）
|
||||
pub(crate) trace_id: String,
|
||||
}
|
||||
|
||||
impl AgentExecutionService {
|
||||
@ -341,6 +345,7 @@ impl AgentExecutionService {
|
||||
Some(request.sender_id),
|
||||
Some(&user_message.id),
|
||||
original_topic_id.as_deref(),
|
||||
request.trace_id,
|
||||
)?;
|
||||
if let Some(handler) = request.live_emitter.clone() {
|
||||
agent = agent.with_emitted_message_handler(handler);
|
||||
@ -477,6 +482,7 @@ impl AgentExecutionService {
|
||||
Some(&user_message.id),
|
||||
request.provider_config.clone(),
|
||||
original_topic_id.as_deref(),
|
||||
&request.trace_id,
|
||||
)?;
|
||||
|
||||
// 获取 store 和 session_id，用于构造消息持久化 handler
|
||||
|
||||
@ -240,6 +240,21 @@ pub async fn list_executions(State(state): State<Arc<GatewayState>>) -> Json<Exe
|
||||
Json(ExecutionsResponse { topic_ids })
|
||||
}
|
||||
|
||||
/// GET /metrics — Prometheus metrics 端点
|
||||
///
|
||||
/// 返回 Prometheus 格式的 metrics 文本。若 recorder 未安装则返回 503。
|
||||
pub async fn metrics_handler(
|
||||
State(state): State<Arc<GatewayState>>,
|
||||
) -> (StatusCode, String) {
|
||||
match &state.prometheus_handle {
|
||||
Some(handle) => (StatusCode::OK, handle.render()),
|
||||
None => (
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"Metrics recorder not initialized".to_string(),
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// GET /api/mcp/status — Return MCP server connection status
|
||||
pub async fn mcp_status(
|
||||
State(state): State<Arc<GatewayState>>,
|
||||
|
||||
@ -72,6 +72,9 @@ pub struct GatewayState {
|
||||
pub subagent_runtime: Arc<SubagentRuntime>,
|
||||
/// per-session 的用户模型选择（覆盖专家配置）
|
||||
pub model_selections: Arc<model_selection::ModelSelectionStore>,
|
||||
/// Prometheus metrics handle（/metrics 端点渲染用）。
|
||||
/// None 表示 recorder 安装失败；热重启时从 OnceLock 缓存复用，不会因重复安装而变为 None。
|
||||
pub prometheus_handle: Option<metrics_exporter_prometheus::PrometheusHandle>,
|
||||
}
|
||||
|
||||
impl GatewayState {
|
||||
@ -132,6 +135,9 @@ impl GatewayState {
|
||||
|
||||
let cancel_manager = CancelManager::new();
|
||||
|
||||
// 安装 Prometheus recorder（幂等：首次安装并缓存 handle，热重启时返回缓存）
|
||||
let prometheus_handle = crate::observability::metrics::init_recorder();
|
||||
|
||||
Ok(Self {
|
||||
config: Arc::new(RwLock::new(config)),
|
||||
session_manager,
|
||||
@ -145,6 +151,7 @@ impl GatewayState {
|
||||
experts,
|
||||
subagent_runtime,
|
||||
model_selections,
|
||||
prometheus_handle,
|
||||
})
|
||||
}
|
||||
|
||||
@ -195,9 +202,10 @@ pub async fn run(
|
||||
) -> Result<bool, Box<dyn std::error::Error>> {
|
||||
let config = Config::load_default()?;
|
||||
let timezone = config.time.parse_timezone()?;
|
||||
let log_format = config.observability.log_format.clone();
|
||||
|
||||
// Initialize logging
|
||||
logging::init_logging(timezone);
|
||||
logging::init_logging(timezone, log_format);
|
||||
tracing::info!("Starting PicoBot Gateway");
|
||||
|
||||
// Restart signal channel
|
||||
@ -332,7 +340,8 @@ pub async fn run(
|
||||
"/api/session/selected-model",
|
||||
routing::get(http::session_selected_model),
|
||||
)
|
||||
.route("/ws", routing::get(ws::ws_handler));
|
||||
.route("/ws", routing::get(ws::ws_handler))
|
||||
.route("/metrics", routing::get(http::metrics_handler));
|
||||
|
||||
// 仅 fallback 按模式区分：嵌入资源 vs 磁盘目录。
|
||||
// fallback 必须在 with_state 之前调用，否则 handler 的 State 类型无法推断。
|
||||
|
||||
@ -171,6 +171,8 @@ impl OutboundDispatcher {
|
||||
channel_name: &str,
|
||||
msg: OutboundMessage,
|
||||
) {
|
||||
let msg_chat_id = msg.chat_id.clone();
|
||||
let msg_trace_id = msg.trace_id.clone();
|
||||
match Self::send_with_retry(channel, msg).await {
|
||||
Ok(()) => {}
|
||||
Err(ChannelError::ChannelFull) => {
|
||||
@ -178,12 +180,16 @@ impl OutboundDispatcher {
|
||||
// 记 warn 而非 error：这是预期的背压丢弃。
|
||||
tracing::warn!(
|
||||
channel = %channel_name,
|
||||
chat_id = %msg_chat_id,
|
||||
trace_id = %msg_trace_id,
|
||||
"Message dropped: channel queue full"
|
||||
);
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::error!(
|
||||
channel = %channel_name,
|
||||
chat_id = %msg_chat_id,
|
||||
trace_id = %msg_trace_id,
|
||||
error = %error,
|
||||
"Failed to send message after retries"
|
||||
);
|
||||
|
||||
@ -147,17 +147,15 @@ impl InboundProcessor {
|
||||
}
|
||||
};
|
||||
|
||||
#[cfg(debug_assertions)]
|
||||
{
|
||||
tracing::debug!(
|
||||
channel = %inbound.channel,
|
||||
chat_id = %inbound.chat_id,
|
||||
sender = %inbound.sender_id,
|
||||
content_len = %inbound.content.len(),
|
||||
media_count = %inbound.media.len(),
|
||||
"Processing inbound message"
|
||||
);
|
||||
}
|
||||
tracing::debug!(
|
||||
channel = %inbound.channel,
|
||||
chat_id = %inbound.chat_id,
|
||||
trace_id = %inbound.trace_id,
|
||||
sender = %inbound.sender_id,
|
||||
content_len = %inbound.content.len(),
|
||||
media_count = %inbound.media.len(),
|
||||
"Processing inbound message"
|
||||
);
|
||||
|
||||
// 2. 获取 semaphore permit（控制并发）
|
||||
let permit = match self.semaphore.clone().acquire_owned().await {
|
||||
@ -172,18 +170,37 @@ impl InboundProcessor {
|
||||
let processor = self.clone();
|
||||
|
||||
// 4. 独立任务处理（包含 permit，任务完成自动释放）
|
||||
tokio::spawn(async move {
|
||||
let _permit = permit; // 持有 permit 直到任务完成
|
||||
if let Err(e) = processor.process_one(inbound).await {
|
||||
tracing::error!(error = %e, "Message processing failed");
|
||||
}
|
||||
});
|
||||
// spawn 不自动传播父 span，用 traced() 重建 span 上下文，
|
||||
// 使 process_one 内所有日志携带 trace_id/chat_id/session_id。
|
||||
let trace_id = inbound.trace_id.clone();
|
||||
let chat_id_for_span = inbound.chat_id.clone();
|
||||
let session_id_for_span =
|
||||
crate::storage::persistent_session_id(&inbound.channel, &inbound.chat_id);
|
||||
tokio::spawn(
|
||||
crate::observability::tracing_ctx::traced(
|
||||
&trace_id,
|
||||
&chat_id_for_span,
|
||||
&session_id_for_span,
|
||||
async move {
|
||||
let _permit = permit; // 持有 permit 直到任务完成
|
||||
if let Err(e) = processor.process_one(inbound).await {
|
||||
tracing::error!(
|
||||
error = %crate::utils::format_error_chain(&e),
|
||||
"Message processing failed"
|
||||
);
|
||||
crate::observability::metrics::record_message_processing_error();
|
||||
}
|
||||
},
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self, inbound), fields(trace_id = %inbound.trace_id, chat_id = %inbound.chat_id, session_id))]
|
||||
async fn process_one(&self, inbound: InboundMessage) -> Result<(), AgentError> {
|
||||
// 计算正确的 session_id（根据 channel_name 和 chat_id）
|
||||
let session_id = persistent_session_id(&inbound.channel, &inbound.chat_id);
|
||||
tracing::Span::current().record("session_id", tracing::field::display(&session_id));
|
||||
|
||||
// 获取当前话题（封装了 session 创建逻辑）
|
||||
let current_topic = self
|
||||
@ -220,14 +237,17 @@ impl InboundProcessor {
|
||||
for msg in &response.messages {
|
||||
if let Err(error) = self
|
||||
.bus
|
||||
.publish_outbound(OutboundMessage::assistant(
|
||||
inbound.channel.clone(),
|
||||
inbound.chat_id.clone(),
|
||||
response.metadata.get("session_id").cloned(),
|
||||
msg.content.clone(),
|
||||
None,
|
||||
inbound.forwarded_metadata.clone(),
|
||||
))
|
||||
.publish_outbound(
|
||||
OutboundMessage::assistant(
|
||||
inbound.channel.clone(),
|
||||
inbound.chat_id.clone(),
|
||||
response.metadata.get("session_id").cloned(),
|
||||
msg.content.clone(),
|
||||
None,
|
||||
inbound.forwarded_metadata.clone(),
|
||||
)
|
||||
.with_trace_id(&inbound.trace_id),
|
||||
)
|
||||
.await
|
||||
{
|
||||
match error {
|
||||
@ -243,14 +263,17 @@ impl InboundProcessor {
|
||||
} else if let Some(error) = response.error {
|
||||
if let Err(e) = self
|
||||
.bus
|
||||
.publish_outbound(OutboundMessage::assistant(
|
||||
inbound.channel.clone(),
|
||||
inbound.chat_id.clone(),
|
||||
response.metadata.get("session_id").cloned(),
|
||||
format!("Error [{}]: {}", error.code, error.message),
|
||||
None,
|
||||
inbound.forwarded_metadata.clone(),
|
||||
))
|
||||
.publish_outbound(
|
||||
OutboundMessage::assistant(
|
||||
inbound.channel.clone(),
|
||||
inbound.chat_id.clone(),
|
||||
response.metadata.get("session_id").cloned(),
|
||||
format!("Error [{}]: {}", error.code, error.message),
|
||||
None,
|
||||
inbound.forwarded_metadata.clone(),
|
||||
)
|
||||
.with_trace_id(&inbound.trace_id),
|
||||
)
|
||||
.await
|
||||
{
|
||||
match e {
|
||||
@ -279,6 +302,7 @@ impl InboundProcessor {
|
||||
inbound.chat_id.clone(),
|
||||
emitter_metadata,
|
||||
self.session_manager.store(),
|
||||
inbound.trace_id.clone(),
|
||||
),
|
||||
self.session_manager.store(),
|
||||
&session_id,
|
||||
@ -307,6 +331,7 @@ impl InboundProcessor {
|
||||
inbound.media,
|
||||
Some(live_emitter),
|
||||
current_topic.as_deref(),
|
||||
&inbound.trace_id,
|
||||
)
|
||||
.await
|
||||
{
|
||||
@ -319,6 +344,8 @@ impl InboundProcessor {
|
||||
.metadata
|
||||
.insert("topic_id".to_string(), topic_id.clone());
|
||||
}
|
||||
// 透传 trace_id 到出站消息，保持端到端追踪贯通
|
||||
outbound.trace_id = inbound.trace_id.clone();
|
||||
if let Err(error) = self.bus.publish_outbound(outbound).await {
|
||||
match error {
|
||||
crate::bus::BusError::Dropped => {
|
||||
@ -414,19 +441,26 @@ impl InboundProcessor {
|
||||
}
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::error!(error = %error, "Failed to handle message");
|
||||
tracing::error!(
|
||||
error = %crate::utils::format_error_chain(&error),
|
||||
"Failed to handle message"
|
||||
);
|
||||
crate::observability::metrics::record_message_processing_error();
|
||||
let mut metadata = inbound.forwarded_metadata.clone();
|
||||
metadata.insert("error_kind".to_string(), "agent_execution".to_string());
|
||||
if let Err(publish_error) = self
|
||||
.bus
|
||||
.publish_outbound(OutboundMessage::error_notification(
|
||||
inbound.channel,
|
||||
inbound.chat_id,
|
||||
None, // session_id
|
||||
error.to_string(),
|
||||
None,
|
||||
metadata,
|
||||
))
|
||||
.publish_outbound(
|
||||
OutboundMessage::error_notification(
|
||||
inbound.channel,
|
||||
inbound.chat_id,
|
||||
None, // session_id
|
||||
error.to_string(),
|
||||
None,
|
||||
metadata,
|
||||
)
|
||||
.with_trace_id(&inbound.trace_id),
|
||||
)
|
||||
.await
|
||||
{
|
||||
match publish_error {
|
||||
@ -454,12 +488,15 @@ impl InboundProcessor {
|
||||
}
|
||||
if let Err(error) = self
|
||||
.bus
|
||||
.publish_outbound(OutboundMessage::execution_completed(
|
||||
channel,
|
||||
chat_id,
|
||||
Some(session_id),
|
||||
completion_metadata,
|
||||
))
|
||||
.publish_outbound(
|
||||
OutboundMessage::execution_completed(
|
||||
channel,
|
||||
chat_id,
|
||||
Some(session_id),
|
||||
completion_metadata,
|
||||
)
|
||||
.with_trace_id(&inbound.trace_id),
|
||||
)
|
||||
.await
|
||||
{
|
||||
match error {
|
||||
|
||||
@ -312,6 +312,8 @@ pub(crate) fn build_session_manager_with_sender(
|
||||
|
||||
let prompt_repository: Arc<dyn PromptInjectionRepository> = store.clone();
|
||||
let model_selections = Arc::new(ModelSelectionStore::new());
|
||||
let observer: Arc<dyn crate::observability::Observer> =
|
||||
crate::observability::metrics::default_observer();
|
||||
let agent_factory = AgentFactory::new(
|
||||
tools.clone(),
|
||||
skills.clone(),
|
||||
@ -322,6 +324,7 @@ pub(crate) fn build_session_manager_with_sender(
|
||||
model_resolver.clone(),
|
||||
model_selections.clone(),
|
||||
compaction_config,
|
||||
Some(observer),
|
||||
);
|
||||
let session_factory = SessionFactory::new(
|
||||
provider_config.clone(),
|
||||
|
||||
@ -47,7 +47,10 @@ impl ScheduledAgentTaskService {
|
||||
.unwrap_or_else(|| "scheduler".to_string());
|
||||
let provider_config = self.provider_configs.select(options.agent.as_deref())?;
|
||||
|
||||
AgentExecutionService::new(self.show_tool_results)
|
||||
// 定时任务没有入站消息，在此生成独立 trace_id 以贯穿 agent → tool → outbound
|
||||
let trace_id = crate::observability::tracing_ctx::new_trace_id();
|
||||
|
||||
let mut outbound_messages = AgentExecutionService::new(self.show_tool_results)
|
||||
.prepare_and_execute_scheduled_task(ScheduledExecutionRequest {
|
||||
session,
|
||||
channel_name,
|
||||
@ -59,7 +62,15 @@ impl ScheduledAgentTaskService {
|
||||
system_prompt: options.system_prompt.as_deref(),
|
||||
metadata: &options.metadata,
|
||||
fresh_session: options.fresh_session,
|
||||
trace_id: trace_id.clone(),
|
||||
})
|
||||
.await
|
||||
.await?;
|
||||
|
||||
// 将 trace_id 透传到出站消息，保持端到端追踪贯通
|
||||
for msg in &mut outbound_messages {
|
||||
msg.trace_id = trace_id.clone();
|
||||
}
|
||||
|
||||
Ok(outbound_messages)
|
||||
}
|
||||
}
|
||||
|
||||
@ -61,6 +61,7 @@ pub struct BusToolCallEmitter {
|
||||
metadata: HashMap<String, String>,
|
||||
store: Arc<SessionStore>,
|
||||
stream_message_id: parking_lot::Mutex<Option<String>>,
|
||||
trace_id: String,
|
||||
}
|
||||
|
||||
impl BusToolCallEmitter {
|
||||
@ -70,6 +71,7 @@ impl BusToolCallEmitter {
|
||||
chat_id: impl Into<String>,
|
||||
metadata: HashMap<String, String>,
|
||||
store: Arc<SessionStore>,
|
||||
trace_id: impl Into<String>,
|
||||
) -> Self {
|
||||
Self {
|
||||
bus,
|
||||
@ -78,6 +80,7 @@ impl BusToolCallEmitter {
|
||||
metadata,
|
||||
store,
|
||||
stream_message_id: parking_lot::Mutex::new(None),
|
||||
trace_id: trace_id.into(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -85,7 +88,7 @@ impl BusToolCallEmitter {
|
||||
#[async_trait]
|
||||
impl EmittedMessageHandler for BusToolCallEmitter {
|
||||
async fn handle(&self, message: ChatMessage) {
|
||||
for outbound in OutboundMessage::from_chat_message(
|
||||
for mut outbound in OutboundMessage::from_chat_message(
|
||||
&self.channel_name,
|
||||
&self.chat_id,
|
||||
None, // session_id
|
||||
@ -93,6 +96,7 @@ impl EmittedMessageHandler for BusToolCallEmitter {
|
||||
&self.metadata,
|
||||
&message,
|
||||
) {
|
||||
outbound.trace_id = self.trace_id.clone();
|
||||
if let Err(error) = self.bus.publish_outbound(outbound).await {
|
||||
match error {
|
||||
crate::bus::BusError::Dropped => {
|
||||
@ -111,7 +115,7 @@ impl EmittedMessageHandler for BusToolCallEmitter {
|
||||
if let Some(ms) = duration_ms {
|
||||
metadata.insert("tool_duration_ms".to_string(), ms.to_string());
|
||||
}
|
||||
for outbound in OutboundMessage::from_chat_message(
|
||||
for mut outbound in OutboundMessage::from_chat_message(
|
||||
&self.channel_name,
|
||||
&self.chat_id,
|
||||
None, // session_id
|
||||
@ -119,6 +123,7 @@ impl EmittedMessageHandler for BusToolCallEmitter {
|
||||
&metadata,
|
||||
&message,
|
||||
) {
|
||||
outbound.trace_id = self.trace_id.clone();
|
||||
if let Err(error) = self.bus.publish_outbound(outbound).await {
|
||||
match error {
|
||||
crate::bus::BusError::Dropped => {
|
||||
@ -147,7 +152,7 @@ impl EmittedMessageHandler for BusToolCallEmitter {
|
||||
};
|
||||
|
||||
// Empty content + no reasoning = stream end signal
|
||||
let outbound = if delta.content.is_empty() && delta.reasoning_content.is_none() {
|
||||
let mut outbound = if delta.content.is_empty() && delta.reasoning_content.is_none() {
|
||||
OutboundMessage::stream_end(
|
||||
&self.channel_name,
|
||||
&self.chat_id,
|
||||
@ -166,6 +171,7 @@ impl EmittedMessageHandler for BusToolCallEmitter {
|
||||
self.metadata.clone(),
|
||||
)
|
||||
};
|
||||
outbound.trace_id = self.trace_id.clone();
|
||||
|
||||
if let Err(error) = self.bus.publish_outbound(outbound).await {
|
||||
match error {
|
||||
@ -302,6 +308,7 @@ impl Session {
|
||||
model_resolver,
|
||||
Arc::new(super::model_selection::ModelSelectionStore::new()),
|
||||
crate::config::CompactionConfig::default(),
|
||||
None,
|
||||
);
|
||||
Self::with_factories(
|
||||
channel_name,
|
||||
@ -613,6 +620,7 @@ impl Session {
|
||||
sender_id: Option<&str>,
|
||||
message_id: Option<&str>,
|
||||
explicit_topic_id: Option<&str>,
|
||||
trace_id: &str,
|
||||
) -> Result<AgentLoop, AgentError> {
|
||||
self.create_agent_with_provider_config(
|
||||
chat_id,
|
||||
@ -621,6 +629,7 @@ impl Session {
|
||||
message_id,
|
||||
self.provider_config.clone(),
|
||||
explicit_topic_id,
|
||||
trace_id,
|
||||
)
|
||||
}
|
||||
|
||||
@ -632,6 +641,7 @@ impl Session {
|
||||
message_id: Option<&str>,
|
||||
provider_config: LLMProviderConfig,
|
||||
explicit_topic_id: Option<&str>,
|
||||
trace_id: &str,
|
||||
) -> Result<AgentLoop, AgentError> {
|
||||
// 优先使用显式传入的 topic_id；回退到当前 chat 的活跃 topic
|
||||
let topic_id = explicit_topic_id
|
||||
@ -657,6 +667,7 @@ impl Session {
|
||||
message_id,
|
||||
provider_config,
|
||||
cancel_token,
|
||||
trace_id: Some(trace_id.to_string()),
|
||||
})
|
||||
}
|
||||
}
|
||||
@ -908,6 +919,7 @@ impl SessionManager {
|
||||
media: Vec<crate::bus::MediaItem>,
|
||||
live_emitter: Option<Arc<dyn EmittedMessageHandler>>,
|
||||
topic_id: Option<&str>,
|
||||
trace_id: &str,
|
||||
) -> Result<Vec<OutboundMessage>, AgentError> {
|
||||
self.messages
|
||||
.handle_message(
|
||||
@ -918,6 +930,7 @@ impl SessionManager {
|
||||
media,
|
||||
live_emitter,
|
||||
topic_id,
|
||||
trace_id,
|
||||
)
|
||||
.await
|
||||
}
|
||||
@ -1305,6 +1318,7 @@ mod tests {
|
||||
Vec::new(),
|
||||
None,
|
||||
None,
|
||||
"test-trace",
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
@ -2095,8 +2109,14 @@ mod tests {
|
||||
async fn test_bus_tool_call_emitter_emits_completed_tool_results() {
|
||||
let store = Arc::new(SessionStore::in_memory().unwrap());
|
||||
let bus = MessageBus::new(4);
|
||||
let emitter =
|
||||
BusToolCallEmitter::new(bus.clone(), "test-channel", "chat-1", HashMap::new(), store);
|
||||
let emitter = BusToolCallEmitter::new(
|
||||
bus.clone(),
|
||||
"test-channel",
|
||||
"chat-1",
|
||||
HashMap::new(),
|
||||
store,
|
||||
"test-trace-id",
|
||||
);
|
||||
|
||||
emitter
|
||||
.handle(ChatMessage::tool("call-1", "calculator", "2"))
|
||||
|
||||
@ -51,7 +51,8 @@ impl SessionMessageSender for BusSessionMessageSender {
|
||||
text,
|
||||
None,
|
||||
metadata.clone(),
|
||||
);
|
||||
)
|
||||
.with_trace_id(context.trace_id.as_deref().unwrap_or(""));
|
||||
if attachment_count > 0 {
|
||||
outbound.media = request.attachments.clone();
|
||||
}
|
||||
@ -88,7 +89,8 @@ impl SessionMessageSender for BusSessionMessageSender {
|
||||
String::new(),
|
||||
None,
|
||||
metadata.clone(),
|
||||
);
|
||||
)
|
||||
.with_trace_id(context.trace_id.as_deref().unwrap_or(""));
|
||||
outbound.media = vec![attachment];
|
||||
match self.bus.publish_outbound(outbound).await {
|
||||
Ok(()) => {
|
||||
|
||||
@ -29,6 +29,7 @@ impl SessionMessageService {
|
||||
media: Vec<MediaItem>,
|
||||
live_emitter: Option<Arc<dyn EmittedMessageHandler>>,
|
||||
topic_id: Option<&str>,
|
||||
trace_id: &str,
|
||||
) -> Result<Vec<OutboundMessage>, AgentError> {
|
||||
#[cfg(debug_assertions)]
|
||||
{
|
||||
@ -56,6 +57,7 @@ impl SessionMessageService {
|
||||
media,
|
||||
live_emitter,
|
||||
topic_id: topic_id.map(|s| s.to_string()),
|
||||
trace_id,
|
||||
})
|
||||
.await?;
|
||||
|
||||
|
||||
@ -397,6 +397,7 @@ async fn handle_inbound(
|
||||
media,
|
||||
metadata: HashMap::new(),
|
||||
forwarded_metadata: HashMap::new(),
|
||||
trace_id: crate::observability::tracing_ctx::new_trace_id(),
|
||||
})
|
||||
.await
|
||||
.map_err(|error| AgentError::Other(error.to_string()))?;
|
||||
|
||||
@ -3,9 +3,11 @@ use chrono_tz::Tz;
|
||||
use std::path::PathBuf;
|
||||
use tracing_appender::rolling::{RollingFileAppender, Rotation};
|
||||
use tracing_subscriber::{
|
||||
fmt, fmt::time::FormatTime, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter,
|
||||
fmt, fmt::time::FormatTime, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer,
|
||||
};
|
||||
|
||||
use crate::config::LogFormat;
|
||||
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
struct ConfiguredTimestamp {
|
||||
timezone: Tz,
|
||||
@ -40,7 +42,10 @@ pub fn get_default_config_path() -> PathBuf {
|
||||
|
||||
/// Initialize logging with file appender
|
||||
/// Logs are written to ~/.picobot/logs/ with daily rotation
|
||||
pub fn init_logging(timezone: Tz) {
|
||||
///
|
||||
/// `log_format` 控制文件日志格式：Text（默认）或 Json（便于日志聚合）。
|
||||
/// 控制台始终使用文本格式（便于人读）。
|
||||
pub fn init_logging(timezone: Tz, log_format: LogFormat) {
|
||||
use std::sync::Once;
|
||||
static INIT: Once = Once::new();
|
||||
|
||||
@ -72,14 +77,28 @@ pub fn init_logging(timezone: Tz) {
|
||||
// Build subscriber with both console and file output
|
||||
let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
|
||||
|
||||
let file_layer = fmt::layer()
|
||||
.with_writer(file_appender)
|
||||
.with_timer(ConfiguredTimestamp { timezone })
|
||||
.with_ansi(false)
|
||||
.with_target(true)
|
||||
.with_level(true)
|
||||
.with_thread_ids(true);
|
||||
// 文件层：根据 log_format 选择 text 或 json
|
||||
let file_layer = match log_format {
|
||||
LogFormat::Json => fmt::layer()
|
||||
.with_writer(file_appender)
|
||||
.with_timer(ConfiguredTimestamp { timezone })
|
||||
.with_ansi(false)
|
||||
.with_target(true)
|
||||
.with_level(true)
|
||||
.with_thread_ids(true)
|
||||
.json()
|
||||
.boxed(),
|
||||
LogFormat::Text => fmt::layer()
|
||||
.with_writer(file_appender)
|
||||
.with_timer(ConfiguredTimestamp { timezone })
|
||||
.with_ansi(false)
|
||||
.with_target(true)
|
||||
.with_level(true)
|
||||
.with_thread_ids(true)
|
||||
.boxed(),
|
||||
};
|
||||
|
||||
// 控制台层：始终文本格式
|
||||
let console_layer = fmt::layer()
|
||||
.with_timer(ConfiguredTimestamp { timezone })
|
||||
.with_target(true)
|
||||
@ -91,7 +110,11 @@ pub fn init_logging(timezone: Tz) {
|
||||
.with(file_layer)
|
||||
.init();
|
||||
|
||||
tracing::info!("Logging initialized. Log directory: {}", log_dir.display());
|
||||
tracing::info!(
|
||||
log_format = ?log_format,
|
||||
log_dir = %log_dir.display(),
|
||||
"Logging initialized"
|
||||
);
|
||||
}
|
||||
|
||||
/// Initialize logging without file output (console only)
|
||||
|
||||
157
src/observability/metrics.rs
Normal file
157
src/observability/metrics.rs
Normal file
@ -0,0 +1,157 @@
|
||||
//! Metrics 基础设施：基于 `metrics` + `metrics-exporter-prometheus`。
|
||||
//!
|
||||
//! 提供 `MetricsObserver`（实现 `Observer` trait，桥接 agent_loop 事件到 metrics），
|
||||
//! 以及 Prometheus recorder 初始化。
|
||||
//!
|
||||
//! 设计原则：
|
||||
//! - 业务层（agent_loop）只认 `dyn Observer` trait，不感知 metrics 实现。
|
||||
//! - 具体指标名和 label 约定集中于此模块。
|
||||
//! - metrics 后端可替换（换掉 recorder + Observer 实现即可）。
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};
|
||||
|
||||
use super::{Observer, ObserverEvent};
|
||||
|
||||
// ============================================================================
|
||||
// 指标名常量
|
||||
// ============================================================================
|
||||
|
||||
/// LLM 请求耗时（直方图，秒）。Labels: provider, model
|
||||
pub const LLM_REQUEST_DURATION: &str = "picobot_llm_request_duration_seconds";
|
||||
/// LLM token 使用量（计数器）。Labels: provider, model, type (prompt/completion/total)
|
||||
pub const LLM_TOKENS_USED: &str = "picobot_llm_tokens_used_total";
|
||||
/// 工具执行耗时（直方图，秒）。Labels: tool
|
||||
pub const TOOL_EXECUTION_DURATION: &str = "picobot_tool_execution_duration_seconds";
|
||||
/// 工具执行总数（计数器）。Labels: tool, success (true/false)
|
||||
pub const TOOL_EXECUTION_TOTAL: &str = "picobot_tool_execution_total";
|
||||
/// Agent 迭代总数（计数器）
|
||||
pub const AGENT_ITERATIONS: &str = "picobot_agent_iterations_total";
|
||||
/// 消息处理错误总数（计数器）
|
||||
pub const MESSAGE_PROCESSING_ERRORS: &str = "picobot_message_processing_errors_total";
|
||||
|
||||
// ============================================================================
|
||||
// Recorder 初始化
|
||||
// ============================================================================
|
||||
|
||||
/// 安装 Prometheus recorder，返回 handle 供 `/metrics` 端点渲染。
|
||||
///
|
||||
/// 幂等：首次调用安装 recorder 并缓存 handle；后续调用（含热重启）返回缓存的 handle。
|
||||
/// 这避免了热重启后 `install_recorder()` 因 recorder 已安装而失败、导致 `/metrics` 返回 503 的问题。
|
||||
/// 返回 None 表示安装失败（非致命，metrics 静默降级）。
|
||||
static PROMETHEUS_HANDLE: std::sync::OnceLock<Option<PrometheusHandle>> = std::sync::OnceLock::new();
|
||||
|
||||
pub fn init_recorder() -> Option<PrometheusHandle> {
|
||||
PROMETHEUS_HANDLE
|
||||
.get_or_init(|| {
|
||||
let builder = PrometheusBuilder::new();
|
||||
match builder.install_recorder() {
|
||||
Ok(handle) => {
|
||||
tracing::info!("Prometheus metrics recorder installed");
|
||||
Some(handle)
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "Failed to install Prometheus recorder (metrics will be no-op)");
|
||||
None
|
||||
}
|
||||
}
|
||||
})
|
||||
.clone()
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// MetricsObserver — 桥接 ObserverEvent 到 metrics 宏
|
||||
// ============================================================================
|
||||
|
||||
/// 将 `ObserverEvent` 转换为 metrics 指标的 Observer 实现。
|
||||
///
|
||||
/// 通过 `AgentFactory` 依赖注入到 `AgentLoop`，agent_loop 不感知 metrics 实现。
|
||||
pub struct MetricsObserver;
|
||||
|
||||
impl MetricsObserver {
|
||||
pub fn new() -> Self {
|
||||
Self
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for MetricsObserver {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl Observer for MetricsObserver {
|
||||
fn record_event(&self, event: &ObserverEvent) {
|
||||
match event {
|
||||
ObserverEvent::ToolCallStart { tool, .. } => {
|
||||
// 工具开始：不记录指标，仅 span 日志已覆盖
|
||||
let _ = tool;
|
||||
}
|
||||
ObserverEvent::ToolCall {
|
||||
tool,
|
||||
duration,
|
||||
success,
|
||||
} => {
|
||||
let duration_secs = duration.as_secs_f64();
|
||||
metrics::histogram!(TOOL_EXECUTION_DURATION, "tool" => tool.clone())
|
||||
.record(duration_secs);
|
||||
metrics::counter!(
|
||||
TOOL_EXECUTION_TOTAL,
|
||||
"tool" => tool.clone(),
|
||||
"success" => success.to_string()
|
||||
)
|
||||
.increment(1);
|
||||
}
|
||||
ObserverEvent::AgentStart { provider, model } => {
|
||||
metrics::counter!(
|
||||
AGENT_ITERATIONS,
|
||||
"provider" => provider.clone(),
|
||||
"model" => model.clone()
|
||||
)
|
||||
.increment(1);
|
||||
}
|
||||
ObserverEvent::AgentEnd {
|
||||
provider,
|
||||
model,
|
||||
duration,
|
||||
tokens_used,
|
||||
} => {
|
||||
let duration_secs = duration.as_secs_f64();
|
||||
metrics::histogram!(
|
||||
LLM_REQUEST_DURATION,
|
||||
"provider" => provider.clone(),
|
||||
"model" => model.clone()
|
||||
)
|
||||
.record(duration_secs);
|
||||
if let Some(tokens) = tokens_used {
|
||||
metrics::counter!(
|
||||
LLM_TOKENS_USED,
|
||||
"provider" => provider.clone(),
|
||||
"model" => model.clone(),
|
||||
"type" => "total"
|
||||
)
|
||||
.increment(*tokens);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn name(&self) -> &str {
|
||||
"metrics_observer"
|
||||
}
|
||||
}
|
||||
|
||||
/// 创建默认的 `Arc<dyn Observer>`（供 AgentFactory 注入）。
|
||||
pub fn default_observer() -> Arc<dyn Observer> {
|
||||
Arc::new(MetricsObserver::new())
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// 辅助函数：供非 agent_loop 路径直接记录指标
|
||||
// ============================================================================
|
||||
|
||||
/// 记录消息处理错误（供 processor 错误路径调用）。
|
||||
pub fn record_message_processing_error() {
|
||||
metrics::counter!(MESSAGE_PROCESSING_ERRORS).increment(1);
|
||||
}
|
||||
@ -3,6 +3,9 @@
|
||||
//! This module provides an Observer pattern for emitting and collecting
|
||||
//! telemetry events during agent execution.
|
||||
|
||||
pub mod metrics;
|
||||
pub mod tracing_ctx;
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
|
||||
43
src/observability/tracing_ctx.rs
Normal file
43
src/observability/tracing_ctx.rs
Normal file
@ -0,0 +1,43 @@
|
||||
//! 观测上下文辅助：trace_id 生成与 span 创建。
|
||||
//!
|
||||
//! 集中管理 trace_id 的生成和 span 字段命名，避免散落在各模块。
|
||||
//! 业务代码调用此模块的辅助函数，不直接拼 span 字段。
|
||||
|
||||
use tracing::Instrument;
|
||||
|
||||
use std::future::Future;
|
||||
|
||||
/// 生成新的 trace_id（UUID v4）。
|
||||
pub fn new_trace_id() -> String {
|
||||
uuid::Uuid::new_v4().to_string()
|
||||
}
|
||||
|
||||
/// 在携带 trace_id/chat_id/session_id 的 span 内执行 future。
|
||||
///
|
||||
/// 用于 `tokio::spawn` 边界：spawn 不自动传播父 span，
|
||||
/// 调用此函数在 spawn 的 async block 内重建 span 上下文。
|
||||
///
|
||||
/// # 示例
|
||||
/// ```ignore
|
||||
/// tokio::spawn(
|
||||
/// traced(&trace_id, &chat_id, &session_id, async move {
|
||||
/// // 此处所有 tracing 日志自动携带 trace_id/chat_id/session_id
|
||||
/// process_one(inbound).await
|
||||
/// })
|
||||
/// );
|
||||
/// ```
|
||||
pub fn traced<F>(trace_id: &str, chat_id: &str, session_id: &str, f: F) -> Instrumented<F>
|
||||
where
|
||||
F: Future,
|
||||
{
|
||||
let span = tracing::info_span!(
|
||||
"request",
|
||||
trace_id = %trace_id,
|
||||
chat_id = %chat_id,
|
||||
session_id = %session_id
|
||||
);
|
||||
f.instrument(span)
|
||||
}
|
||||
|
||||
/// tracing::Instrumented 的重新导出，便于调用方使用。
|
||||
pub type Instrumented<F> = tracing::instrument::Instrumented<F>;
|
||||
@ -252,6 +252,7 @@ struct AnthropicUsage {
|
||||
|
||||
#[async_trait]
|
||||
impl LLMProvider for AnthropicProvider {
|
||||
#[tracing::instrument(skip(self, request), fields(provider = %self.name, model = %self.model_id))]
|
||||
async fn chat(
|
||||
&self,
|
||||
request: ChatCompletionRequest,
|
||||
@ -259,6 +260,14 @@ impl LLMProvider for AnthropicProvider {
|
||||
let url = format!("{}/v1/messages", self.base_url);
|
||||
let max_tokens = request.max_tokens.or(self.max_tokens).unwrap_or(1024);
|
||||
|
||||
tracing::info!(
|
||||
provider = %self.name,
|
||||
model = %self.model_id,
|
||||
message_count = request.messages.len(),
|
||||
has_tools = request.tools.is_some(),
|
||||
"Anthropic: sending chat completion request"
|
||||
);
|
||||
|
||||
let tools = request.tools.map(|tools| {
|
||||
tools
|
||||
.iter()
|
||||
@ -304,7 +313,16 @@ impl LLMProvider for AnthropicProvider {
|
||||
req_builder = req_builder.header(key.as_str(), value.as_str());
|
||||
}
|
||||
|
||||
let resp = req_builder.json(&body).send().await?;
|
||||
let resp = req_builder.json(&body).send().await.map_err(|e| {
|
||||
tracing::error!(
|
||||
provider = %self.name,
|
||||
model = %self.model_id,
|
||||
url = %url,
|
||||
error = %format_error_chain(&e),
|
||||
"Anthropic: HTTP request failed"
|
||||
);
|
||||
e
|
||||
})?;
|
||||
let status = resp.status();
|
||||
let text = resp.text().await?;
|
||||
|
||||
@ -321,11 +339,13 @@ impl LLMProvider for AnthropicProvider {
|
||||
return Err(format!("API error {}: {}", status, text).into());
|
||||
}
|
||||
|
||||
#[cfg(debug_assertions)]
|
||||
{
|
||||
let resp_preview: String = text.chars().take(100).collect();
|
||||
tracing::debug!(status = %status, response_preview = %resp_preview, response_len = %text.len(), timeout_secs = self.llm_timeout_secs, "Anthropic response (first 100 chars shown)");
|
||||
}
|
||||
tracing::debug!(
|
||||
provider = %self.name,
|
||||
model = %self.model_id,
|
||||
status = %status,
|
||||
response_len = text.len(),
|
||||
"Anthropic response received"
|
||||
);
|
||||
|
||||
let anthropic_resp: AnthropicResponse = serde_json::from_str(&text).map_err(|e| {
|
||||
tracing::error!(
|
||||
@ -364,18 +384,29 @@ impl LLMProvider for AnthropicProvider {
|
||||
}
|
||||
}
|
||||
|
||||
let usage = Usage {
|
||||
prompt_tokens: anthropic_resp.usage.input_tokens,
|
||||
completion_tokens: anthropic_resp.usage.output_tokens,
|
||||
total_tokens: anthropic_resp.usage.input_tokens + anthropic_resp.usage.output_tokens,
|
||||
};
|
||||
|
||||
tracing::info!(
|
||||
provider = %self.name,
|
||||
model = %self.model_id,
|
||||
prompt_tokens = usage.prompt_tokens,
|
||||
completion_tokens = usage.completion_tokens,
|
||||
total_tokens = usage.total_tokens,
|
||||
has_tool_calls = !tool_calls.is_empty(),
|
||||
"Anthropic: chat completion completed"
|
||||
);
|
||||
|
||||
Ok(ChatCompletionResponse {
|
||||
id: anthropic_resp.id,
|
||||
model: anthropic_resp.model,
|
||||
content,
|
||||
reasoning_content: None,
|
||||
tool_calls,
|
||||
usage: Usage {
|
||||
prompt_tokens: anthropic_resp.usage.input_tokens,
|
||||
completion_tokens: anthropic_resp.usage.output_tokens,
|
||||
total_tokens: anthropic_resp.usage.input_tokens
|
||||
+ anthropic_resp.usage.output_tokens,
|
||||
},
|
||||
usage,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@ -115,12 +115,13 @@ struct SubAgentEmitter {
|
||||
/// 子/孙智能体自身的 task_id，用于持久化时作为 scope_key
|
||||
task_id: String,
|
||||
stream_message_id: parking_lot::Mutex<Option<String>>,
|
||||
trace_id: Option<String>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EmittedMessageHandler for SubAgentEmitter {
|
||||
async fn handle(&self, message: ChatMessage) {
|
||||
for outbound in OutboundMessage::from_chat_message(
|
||||
for mut outbound in OutboundMessage::from_chat_message(
|
||||
&self.channel_name,
|
||||
&self.chat_id,
|
||||
None,
|
||||
@ -128,6 +129,9 @@ impl EmittedMessageHandler for SubAgentEmitter {
|
||||
&self.metadata,
|
||||
&message,
|
||||
) {
|
||||
if let Some(ref tid) = self.trace_id {
|
||||
outbound.trace_id = tid.clone();
|
||||
}
|
||||
if let Err(error) = self.bus.publish_outbound(outbound).await {
|
||||
match error {
|
||||
crate::bus::BusError::Dropped => {
|
||||
@ -146,7 +150,7 @@ impl EmittedMessageHandler for SubAgentEmitter {
|
||||
if let Some(ms) = duration_ms {
|
||||
metadata.insert("tool_duration_ms".to_string(), ms.to_string());
|
||||
}
|
||||
for outbound in OutboundMessage::from_chat_message(
|
||||
for mut outbound in OutboundMessage::from_chat_message(
|
||||
&self.channel_name,
|
||||
&self.chat_id,
|
||||
None,
|
||||
@ -154,6 +158,9 @@ impl EmittedMessageHandler for SubAgentEmitter {
|
||||
&metadata,
|
||||
&message,
|
||||
) {
|
||||
if let Some(ref tid) = self.trace_id {
|
||||
outbound.trace_id = tid.clone();
|
||||
}
|
||||
if let Err(error) = self.bus.publish_outbound(outbound).await {
|
||||
match error {
|
||||
crate::bus::BusError::Dropped => {
|
||||
@ -180,7 +187,7 @@ impl EmittedMessageHandler for SubAgentEmitter {
|
||||
.clone()
|
||||
};
|
||||
|
||||
let outbound = if delta.content.is_empty() && delta.reasoning_content.is_none() {
|
||||
let mut outbound = if delta.content.is_empty() && delta.reasoning_content.is_none() {
|
||||
OutboundMessage::stream_end(
|
||||
&self.channel_name,
|
||||
&self.chat_id,
|
||||
@ -199,6 +206,9 @@ impl EmittedMessageHandler for SubAgentEmitter {
|
||||
self.metadata.clone(),
|
||||
)
|
||||
};
|
||||
if let Some(ref tid) = self.trace_id {
|
||||
outbound.trace_id = tid.clone();
|
||||
}
|
||||
|
||||
if let Err(error) = self.bus.publish_outbound(outbound).await {
|
||||
match error {
|
||||
@ -306,16 +316,23 @@ fn build_subagent_event_metadata(session: &TaskSession) -> HashMap<String, Strin
|
||||
}
|
||||
|
||||
/// 发布子智能体执行完成事件（ExecutionCompleted），metadata 含 subagent_task_id。
|
||||
async fn publish_subagent_completion(bus: &Option<Arc<MessageBus>>, session: &TaskSession) {
|
||||
async fn publish_subagent_completion(
|
||||
bus: &Option<Arc<MessageBus>>,
|
||||
session: &TaskSession,
|
||||
trace_id: &str,
|
||||
) {
|
||||
if let Some(bus) = bus {
|
||||
let metadata = build_subagent_event_metadata(session);
|
||||
if let Err(e) = bus
|
||||
.publish_outbound(OutboundMessage::execution_completed(
|
||||
session.parent_channel_name.clone(),
|
||||
session.parent_chat_id.clone(),
|
||||
Some(session.parent_session_id.clone()),
|
||||
metadata,
|
||||
))
|
||||
.publish_outbound(
|
||||
OutboundMessage::execution_completed(
|
||||
session.parent_channel_name.clone(),
|
||||
session.parent_chat_id.clone(),
|
||||
Some(session.parent_session_id.clone()),
|
||||
metadata,
|
||||
)
|
||||
.with_trace_id(trace_id),
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, task_id = %session.id, "Failed to publish subagent execution_completed");
|
||||
@ -328,18 +345,22 @@ async fn publish_subagent_error(
|
||||
bus: &Option<Arc<MessageBus>>,
|
||||
session: &TaskSession,
|
||||
error_msg: &str,
|
||||
trace_id: &str,
|
||||
) {
|
||||
if let Some(bus) = bus {
|
||||
let metadata = build_subagent_event_metadata(session);
|
||||
if let Err(e) = bus
|
||||
.publish_outbound(OutboundMessage::error_notification(
|
||||
session.parent_channel_name.clone(),
|
||||
session.parent_chat_id.clone(),
|
||||
Some(session.parent_session_id.clone()),
|
||||
error_msg.to_string(),
|
||||
None,
|
||||
metadata,
|
||||
))
|
||||
.publish_outbound(
|
||||
OutboundMessage::error_notification(
|
||||
session.parent_channel_name.clone(),
|
||||
session.parent_chat_id.clone(),
|
||||
Some(session.parent_session_id.clone()),
|
||||
error_msg.to_string(),
|
||||
None,
|
||||
metadata,
|
||||
)
|
||||
.with_trace_id(trace_id),
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, task_id = %session.id, "Failed to publish subagent error notification");
|
||||
@ -480,6 +501,7 @@ impl DefaultSubAgentRuntime {
|
||||
def: Option<&SubagentDef>,
|
||||
parent_nesting_depth: u32,
|
||||
parent_task_id: Option<String>,
|
||||
trace_id: Option<String>,
|
||||
) -> Result<AgentLoop, TaskError> {
|
||||
let prompt_provider = Arc::new(StaticSystemPromptProvider::new(system_prompt));
|
||||
|
||||
@ -529,6 +551,8 @@ impl DefaultSubAgentRuntime {
|
||||
// 子代理自身的 capability 作为孙代理的 parent_capability，
|
||||
// 使孙代理的 TaskTool 能按此策略校验（与主 agent 注入专家 capability 同构）
|
||||
parent_capability: def.map(|d| d.capability.clone()),
|
||||
// 从父 ToolContext 继承 trace_id，保持端到端追踪贯通子代理
|
||||
trace_id: trace_id.clone(),
|
||||
});
|
||||
|
||||
// 如果有 MessageBus，附加实时广播 emitter
|
||||
@ -550,6 +574,7 @@ impl DefaultSubAgentRuntime {
|
||||
store: self.store.clone(),
|
||||
task_id: session.id.clone(),
|
||||
stream_message_id: parking_lot::Mutex::new(None),
|
||||
trace_id: trace_id.clone(),
|
||||
},
|
||||
self.conversation_repository.clone(),
|
||||
session.session_id.clone(),
|
||||
@ -658,6 +683,7 @@ impl DefaultSubAgentRuntime {
|
||||
&self,
|
||||
session: TaskSession,
|
||||
error: TaskError,
|
||||
trace_id: &str,
|
||||
) -> Result<TaskToolResult, TaskError> {
|
||||
let status = error.as_status();
|
||||
tracing::warn!(
|
||||
@ -674,7 +700,7 @@ impl DefaultSubAgentRuntime {
|
||||
session.mark_failed(error.to_string());
|
||||
}
|
||||
self.task_repository.save_task_session(&session).await?;
|
||||
publish_subagent_error(&self.bus, &session, &error.to_string()).await;
|
||||
publish_subagent_error(&self.bus, &session, &error.to_string(), trace_id).await;
|
||||
Ok(TaskToolResult {
|
||||
status: status.to_string(),
|
||||
summary: error.to_string(),
|
||||
@ -791,6 +817,7 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
|
||||
tool_arguments: None,
|
||||
reasoning_content: None,
|
||||
message_id: None,
|
||||
trace_id: parent_context.trace_id.clone().unwrap_or_default(),
|
||||
};
|
||||
|
||||
if let Err(e) = bus.publish_outbound(event).await {
|
||||
@ -844,6 +871,7 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
|
||||
Some(&def),
|
||||
parent_context.nesting_depth,
|
||||
parent_context.task_id.clone(),
|
||||
parent_context.trace_id.clone(),
|
||||
)?;
|
||||
|
||||
// 8. 执行任务
|
||||
@ -852,6 +880,7 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
|
||||
};
|
||||
|
||||
// 9. 更新会话状态并保存
|
||||
let trace_id = parent_context.trace_id.as_deref().unwrap_or("");
|
||||
match result {
|
||||
Ok(tool_result) => {
|
||||
let mut session = session;
|
||||
@ -863,13 +892,13 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
|
||||
);
|
||||
self.task_repository.save_task_session(&session).await?;
|
||||
// 发布子智能体 ExecutionCompleted，metadata 注入 subagent_task_id 供前端路由到对应子智能体层
|
||||
publish_subagent_completion(&self.bus, &session).await;
|
||||
publish_subagent_completion(&self.bus, &session, trace_id).await;
|
||||
Ok(tool_result)
|
||||
}
|
||||
Err(e) => {
|
||||
// 会话创建后的任何失败（含 AgentCreationFailed、Timeout、ExecutionFailed）
|
||||
// 统一返回结构化结果，携带 task_id 供前端导航
|
||||
self.handle_task_failure(session, e).await
|
||||
self.handle_task_failure(session, e, trace_id).await
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -939,24 +968,26 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
|
||||
Some(&def),
|
||||
parent_context.nesting_depth,
|
||||
parent_context.task_id.clone(),
|
||||
parent_context.trace_id.clone(),
|
||||
)?;
|
||||
self.execute_task_with_history(agent, &session, additional_prompt)
|
||||
.await
|
||||
};
|
||||
|
||||
// 7. 更新会话状态
|
||||
let trace_id = parent_context.trace_id.as_deref().unwrap_or("");
|
||||
match result {
|
||||
Ok(tool_result) => {
|
||||
let mut session = session;
|
||||
session.mark_completed(tool_result.summary.clone());
|
||||
self.task_repository.save_task_session(&session).await?;
|
||||
// 发布子智能体 ExecutionCompleted，metadata 注入 subagent_task_id 供前端路由到对应子智能体层
|
||||
publish_subagent_completion(&self.bus, &session).await;
|
||||
publish_subagent_completion(&self.bus, &session, trace_id).await;
|
||||
Ok(tool_result)
|
||||
}
|
||||
Err(e) => {
|
||||
// 修复：原代码一律 mark_failed，未处理 timeout；现统一走 handle_task_failure
|
||||
self.handle_task_failure(session, e).await
|
||||
self.handle_task_failure(session, e, trace_id).await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -184,6 +184,7 @@ mod tests {
|
||||
parent_task_id: None,
|
||||
tool_call_id: None,
|
||||
parent_capability: None,
|
||||
trace_id: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -486,6 +486,7 @@ mod tests {
|
||||
parent_task_id: None,
|
||||
tool_call_id: None,
|
||||
parent_capability: None,
|
||||
trace_id: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -32,6 +32,9 @@ pub struct ToolContext {
|
||||
/// TaskTool 据此强制校验子代理加载（白/黑名单），与 spawn/resume 安全范式一致。
|
||||
/// 以数据形式传递，避免 task 模块反向依赖 experts 模块。
|
||||
pub parent_capability: Option<CapabilityPolicy>,
|
||||
/// 端到端追踪 ID（从 InboundMessage 继承，用于 tool 执行路径的日志关联）。
|
||||
/// None 表示无追踪上下文（如子代理独立执行或测试环境）。
|
||||
pub trace_id: Option<String>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user