feat: add workflow toolkit and runtime support

This commit is contained in:
2026-07-06 10:45:19 +08:00
parent efc6fa7067
commit ed7f89d44f
44 changed files with 3770 additions and 127 deletions
Generated
+23
View File
@@ -15,13 +15,18 @@ dependencies = [
name = "ai"
version = "0.1.0"
dependencies = [
"anyhow",
"chrono",
"common",
"regex",
"reqwest",
"serde",
"serde_json",
"serde_yaml",
"thiserror",
"tokio",
"tracing",
"tracing-subscriber",
]
[[package]]
@@ -322,8 +327,11 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
name = "common"
version = "0.1.0"
dependencies = [
"anyhow",
"reqwest",
"serde",
"serde_json",
"serde_yaml",
"thiserror",
]
@@ -1374,6 +1382,18 @@ dependencies = [
"bitflags",
]
[[package]]
name = "regex"
version = "1.12.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f1292b7759ae1cb9ec195452d1390a074f0cd8541ab7a5a8c31cd6db45d4a6ba"
dependencies = [
"aho-corasick",
"memchr",
"regex-automata",
"regex-syntax",
]
[[package]]
name = "regex-automata"
version = "0.4.14"
@@ -1602,6 +1622,9 @@ dependencies = [
"sqlx",
"thiserror",
"tokio",
"tracing",
"tracing-subscriber",
"uuid",
"wechat",
]
+8
View File
@@ -15,3 +15,11 @@ serde_json = "1.0" # JSON 处理
serde_yaml = "0.9" # YAML 解析(工具规范文件)
# ── 时间 ──
chrono = "0.4"
# ── 错误处理 ──
anyhow = "1"
thiserror = "2"
# ── 日志管理 ──
tracing = "0.1"
tracing-subscriber = "0.3"
# ── 正则 ──
regex = "1"
+42 -11
View File
@@ -1,4 +1,4 @@
use crate::toolkit::{call_tools, get_tool};
use crate::toolkit::{call_tools, get_tool, tool_notice_text};
use common::model::ai::{
ChatCompletionRequestBody, ChatCompletionRequestExtra, ChatCompletionRequestMessage,
ChatCompletionRequestThinking, ChatCompletionResponseBody,
@@ -9,11 +9,16 @@ pub async fn send_message(
sender: tokio::sync::mpsc::Sender<QueueMessage>,
messages: Vec<ChatCompletionRequestMessage>,
channel: ChannelMessage,
) -> Result<(), reqwest::Error> {
) -> anyhow::Result<()> {
let result = request_llm(messages).await;
match result {
Ok(result) => {
for choice in result.unwrap().choices {
let Some(result) = result else {
eprintln!("LLM 请求没有返回有效响应体");
return Ok(());
};
for choice in result.choices {
let messsage_content = choice.message.content;
let messsage_reasoning = choice.message.reasoning_content.unwrap_or_default();
let tool_calls = choice.message.tool_calls.clone();
@@ -32,11 +37,30 @@ pub async fn send_message(
match choice.message.tool_calls {
Some(tool_calls) => {
println!("调用了{}个工具", tool_calls.len());
let queue_message = call_tools(channel.clone(), tool_calls).await.unwrap();
if let Some(queue_message) = queue_message {
if tool_calls.is_empty() {
eprintln!("模型返回 tool_calls 字段,但列表为空");
continue;
}
println!("模型请求调用{}个工具", tool_calls.len());
if let Some(text) = tool_notice_text(&tool_calls) {
sender
.send(QueueMessage::Notice(channel.clone(), text))
.await
.unwrap();
}
match call_tools(channel.clone(), tool_calls).await {
Ok(Some(queue_message)) => {
sender.send(queue_message).await.unwrap();
}
Ok(None) => {
eprintln!("工具调用没有生成后续队列消息");
}
Err(error) => {
eprintln!("工具调用失败: {error}");
}
}
}
None => {
eprintln!("没有调用工具");
@@ -54,11 +78,13 @@ pub async fn send_message(
pub async fn request_llm(
messages: Vec<ChatCompletionRequestMessage>,
) -> Result<Option<ChatCompletionResponseBody>, reqwest::Error> {
) -> anyhow::Result<Option<ChatCompletionResponseBody>> {
let client = reqwest::Client::new();
let api_key =
std::env::var("ROUTER_API_KEY").map_err(|_| anyhow::anyhow!("必须设置ROUTER_API_KEY"))?;
let body = ChatCompletionRequestBody {
model: "deepseek-v4-flash".to_string(),
messages: messages,
messages,
stream: false,
tools: get_tool(),
extra_body: Some(ChatCompletionRequestExtra {
@@ -69,7 +95,7 @@ pub async fn request_llm(
};
let resp = client
.post("http://100.64.52.162:9807/v1/chat/completions")
.bearer_auth(std::env::var("ROUTER_API_KEY").unwrap())
.bearer_auth(api_key)
.json(&body)
.send()
.await;
@@ -80,8 +106,13 @@ pub async fn request_llm(
let text = resp.text().await.unwrap_or_default();
// println!("response:\n{}", text);
if status.is_success() {
let result = serde_json::from_str::<ChatCompletionResponseBody>(&text).unwrap();
return Ok(Some(result));
match serde_json::from_str::<ChatCompletionResponseBody>(&text) {
Ok(result) => return Ok(Some(result)),
Err(error) => {
eprintln!("LLM 响应解析失败: {error}");
eprintln!("body: {text}");
}
}
} else {
eprintln!("HTTP 请求失败");
eprintln!("status: {}", status);
+2
View File
@@ -1,2 +1,4 @@
pub mod core;
pub mod toolkit;
pub mod tools;
pub mod workflow;
+179 -44
View File
@@ -1,34 +1,113 @@
use chrono::Local;
use serde_json::error;
use crate::{
tools::date_time::get_system_date_time,
tools::web::create_web_page,
workflow::core::{call_flow, load_tools_define},
};
use common::{
model::ai::ChatCompletionToolCall,
model::ai::{ChatCompletionFunction, ChatCompletionToolCall},
queue::{ChannelMessage, QueueMessage},
};
use serde_json::Value;
pub async fn call_tools(
channel: ChannelMessage,
tools: Vec<ChatCompletionToolCall>,
) -> Result<Option<QueueMessage>, error::Error> {
if let Some(tool) = tools.first() {
if let Some(funcation) = tool.function.as_ref() {
if funcation.name == "query_date_time" {
let date_time = Local::now();
let formatted = date_time.format("%Y-%m-%d %H:%M:%S").to_string();
return Ok(Some(QueueMessage::Tool(
) -> anyhow::Result<Option<QueueMessage>> {
let tool_path = tool_path();
if tools.is_empty() {
eprintln!("模型返回了空工具调用列表");
return Ok(None);
}
if tools.len() > 1 {
eprintln!("模型一次返回了{}个工具调用,当前仅处理第一个", tools.len());
}
let Some(tool) = tools.first() else {
return Ok(None);
};
let (tool_name, tool_result) = match tool.function.as_ref() {
Some(function) => {
println!(
"开始调用工具: {}, arguments={}",
function.name, function.arguments
);
(
function.name.clone(),
execute_tool(&tool_path, function).await,
)
}
None => {
eprintln!("工具调用缺少 function 字段: id={}", tool.id);
(
"unknown".to_string(),
"工具调用失败: tool call 缺少 function 字段".to_string(),
)
}
};
let tool_result = if tool_result.trim().is_empty() {
format!("工具调用失败: {tool_name} 返回空结果")
} else {
tool_result
};
println!(
"工具调用完成: {}, result_len={}",
tool_name,
tool_result.len()
);
Ok(Some(QueueMessage::Tool(
channel,
tool.id.clone(),
funcation.name.clone(),
format!("现在时间:{}", formatted),
)));
tool_name,
tool_result,
)))
}
async fn execute_tool(tool_path: &str, function: &ChatCompletionFunction) -> String {
match function.name.as_str() {
"query_date_time" => match get_system_date_time() {
Ok(value) => value,
Err(error) => format!("工具调用失败: {error}"),
},
"query_capabilities" => match load_tools_define(tool_path.to_string()).await {
Ok(value) => value,
Err(error) => format!("工具调用失败: 加载工具定义失败: {error}"),
},
"create_web_page" => create_web_page(&function.arguments).await,
"call_capability" => call_flow(tool_path.to_string(), function.arguments.clone()).await,
other => call_named_capability(tool_path, other, &function.arguments).await,
}
}
}
Ok(None)
}
async fn call_named_capability(tool_path: &str, name: &str, arguments: &str) -> String {
let prompt: Value = match serde_json::from_str(arguments) {
Ok(value) => value,
Err(error) => return format!("工具调用失败: 解析 {name} 参数失败: {error}"),
};
let args = serde_json::json!({
"name": name,
"prompt": prompt,
});
call_flow(tool_path.to_string(), args.to_string()).await
}
pub fn tool_notice_text(tools: &[ChatCompletionToolCall]) -> Option<String> {
let function = tools.first()?.function.as_ref()?;
let args: Value = serde_json::from_str(&function.arguments).ok()?;
args.get("text")
.and_then(Value::as_str)
.map(str::trim)
.filter(|text| !text.is_empty())
.map(ToString::to_string)
}
pub fn get_tool() -> Vec<serde_json::Value> {
return vec![
vec![
serde_json::json!({
"type": "function",
"function": {
@@ -37,31 +116,87 @@ pub fn get_tool() -> Vec<serde_json::Value> {
"parameters": {"type": "object", "properties": {}, "required": []}
}
}),
// serde_json::json!({
// "type": "function",
// "function": {
// "name": "query_capabilities",
// "description": "查询所有可用工具的列表、参数格式和用法说明",
// "parameters": {"type": "object", "properties": {}, "required": []}
// }
// }),
// serde_json::json!({
// "type": "function",
// "function": {
// "name": "call_capability",
// "description": "调用指定的工具。先通过 query_capabilities 获取工具名和参数格式",
// "parameters": {
// "type": "object",
// "properties": {
// "name": {"type": "string", "description": "工具名称"},
// "prompt": {
// "type": "object",
// "description": "工具参数对象,如 {\"location\":\"北京\"}"
// }
// },
// "required": ["name", "prompt"]
// }
// }
// }),
];
serde_json::json!({
"type": "function",
"function": {
"name": "query_capabilities",
"description": "查询所有可用工具的列表、参数格式和用法说明",
"parameters": {"type": "object", "properties": {}, "required": []}
}
}),
serde_json::json!({
"type": "function",
"function": {
"name": "create_web_page",
"description": "创建一个可通过链接访问的动态内容页面,用于把较长或结构复杂的内容交给用户查看。邮件提醒请使用 page_type=mail,工具结果会返回 /mail/{slug} 链接。",
"parameters": {
"type": "object",
"properties": {
"page_type": {
"type": "string",
"description": "页面类型,只能包含字母、数字、下划线或连字符。普通内容可用 page,邮件内容用 mail。"
},
"title": {
"type": "string",
"description": "页面标题,例如邮件主题或内容摘要标题。"
},
"summary": {
"type": "string",
"description": "简短摘要,会显示在页面标题下方,也可用于最终回复用户。"
},
"content": {
"type": "string",
"description": "页面正文。可以放较长内容,系统会按纯文本安全渲染并保留换行。"
},
"source_label": {
"type": "string",
"description": "内容来源,例如发件人、系统名或服务名。"
},
"metadata": {
"type": "object",
"description": "附加结构化信息,例如 {\"from\":\"xxx\",\"subject\":\"xxx\",\"received_at\":\"2026-07-06 09:00\"}。"
},
"text": {
"type": "string",
"description": "可选。调用工具前发送给用户的提醒消息,用于说明接下来会创建详情页面;为空则不发送。"
}
},
"required": ["title", "content"]
}
}
}),
serde_json::json!({
"type": "function",
"function": {
"name": "call_capability",
"description": "调用指定的工具。先通过 query_capabilities 获取工具名和参数格式",
"parameters": {
"type": "object",
"properties": {
"name": {"type": "string", "description": "工具名称"},
"prompt": {
"type": "object",
"description": "工具参数对象,如 {\"location\":\"北京\"}"
},
"text": {
"type": "string",
"description": "可选。调用工具前发送给用户的提醒消息,用于说明接下来会做什么;为空则不发送"
}
},
"required": ["name", "prompt"]
}
}
}),
]
}
fn tool_path() -> String {
std::env::var("IA_TOOLS_PATH").unwrap_or_else(|_| {
std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.parent()
.unwrap_or_else(|| std::path::Path::new("."))
.join("tools")
.display()
.to_string()
})
}
+7
View File
@@ -0,0 +1,7 @@
use chrono::Local;
pub fn get_system_date_time() -> anyhow::Result<String> {
let date_time = Local::now();
let formatted = date_time.format("%Y-%m-%d %H:%M:%S").to_string();
Ok(format!("现在时间:{}", formatted))
}
+381
View File
@@ -0,0 +1,381 @@
use std::collections::HashMap;
use std::process::Stdio;
use anyhow::{Context, Result, anyhow};
use serde::Serialize;
use tokio::io::AsyncWriteExt;
use tokio::process::Command;
/// 整体超时默认值(秒)
const DEFAULT_TIMEOUT: u64 = 30;
// ────────────────────────────── 请求基座 ──────────────────────────────
/// HTTP 请求基座:封装 curl + jq 的执行细节,供各类 HTTP 工具复用
///
/// 所有参数以 `Vec<String>` 直接传给 `curl` 子进程,**不经过 shell**
/// 因此 URL / Header / Body 中包含空格、引号、`$`、`;` 等特殊字符也不会被解释,
/// 天然规避命令注入。
///
/// 采用 builder 风格链式配置,后续具体工具基于此基座实现,例如:
/// ```ignore
/// // 一个「查天气」工具,复用基座
/// pub async fn get_weather(city: &str, token: &str) -> Result<f64> {
/// let resp = HttpRequest::new(format!("https://api.example.com/weather?q={city}"))
/// .header("Authorization", format!("Bearer {token}"))
/// .jq(".current.temp_c")
/// .execute()
/// .await?;
/// Ok(resp.json()?.as_f64().unwrap_or(0.0))
/// }
/// ```
pub struct HttpRequest {
/// 请求地址
pub url: String,
/// HTTP 方法
pub method: String,
/// 请求头
pub headers: HashMap<String, String>,
/// 请求体
pub body: Option<String>,
/// jq 过滤表达式(为空则不过滤)
pub jq: Option<String>,
/// 整体超时(秒)
pub timeout: u64,
}
impl HttpRequest {
/// 以指定 URL 创建一个 GET 请求
pub fn new(url: impl Into<String>) -> Self {
Self {
url: url.into(),
method: "GET".to_string(),
headers: HashMap::new(),
body: None,
jq: None,
timeout: DEFAULT_TIMEOUT,
}
}
/// 设置 HTTP 方法(自动转大写,便于容忍小写输入)
pub fn method(mut self, method: impl Into<String>) -> Self {
self.method = method.into().to_uppercase();
self
}
/// 追加一个请求头(同名键会覆盖)
pub fn header(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
self.headers.insert(key.into(), value.into());
self
}
/// 设置原始请求体(不自动设置 Content-Type
pub fn body(mut self, body: impl Into<String>) -> Self {
self.body = Some(body.into());
self
}
/// 设置 JSON 请求体,并自动加上 `Content-Type: application/json`
pub fn json_body(mut self, value: &serde_json::Value) -> Self {
self.body = Some(serde_json::to_string(value).unwrap_or_default());
self.headers
.insert("Content-Type".into(), "application/json".into());
self
}
/// 设置 jq 过滤表达式,对 JSON 响应体做提取/加工
pub fn jq(mut self, filter: impl Into<String>) -> Self {
self.jq = Some(filter.into());
self
}
/// 设置整体超时(秒)
pub fn timeout(mut self, secs: u64) -> Self {
self.timeout = secs;
self
}
/// 执行请求:先跑 curl,再用 jq(可选)过滤响应体
pub async fn execute(&self) -> Result<HttpResponse> {
let (status, body) = self.run_curl().await?;
// 可选:用 jq 过滤响应体。jq 失败不致命——保留原文并附带错误
let (body, jq_error) = match self.jq.as_deref() {
Some(filter) if !filter.trim().is_empty() => run_jq(filter, &body).await?,
_ => (body, None),
};
Ok(HttpResponse {
status,
body,
jq_error,
})
}
/// 调用 curl 并解析出「状态码 + 响应体」
///
/// 利用 `curl -w '\n%{http_code}'` 在响应体末尾追加一行状态码,
/// 随后以最后一个换行为界拆分:之前是 body,之后是 status。
///
/// 注意:HTTP 4xx/5xx 不是错误,会原样返回状态码与响应体;
/// 只有 curl 自身失败(连接超时、DNS 解析失败等)才返回 `Err`。
async fn run_curl(&self) -> Result<(u16, String)> {
let method = self.method.to_uppercase();
// 组装 curl 参数
let mut args: Vec<String> = vec![
"-s".into(), // 静默进度条
"-S".into(), // 出错时仍打印错误到 stderr
"-L".into(), // 跟随重定向
"--compressed".into(), // 自动解压 gzip/br/zstd
"--connect-timeout".into(),
"10".into(),
"--max-time".into(),
self.timeout.to_string(),
"-X".into(),
method,
// 末尾追加一行状态码,用换行与响应体分隔
"-w".into(),
"\n%{http_code}".into(),
];
for (k, v) in &self.headers {
args.push("-H".into());
args.push(format!("{k}: {v}"));
}
if let Some(body) = &self.body {
// --data-raw 不会把开头 @ 当作文件读取,比 -d 更安全
args.push("--data-raw".into());
args.push(body.clone());
}
args.push(self.url.clone());
// 执行 curl
let output = Command::new("curl")
.args(&args)
.output()
.await
.with_context(|| format!("执行 curl 失败: {}", self.url))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
return Err(anyhow!("curl 请求失败: {stderr}"));
}
// 拆分响应体与状态码(最后一行即状态码)
let stdout = String::from_utf8_lossy(&output.stdout).into_owned();
let (body, status) = match stdout.rfind('\n') {
Some(i) => {
let status: u16 = stdout[i + 1..]
.trim()
.parse()
.context("解析 HTTP 状态码失败")?;
(stdout[..i].to_string(), status)
}
None => (String::new(), 0u16),
};
Ok((status, body))
}
}
// ────────────────────────────── 响应 ──────────────────────────────
/// HTTP 请求结果
#[derive(Debug, Serialize)]
pub struct HttpResponse {
/// HTTP 状态码;传输层失败时为 0
pub status: u16,
/// 响应体(若指定了 jq 则为过滤后的内容)
pub body: String,
/// jq 执行失败时的错误信息(非致命,原响应体仍会保留在 body 中)
#[serde(skip_serializing_if = "Option::is_none")]
pub jq_error: Option<String>,
}
impl HttpResponse {
/// 是否为 2xx 成功响应
pub fn is_success(&self) -> bool {
(200..300).contains(&self.status)
}
/// 将响应体解析为 JSON(若指定了 jq,则解析过滤后的结果)
pub fn json(&self) -> Result<serde_json::Value> {
serde_json::from_str(&self.body).context("响应体不是合法 JSON")
}
}
// ────────────────────────────── jq 过滤 ──────────────────────────────
/// 用 jq 对 `body` 做过滤
///
/// 成功返回 `(过滤结果, None)`;失败(非 JSON 或表达式错误)则保留原文,
/// 返回 `(原文, Some(错误信息))`,避免因 jq 问题丢失原始响应。
async fn run_jq(filter: &str, body: &str) -> Result<(String, Option<String>)> {
let mut child = Command::new("jq")
.arg(filter)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.context("启动 jq 失败")?;
if let Some(mut stdin) = child.stdin.take() {
// 忽略写入错误:响应体为空时 jq 会自行报错
let _ = stdin.write_all(body.as_bytes()).await;
}
let output = child.wait_with_output().await.context("等待 jq 结束失败")?;
if output.status.success() {
Ok((String::from_utf8_lossy(&output.stdout).into_owned(), None))
} else {
let err = String::from_utf8_lossy(&output.stderr).trim().to_string();
Ok((body.to_string(), Some(err)))
}
}
// ────────────────────────────── 测试 ──────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
/// 读取一条完整的 HTTP 请求(含 body)
fn read_request(stream: &mut TcpStream) -> String {
let mut buf = Vec::new();
let mut tmp = [0u8; 4096];
loop {
let n = stream.read(&mut tmp).unwrap_or(0);
if n == 0 {
break;
}
buf.extend_from_slice(&tmp[..n]);
let s = String::from_utf8_lossy(&buf);
if let Some(split) = s.find("\r\n\r\n") {
let headers = &s[..split];
let body_start = split + 4;
let clen = headers
.lines()
.find_map(|l| {
let l = l.to_lowercase();
l.strip_prefix("content-length: ")
.and_then(|v| v.parse::<usize>().ok())
})
.unwrap_or(0);
if buf.len() >= body_start + clen {
break;
}
}
if buf.len() > 65536 {
break;
}
}
String::from_utf8_lossy(&buf).into_owned()
}
/// 启动一个一次性本地 HTTP 服务:把完整原始请求传给 `respond`,回写其返回值
fn spawn_server<F>(respond: F) -> String
where
F: FnOnce(&str) -> (u16, String) + Send + 'static,
{
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
std::thread::spawn(move || {
if let Ok((mut stream, _)) = listener.accept() {
let req = read_request(&mut stream);
let (status, resp_body) = respond(&req);
let reason = match status {
200 => "OK",
404 => "Not Found",
_ => "OK",
};
let resp = format!(
"HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{resp_body}",
resp_body.len()
);
let _ = stream.write_all(resp.as_bytes());
let _ = stream.flush();
}
});
format!("http://{addr}")
}
#[tokio::test]
async fn builder_get_with_jq() {
let url = spawn_server(|_| (200, r#"{"origin":"1.2.3.4"}"#.to_string()));
let resp = HttpRequest::new(url).jq(".origin").execute().await.unwrap();
assert!(resp.is_success());
assert!(resp.jq_error.is_none(), "jq 不应报错: {:?}", resp.jq_error);
assert_eq!(resp.body.trim(), "\"1.2.3.4\"");
}
#[tokio::test]
async fn builder_post_body_roundtrip() {
// 服务把收到的请求体原样回显到 .echo
let url = spawn_server(|req| {
let body = req.split("\r\n\r\n").nth(1).unwrap_or("");
(200, format!("{{\"echo\":{}}}", body))
});
let resp = HttpRequest::new(url)
.method("post") // 小写也应被接受
.header("Content-Type", "application/json")
.body(r#"{"hello":"world"}"#)
.jq(".echo.hello")
.execute()
.await
.unwrap();
assert_eq!(resp.body.trim(), "\"world\"");
}
#[tokio::test]
async fn json_body_sets_content_type() {
// 服务回显是否带 application/json 头,以及请求体内容
let url = spawn_server(|req| {
let has_ct = req
.lines()
.any(|l| l.eq_ignore_ascii_case("content-type: application/json"));
let body = req.split("\r\n\r\n").nth(1).unwrap_or("");
(200, format!("{{\"ct\":{},\"echo\":{}}}", has_ct, body))
});
let resp = HttpRequest::new(url)
.method("POST")
.json_body(&serde_json::json!({"hello": "world"}))
.jq(".ct")
.execute()
.await
.unwrap();
assert_eq!(resp.body.trim(), "true");
}
#[tokio::test]
async fn bad_jq_keeps_body() {
let url = spawn_server(|_| (200, r#"{"data":"hello"}"#.to_string()));
let resp = HttpRequest::new(url).jq(".[").execute().await.unwrap(); // 非法表达式
assert!(resp.jq_error.is_some(), "应附带 jq 错误");
assert!(
resp.body.contains("hello"),
"应保留原始响应体: {}",
resp.body
);
}
#[tokio::test]
async fn http_404_is_result_not_error() {
let url = spawn_server(|_| (404, r#"{"err":"nope"}"#.to_string()));
let resp = HttpRequest::new(url).execute().await.unwrap();
assert_eq!(resp.status, 404);
assert!(!resp.is_success());
assert!(resp.body.contains("nope"));
}
#[tokio::test]
async fn response_json_parses() {
let url = spawn_server(|_| (200, r#"{"a":1,"b":[2,3]}"#.to_string()));
let resp = HttpRequest::new(url).execute().await.unwrap();
let v = resp.json().unwrap();
assert_eq!(v["a"], 1);
assert_eq!(v["b"][1], 3);
}
}
+3
View File
@@ -0,0 +1,3 @@
pub mod date_time;
pub mod http;
pub mod web;
+95
View File
@@ -0,0 +1,95 @@
use anyhow::{Context, anyhow};
use serde::Deserialize;
use serde_json::{Value, json};
#[derive(Debug, Deserialize)]
struct CreateWebPageArgs {
page_type: Option<String>,
title: String,
summary: Option<String>,
content: String,
source_label: Option<String>,
metadata: Option<Value>,
#[allow(dead_code)]
text: Option<String>,
}
#[derive(Debug, Deserialize)]
struct CreateWebPageResponse {
success: bool,
title: String,
page_type: String,
summary: Option<String>,
path: String,
url: String,
}
pub async fn create_web_page(arguments: &str) -> String {
match create_web_page_inner(arguments).await {
Ok(result) => result,
Err(error) => format!("工具调用失败: 创建 web 页面失败: {error}"),
}
}
async fn create_web_page_inner(arguments: &str) -> anyhow::Result<String> {
let args: CreateWebPageArgs =
serde_json::from_str(arguments).context("解析 create_web_page 参数失败")?;
if args.title.trim().is_empty() {
return Err(anyhow!("title 不能为空"));
}
if args.content.trim().is_empty() {
return Err(anyhow!("content 不能为空"));
}
let payload = json!({
"page_type": args.page_type,
"title": args.title,
"summary": args.summary,
"content": args.content,
"source_label": args.source_label,
"metadata": args.metadata,
});
let base_url = web_base_url();
let endpoint = format!("{base_url}/web/pages");
let client = reqwest::Client::new();
let response = client
.post(&endpoint)
.json(&payload)
.send()
.await
.with_context(|| format!("请求 web 页面接口失败: {endpoint}"))?;
let status = response.status();
let body = response.text().await.unwrap_or_default();
if !status.is_success() {
return Err(anyhow!("接口返回失败: status={status}, body={body}"));
}
let result: CreateWebPageResponse =
serde_json::from_str(&body).context("解析 web 页面接口响应失败")?;
if !result.success {
return Err(anyhow!("接口未成功创建页面: {body}"));
}
let summary = result
.summary
.as_deref()
.filter(|summary| !summary.trim().is_empty())
.map(|summary| format!("\n摘要:{}", summary.trim()))
.unwrap_or_default();
Ok(format!(
"web 页面已创建:{}\n路径:{}\n类型:{}\n标题:{}{}",
result.url, result.path, result.page_type, result.title, summary
))
}
fn web_base_url() -> String {
std::env::var("IA_WEB_BASE_URL")
.or_else(|_| std::env::var("WEB_BASE_URL"))
.or_else(|_| std::env::var("HOST").map(|host| format!("http://{host}")))
.unwrap_or_else(|_| "http://127.0.0.1:9003".to_string())
.trim()
.trim_end_matches('/')
.replace("0.0.0.0", "127.0.0.1")
}
+73
View File
@@ -0,0 +1,73 @@
use serde_json::Value;
use tokio::fs;
use crate::workflow::flow::start_flow;
use crate::workflow::types::FlowConfig;
pub async fn call_flow(path: String, args: String) -> String {
let all_flow = match load_configs(path).await {
Ok(flows) => flows,
Err(err) => return format!("工具调用失败:{:?}", err),
};
let args_json: Value = match serde_json::from_str(&args) {
Ok(json) => json,
Err(err) => return format!("工具调用失败:{:?}", err),
};
let flow_name = args_json["name"].as_str().unwrap_or_default();
if let Some(config) = all_flow.iter().find(|e| flow_name == e.name) {
let flow_result = start_flow(config, args_json["prompt"].clone()).await;
match flow_result {
Ok(value) => value,
Err(err) => format!("工具调用失败:{:?}", err),
}
} else {
"工具不存在,请检查工具名称".to_string()
}
}
pub async fn load_tools_define(path: String) -> anyhow::Result<String> {
let configs = load_configs(path).await?;
Ok(serde_json::to_string(&configs).unwrap_or_default())
}
pub async fn load_configs(path: String) -> anyhow::Result<Vec<FlowConfig>> {
let mut configs: Vec<FlowConfig> = vec![];
let config_files = scan_config_dir(path).await?;
if !config_files.is_empty() {
for file in config_files {
let config = load_yaml(file).await?;
configs.push(config);
}
}
Ok(configs)
}
pub async fn load_yaml(path: String) -> anyhow::Result<FlowConfig> {
let content: String = fs::read_to_string(path).await?;
let config = serde_yaml::from_str(&content)?;
Ok(config)
}
pub async fn scan_config_dir(path: String) -> anyhow::Result<Vec<String>> {
let mut dir = fs::read_dir(path).await?;
let mut files: Vec<String> = vec![];
while let Some(entry) = dir.next_entry().await? {
let path = entry.path();
let abs = fs::canonicalize(&path).await?;
files.push(abs.display().to_string());
}
Ok(files)
}
#[tokio::test]
async fn list_file() {
println!();
let tools = load_tools_define("../tools".to_string()).await.unwrap();
println!("{}", tools);
println!();
}
+510
View File
@@ -0,0 +1,510 @@
use std::str::FromStr;
use std::{
collections::HashMap,
time::{Duration, Instant},
};
use reqwest::{
Client, Method,
header::{HeaderMap, HeaderName, HeaderValue},
};
use serde_json::Value;
use thiserror::Error;
use tracing::{debug, error, info, warn};
use crate::workflow::resolve::resolve_extract;
use crate::workflow::types::StepResult;
use crate::workflow::{
template::render_template,
types::{FlowConfig, FlowStep},
};
#[derive(Debug, Error)]
pub enum FlowError {
#[error("{name}字段解析失败")]
Abort { name: String },
#[error("跳转到第{index}步")]
Jump { index: usize },
#[error("执行任务出现错误:{message}")]
Other { message: String },
}
const DEFAULT_TIMEOUT: u64 = 30;
pub async fn start_flow(config: &FlowConfig, args: Value) -> Result<String, FlowError> {
info!(
target: "ias::flow",
flow = %config.name,
param_count = config.params.len(),
step_count = config.steps.len(),
"flow 开始"
);
let arg_keys = args
.as_object()
.map(|values| values.keys().cloned().collect::<Vec<_>>())
.unwrap_or_default();
debug!(target: "ias::flow", flow = %config.name, ?arg_keys, "flow 入参键");
let mut params: HashMap<String, String> = HashMap::new();
for param in config.params.clone() {
let name = param.name;
if !name.is_empty() {
if let Some(value) = args.get(name.clone()) {
params.insert(name, value_to_string(value));
} else if param.required {
if let Some(default) = param.default {
params.insert(name, default);
} else {
warn!(
target: "ias::flow",
flow = %config.name,
param = %name,
"flow 缺少必填参数"
);
return Ok(format!("参数{}为必传", name));
}
} else if let Some(default) = param.default {
params.insert(name, default);
}
}
}
let param_keys = params.keys().cloned().collect::<Vec<_>>();
debug!(target: "ias::flow", flow = %config.name, ?param_keys, "flow 参数归一化完成");
let mut flow_result: String = "工具错误,没有正确处理输出".to_string();
let mut step_results: HashMap<String, StepResult> = HashMap::new();
let mut cache: HashMap<String, StepResult> = HashMap::new();
let mut index = 0;
let steps = config.steps.clone();
while index < steps.len() {
if let Some(now_step) = steps.get(index) {
info!(target: "ias::flow", flow = %config.name, step_index = index, "flow 步骤开始");
let result = match deal_steps(index, now_step, &params, &step_results, &mut cache).await
{
Ok(result) => result,
Err(e) => match e {
FlowError::Jump { index: jump } => {
warn!(
target: "ias::flow",
flow = %config.name,
step_index = index,
jump_to = jump,
"flow 步骤触发跳转"
);
if jump >= steps.len() {
error!(
target: "ias::flow",
flow = %config.name,
step_index = index,
jump_to = jump,
step_count = steps.len(),
"flow 跳转目标不存在"
);
return get_other_error(format!("跳转步骤{}不存在", jump));
}
index = jump;
continue;
}
FlowError::Abort { name } => {
warn!(
target: "ias::flow",
flow = %config.name,
step_index = index,
param = %name,
"flow 步骤中止"
);
return get_other_error(format!("参数{}为必传", name));
}
FlowError::Other { message } => {
error!(
target: "ias::flow",
flow = %config.name,
step_index = index,
error = %message,
"flow 步骤执行失败"
);
return get_other_error(message);
}
},
};
info!(
target: "ias::flow",
flow = %config.name,
step_index = index,
raw_len = result.raw.len(),
var_count = result.var.len(),
"flow 步骤完成"
);
flow_result = result.raw.clone();
step_results.insert(format!("step{}", index), result);
if !now_step.result.is_empty() {
flow_result = match render_template(&now_step.result, &params, &step_results) {
Ok(result) => {
info!(
target: "ias::flow",
flow = %config.name,
step_index = index,
result_len = result.len(),
"flow 命中结果模板,结束"
);
result
}
Err(e) => {
error!(
target: "ias::flow",
flow = %config.name,
step_index = index,
error = %e,
"flow 结果模板渲染失败"
);
return get_other_error(e.to_string());
}
};
return Ok(flow_result);
}
if let Some(next) = now_step.next {
if next >= steps.len() {
error!(
target: "ias::flow",
flow = %config.name,
step_index = index,
next_step = next,
step_count = steps.len(),
"flow next 目标不存在"
);
return get_other_error(format!("跳转步骤{}不存在", next));
}
info!(
target: "ias::flow",
flow = %config.name,
step_index = index,
next_step = next,
"flow 使用显式下一步"
);
index = next;
} else {
index += 1;
}
};
}
info!(
target: "ias::flow",
flow = %config.name,
result_len = flow_result.len(),
"flow 完成"
);
Ok(flow_result)
}
async fn deal_steps(
step_index: usize,
step: &FlowStep,
params: &HashMap<String, String>,
step_results: &HashMap<String, StepResult>,
cache: &mut HashMap<String, StepResult>,
) -> Result<StepResult, FlowError> {
let template_step = deal_template(step_index, step, params, step_results).await?;
deal_requst(step_index, &template_step, cache).await
}
async fn deal_template(
step_index: usize,
step: &FlowStep,
params: &HashMap<String, String>,
step_results: &HashMap<String, StepResult>,
) -> Result<FlowStep, FlowError> {
let json = match serde_json::to_string(step) {
Ok(json) => Ok(json),
Err(e) => {
error!(target: "ias::flow", step_index, error = %e, "flow 步骤序列化失败");
get_other_error(e.to_string())
}
}?;
let template_json = match render_template(&json, params, step_results) {
Ok(json) => Ok(json),
Err(e) => {
error!(target: "ias::flow", step_index, error = %e, "flow 步骤模板渲染失败");
get_other_error(e.to_string())
}
}?;
match serde_json::from_str(&template_json) {
Ok(flow) => Ok(flow),
Err(e) => {
error!(target: "ias::flow", step_index, error = %e, "flow 步骤模板解析失败");
get_other_error(e.to_string())
}
}
}
async fn deal_requst(
step_index: usize,
step: &FlowStep,
cache: &mut HashMap<String, StepResult>,
) -> Result<StepResult, FlowError> {
let url = step.url.clone();
if url.is_empty() || !url.starts_with("http") {
warn!(target: "ias::flow", step_index, url = %url, "flow 步骤 URL 无效");
return Ok(StepResult {
var: HashMap::new(),
raw: "请输入正确的url".to_string(),
});
}
let method = deal_method(step.method.clone());
let method_name = method.as_str().to_string();
let should_send_body = method != Method::GET;
let cache_key = step.cache.unwrap_or(false).then(|| build_cache_key(step));
if let Some(key) = &cache_key
&& let Some(result) = cache.get(key)
{
info!(
target: "ias::flow",
step_index,
cache_key_len = key.len(),
raw_len = result.raw.len(),
var_count = result.var.len(),
"flow 缓存命中"
);
return Ok(result.clone());
}
let client = Client::new();
let mut request = client.request(method, url.clone());
request = request.query(&step.query.clone());
let headers = match deal_headers(step.headers.clone()) {
Ok(headers) => headers,
Err(e) => {
error!(target: "ias::flow", step_index, error = %e, "flow 请求头解析失败");
return Err(e);
}
};
request = request.headers(headers);
if should_send_body && let Some(body) = step.body.clone() {
request = request.body(body);
}
let timeout = step.timeout.unwrap_or(DEFAULT_TIMEOUT);
request = request.timeout(Duration::from_secs(timeout));
let query_count = step.query.as_ref().map_or(0, Vec::len);
let header_count = step.headers.as_ref().map_or(0, HashMap::len);
let body_len = step.body.as_ref().map_or(0, String::len);
info!(
target: "ias::flow",
step_index,
method = %method_name,
url = %url,
timeout_secs = timeout,
query_count,
header_count,
body_len,
"flow HTTP 请求开始"
);
let started_at = Instant::now();
let raw = match request.send().await {
Ok(response) => {
let status = response.status();
match response.text().await {
Ok(raw) => {
info!(
target: "ias::flow",
step_index,
status = %status,
elapsed_ms = started_at.elapsed().as_millis(),
response_len = raw.len(),
"flow HTTP 响应完成"
);
raw
}
Err(e) => {
error!(
target: "ias::flow",
step_index,
status = %status,
elapsed_ms = started_at.elapsed().as_millis(),
error = %e,
"flow HTTP 响应读取失败"
);
return get_other_error(e.to_string());
}
}
}
Err(e) => {
error!(
target: "ias::flow",
step_index,
elapsed_ms = started_at.elapsed().as_millis(),
error = %e,
"flow HTTP 请求失败"
);
return get_other_error(e.to_string());
}
};
let vars = deal_var(step_index, step, &raw)?;
let result = StepResult { var: vars, raw };
if let Some(key) = cache_key {
debug!(
target: "ias::flow",
step_index,
cache_key_len = key.len(),
"flow 写入缓存"
);
cache.insert(key, result.clone());
}
Ok(result)
}
fn deal_var(
step_index: usize,
step: &FlowStep,
raw: &str,
) -> Result<HashMap<String, String>, FlowError> {
let mut vars: HashMap<String, String> = HashMap::new();
if !step.var.is_empty() {
debug!(
target: "ias::flow",
step_index,
var_count = step.var.len(),
response_len = raw.len(),
"flow 开始提取变量"
);
for var in step.var.clone() {
let mut value = String::new();
let name = var.name;
if let Some(expr) = &var.extract {
match resolve_extract(raw, expr) {
Ok(Some(v)) => {
debug!(
target: "ias::flow",
step_index,
var = %name,
value_len = v.len(),
"flow 变量提取成功"
);
value = v;
}
Ok(None) | Err(_) => {
warn!(
target: "ias::flow",
step_index,
var = %name,
extract = %expr,
"flow 变量提取为空或失败"
);
if let Some(abort) = var.abort
&& abort
{
warn!(
target: "ias::flow",
step_index,
var = %name,
"flow 变量提取失败,触发中止"
);
return Err(FlowError::Abort { name });
}
if let Some(jump) = var.jump {
warn!(
target: "ias::flow",
step_index,
var = %name,
jump_to = jump,
"flow 变量提取失败,触发跳转"
);
return Err(FlowError::Jump { index: jump });
}
if let Some(default) = &var.default {
debug!(
target: "ias::flow",
step_index,
var = %name,
default_len = default.len(),
"flow 使用变量默认值"
);
value = default.clone();
}
}
}
}
if !value.is_empty() {
debug!(
target: "ias::flow",
step_index,
var = %name,
value_len = value.len(),
"flow 写入变量"
);
vars.insert(name, value);
} else {
debug!(
target: "ias::flow",
step_index,
var = %name,
"flow 变量为空,跳过写入"
);
}
}
}
Ok(vars)
}
fn deal_method(origin: Option<String>) -> Method {
let method_str = match origin {
Some(method) => method.to_uppercase(),
None => "GET".to_string(),
};
match method_str.as_str() {
"POST" => Method::POST,
"PUT" => Method::PUT,
"PATCH" => Method::PATCH,
"DELETE" => Method::DELETE,
_ => Method::GET,
}
}
fn deal_headers(origin: Option<HashMap<String, String>>) -> Result<HeaderMap, FlowError> {
let headers = origin.unwrap_or_default();
let mut result = HeaderMap::new();
for (k, v) in headers {
let name = match HeaderName::from_str(&k) {
Ok(name) => name,
Err(_) => return get_other_error(format!("请求头{}无效", k)),
};
let value = match HeaderValue::from_str(&v) {
Ok(value) => value,
Err(_) => return get_other_error(format!("请求头{}的值无效", k)),
};
result.append(name, value);
}
Ok(result)
}
fn build_cache_key(step: &FlowStep) -> String {
let mut headers: Vec<(String, String)> = step
.headers
.clone()
.unwrap_or_default()
.into_iter()
.collect();
headers.sort_by(|a, b| a.0.cmp(&b.0).then(a.1.cmp(&b.1)));
format!(
"{}|{:?}|{:?}|{:?}|{}",
step.url,
step.method,
headers,
step.query,
step.body.as_deref().unwrap_or("")
)
}
fn get_other_error<T>(message: String) -> Result<T, FlowError> {
Err(FlowError::Other { message })
}
fn value_to_string(value: &Value) -> String {
match value {
Value::String(value) => value.clone(),
Value::Number(value) => value.to_string(),
Value::Bool(value) => value.to_string(),
Value::Null => String::new(),
_ => value.to_string(),
}
}
+5
View File
@@ -0,0 +1,5 @@
pub mod core;
pub mod flow;
pub mod resolve;
pub mod template;
pub mod types;
+168
View File
@@ -0,0 +1,168 @@
use regex::Regex;
use serde_json::Value;
enum Segment {
Key(String),
Index(usize),
All,
}
/// a.b / a.b[0].c / a.b[].c → Vec<Segment>
fn parse_path(expr: &str) -> Result<Vec<Segment>, String> {
if expr.is_empty() {
return Ok(vec![]);
}
let mut segments = Vec::new();
for part in expr.split('.') {
if part.is_empty() {
return Err(format!("路径 \"{}\" 中存在空段", expr));
}
if let Some(key) = part.strip_suffix("[]") {
if key.is_empty() {
return Err(format!("路径 \"{}\"\"[]\" 前缺少键名", expr));
}
segments.push(Segment::Key(key.to_string()));
segments.push(Segment::All);
} else if let Some(b) = part.find('[') {
let key = &part[..b];
let inner = &part[b + 1..];
let inner = inner
.strip_suffix(']')
.ok_or_else(|| format!("路径 \"{}\"\"{}\" 缺少闭合 ']'", expr, part))?;
if key.is_empty() {
return Err(format!(
"路径 \"{}\" 中索引 \"[{}\" 前缺少键名",
expr, inner
));
}
let idx: usize = inner
.parse()
.map_err(|_| format!("路径 \"{}\" 中索引 \"{}\" 不是数字", expr, inner))?;
segments.push(Segment::Key(key.to_string()));
segments.push(Segment::Index(idx));
} else {
segments.push(Segment::Key(part.to_string()));
}
}
Ok(segments)
}
fn resolve_path(value: &Value, segments: &[Segment]) -> Option<Value> {
if segments.is_empty() {
return Some(value.clone());
}
match &segments[0] {
Segment::Key(k) => value
.as_object()?
.get(k)
.and_then(|v| resolve_path(v, &segments[1..])),
Segment::Index(i) => value
.as_array()?
.get(*i)
.and_then(|v| resolve_path(v, &segments[1..])),
Segment::All => {
let arr = value.as_array()?;
let results: Vec<Value> = arr
.iter()
.filter_map(|item| resolve_path(item, &segments[1..]))
.collect();
Some(Value::Array(results))
}
}
}
fn value_to_string(v: &Value) -> String {
match v {
Value::String(s) => s.clone(),
Value::Number(n) => n.to_string(),
Value::Bool(b) => b.to_string(),
Value::Null => String::new(),
_ => v.to_string(),
}
}
fn detect_and_parse(raw: &str) -> Option<Value> {
serde_json::from_str(raw)
.ok()
.or_else(|| serde_yaml::from_str(raw).ok())
}
/// 统一提取入口:/pattern/ → 正则引擎,否则 → 路径引擎
pub fn resolve_extract(raw: &str, expr: &str) -> Result<Option<String>, String> {
if let Some(inner) = expr.strip_prefix('/').and_then(|s| s.strip_suffix('/')) {
let re = Regex::new(inner).map_err(|e| format!("正则语法错误: {}", e))?; // 编译正则
Ok(re
.captures(raw)
.and_then(|caps| caps.get(1).map(|m| m.as_str().to_string()))) // 取捕获组1
} else {
let value =
detect_and_parse(raw).ok_or_else(|| "响应体无法解析为 JSON 或 YAML".to_string())?; // JSON → YAML 自动检测
let segments = parse_path(expr)?; // 解析 a.b[0].c 语法
Ok(resolve_path(&value, &segments).map(|v| value_to_string(&v))) // None = 键/索引不存在
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_parse_simple() {
let segs = parse_path("a.b.c").unwrap();
assert_eq!(segs.len(), 3);
}
#[test]
fn test_parse_index() {
let segs = parse_path("items[0].name").unwrap();
assert_eq!(segs.len(), 3); // Key("items"), Index(0), Key("name")
}
#[test]
fn test_parse_all() {
let segs = parse_path("books[].title").unwrap();
assert_eq!(segs.len(), 3); // Key("books"), All, Key("title")
}
#[test]
fn test_resolve_simple() {
let json = r#"{"auth":{"token":"abc123"}}"#;
assert_eq!(
resolve_extract(json, "auth.token").unwrap(),
Some("abc123".into())
);
}
#[test]
fn test_resolve_array_index() {
let json = r#"{"items":[{"name":"a"},{"name":"b"}]}"#;
assert_eq!(
resolve_extract(json, "items[0].name").unwrap(),
Some("a".into())
);
}
#[test]
fn test_resolve_array_all() {
let json = r#"{"books":[{"title":"x"},{"title":"y"}]}"#;
assert_eq!(
resolve_extract(json, "books[].title").unwrap(),
Some("[\"x\",\"y\"]".into())
);
}
#[test]
fn test_resolve_regex() {
let raw = r#"token=abc123; path=/"#;
assert_eq!(
resolve_extract(raw, "/token=([^;]+)/").unwrap(),
Some("abc123".into())
);
}
#[test]
fn test_resolve_key_not_found() {
let json = r#"{"a":1}"#;
assert_eq!(resolve_extract(json, "a.b").unwrap(), None);
}
}
+71
View File
@@ -0,0 +1,71 @@
use std::collections::HashMap;
use anyhow::Ok;
use regex::{Captures, Regex};
use crate::workflow::types::StepResult;
pub fn render_template(
input: &str,
param: &HashMap<String, String>,
steps_result: &HashMap<String, StepResult>,
) -> anyhow::Result<String> {
let re = Regex::new(r"\$\{(env|param|step[0-9]+):([A-Za-z0-9_]+)\}").unwrap();
let result = re
.replace_all(input, |caps: &Captures| {
let kind = &caps[1];
let key = &caps[2];
match kind {
"env" => std::env::var(key).unwrap_or_default(),
"param" => param.get(key).cloned().unwrap_or_default(),
_ if kind.starts_with("step") => {
deal_step_data(kind, key, steps_result).unwrap_or_default()
}
_ => "".to_string(),
}
})
.to_string();
Ok(result)
}
fn deal_step_data(
kind: &str,
key: &str,
steps_result: &HashMap<String, StepResult>,
) -> anyhow::Result<String> {
Ok(steps_result
.get(kind)
.and_then(|step| step.var.get(key))
.cloned()
.unwrap_or_default())
}
#[test]
fn test_template() {
unsafe {
std::env::set_var("BASE_URL", "https://api.example.com");
}
let mut params = HashMap::new();
params.insert("order_id".to_string(), "12345".to_string());
let mut step_vars = HashMap::new();
step_vars.insert("name".to_string(), "demo".to_string());
let mut steps_result = HashMap::new();
steps_result.insert(
"step0".to_string(),
StepResult {
raw: String::new(),
var: step_vars,
},
);
let input = "${env:BASE_URL}/orders/${param:order_id}?key=${step0:name}";
let output = render_template(input, &params, &steps_result);
assert_eq!(
output.unwrap(),
"https://api.example.com/orders/12345?key=demo"
);
}
+57
View File
@@ -0,0 +1,57 @@
use std::collections::HashMap;
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone)]
pub struct StepResult {
pub raw: String,
pub var: HashMap<String, String>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct FlowConfig {
pub name: String,
pub desc: String,
#[serde(default)]
pub params: Vec<FlowParam>,
#[serde(default, skip_serializing)]
pub env: Vec<String>,
#[serde(skip_serializing)]
pub steps: Vec<FlowStep>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct FlowParam {
pub name: String,
#[serde(rename = "type")]
pub param_type: String,
pub required: bool,
pub description: String,
#[serde(skip_serializing)]
pub default: Option<String>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct FlowStep {
pub url: String,
pub method: Option<String>,
pub headers: Option<HashMap<String, String>>,
pub query: Option<Vec<(String, String)>>,
pub body: Option<String>,
#[serde(default)]
pub var: Vec<FlowVar>,
pub timeout: Option<u64>,
pub cache: Option<bool>,
pub next: Option<usize>,
#[serde(default)]
pub result: String,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct FlowVar {
pub name: String,
pub extract: Option<String>,
pub default: Option<String>,
pub abort: Option<bool>,
pub jump: Option<usize>,
}
+5 -1
View File
@@ -4,8 +4,12 @@ version = "0.1.0"
edition = "2024"
[dependencies]
# ── HTTP 客户端 ──
reqwest = { version = "0.12", default-features = false, features = ["json", "stream", "rustls-tls"] } # HTTP 请求(微信 API + DeepSeek API + 工具调用)
# ── 序列化 ──
serde = { version = "1.0", features = ["derive"] } # 序列化/反序列化
serde_json = "1.0"
serde_json = "1.0" # JSON 处理
serde_yaml = "0.9"
# ── 错误处理 ──
anyhow = "1"
thiserror = "2"
+1
View File
@@ -0,0 +1 @@
+1 -1
View File
@@ -43,7 +43,7 @@ impl ChatCompletionRequestMessage {
role,
content,
tool_call_id: None,
tool_calls: tool_calls,
tool_calls,
}
}
+1
View File
@@ -5,6 +5,7 @@ use crate::model::ai::ChatCompletionToolCall;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum QueueMessage {
Channel(ChannelMessage),
Notice(ChannelMessage, String),
Aissitant(ChannelMessage, AssistantMessage),
Tool(ChannelMessage, String, String, String),
}
+325
View File
@@ -0,0 +1,325 @@
# TOOLKIT WORKFLOW — YAML 驱动的 HTTP 工具链引擎
## 概述
TOOLKIT WORKFLOW 让你通过编写 YAML 文件来定义 LLM 工具。一个 YAML 文件描述一个工具:它接受哪些参数、调用哪些 HTTP API、如何从响应中提取数据、以及最终返回什么给 LLM。
核心能力:
- 一个 YAML 文件 = 一个完整的 LLM 工具,无需写代码
- 支持多步 HTTP 请求链,步骤之间可以传递变量
- 流程控制:`next` 跳转步骤、`jump` 降级到备源、`abort` 终止执行
- 键级缓存:`cache` 缓存变量,同一调用中缺键自动重请求
- 统一提取:`extract` 字段,语法自描述引擎(路径 / 正则)
---
## YAML 顶层结构
一个工具 YAML 文件包含以下顶层字段:
| 字段 | 必填 | 类型 | 说明 |
|------|------|------|------|
| `name` | 是 | string | 工具名,对应 LLM function.name |
| `desc` | 是 | string | 工具描述,对应 LLM function.description |
| `params` | 否 | array | LLM 可见的参数定义,映射为 function.parameters JSON Schema |
| `env` | 否 | array[string] | 该工具依赖的环境变量名列表,仅作文档和校验 |
| `steps` | 是 | array | 请求步骤链,按顺序执行 |
---
## `params` — 参数定义
定义 LLM 可以传入的参数,映射为 function calling 的 parameters schema。
| 字段 | 必填 | 类型 | 说明 |
|------|------|------|------|
| `name` | 是 | string | 参数名,在模板中通过 `${param:key}` 引用 |
| `type` | 是 | string | 取值:`string` / `number` / `integer` / `boolean` |
| `required` | 是 | boolean | 是否必填 |
| `description` | 是 | string | 参数说明,LLM 据此理解参数含义 |
| `default` | 否 | any | 默认值,LLM 未传时自动使用 |
---
## `steps` — 请求步骤
每个步骤定义一个 HTTP 请求及其后续处理。
### 步骤字段
| 字段 | 必填 | 类型 | 默认值 | 说明 |
|------|------|------|--------|------|
| `url` | 是 | string | — | 请求 URL,支持模板语法 |
| `method` | 否 | string | `GET` | HTTP 方法:`GET` / `POST` / `PUT` / `DELETE` / `PATCH` |
| `headers` | 否 | map | — | 请求头键值对,值支持模板语法 |
| `query` | 否 | array | — | URL 查询参数,格式为 `[{key, value}]` 元组列表,值支持模板语法 |
| `body` | 否 | string | — | 请求体,支持模板语法 |
| `timeout` | 否 | integer | 30 | 超时时间(秒) |
| `cache` | 否 | boolean | false | 是否对该步骤启用键级缓存 |
| `next` | 否 | integer | — | 步骤成功后跳转到第几个步骤(0-based),不设则顺序执行下一步 |
| `var` | 否 | array | — | 变量提取列表 |
| `result` | 否 | string | — | 最终返回模板,有此字段则该步为终步 |
### 步骤示例
```yaml
steps:
- url: ${env:BASE_URL}/auth/token
method: POST
headers:
Content-Type: application/json
body: |
{"client_id": "${env:CLIENT_ID}", "client_secret": "${env:CLIENT_SECRET}"}
timeout: 15
cache: true
var:
- name: token
extract: access_token
abort: true
next: 1
- url: ${env:BASE_URL}/orders/${param:order_id}
method: GET
headers:
Authorization: Bearer ${step0:token}
var:
- name: status
extract: data.status
- name: amount
extract: data.amount
result: |
订单号: ${param:order_id}
状态: ${step1:status}
金额: ${step1:amount}
```
---
## `var` — 变量提取
每个步骤执行完 HTTP 请求后,按顺序处理 `var` 列表中的条目,从响应中提取命名变量。提取的变量通过 `${stepN:var_name}` 在后续步骤中引用。
### `extract` — 提取表达式
一个字段覆盖所有提取场景。值的语法决定使用的引擎:
| 语法 | 引擎 | 说明 |
|------|------|------|
| 不以 `/` 开头 | 路径引擎 | 对 JSON/YAML 响应执行结构化路径导航 |
| 以 `/` 开头且以 `/` 结尾 | 正则引擎 | 对原始响应文本执行正则匹配,取捕获组 1 |
**路径引擎**:自动检测响应格式(先 JSON,失败则 YAML),然后按路径导航取值。
| 语法 | 含义 | 示例 |
|------|------|------|
| `key` | 访问对象字段 | `data.status` |
| `key[N]` | 访问数组第 N 个元素 | `items[0].name` |
| `key[]` | 遍历数组所有元素 | `books[]` |
| `key[].field` | 遍历数组,对每个元素取字段 | `books[].title` |
> 键名建议仅使用 `[a-zA-Z0-9_]`,避免含 `.` `[` `]` 的特殊键名。
**正则引擎**`/pattern/` 形式,对原始响应文本执行正则匹配,取第一个捕获组。
```yaml
# 路径:从 JSON 响应中提取
var:
- name: token
extract: auth.access_token
# 正则:从任意文本中提取
var:
- name: session
extract: /session[= ]([a-f0-9]+)/
```
### 提取失败处理(每个条目最多选一种)
| 配置 | 提取成功 | 提取失败 |
|------|---------|---------|
| 不设任何失败处理 | 存入变量 | 存入空字符串 `""`,继续执行 |
| `default: "值"` | 存入变量 | 存入默认值,继续执行 |
| `abort: true` | 存入变量 | 终止整个工具,返回错误给 LLM |
| `jump: N` | 存入变量 | 跳转到第 N 个步骤执行,N 从 0 开始 |
### var 条目完整字段
| 字段 | 必填 | 类型 | 默认值 | 说明 |
|------|------|------|--------|------|
| `name` | 是 | string | — | 变量名,通过 `${stepN:name}` 引用 |
| `extract` | 否 | string | — | 提取表达式,语法自描述引擎(路径 / 正则) |
| `default` | 否 | string | — | 提取失败时的默认值 |
| `abort` | 否 | boolean | false | 提取失败时是否终止工具 |
| `jump` | 否 | integer | — | 提取失败时跳转到第几个步骤(0-based) |
---
## 流程控制
三种流程控制机制,覆盖顺序执行、条件跳转和异常终止。
### `next` — 成功后跳转
- 位置:步骤级别
- 触发条件:步骤所有 var 都没有因 abort/jump 而失败
- 效果:跳转到指定步骤号(0-based),不设则顺序执行下一步
- 用途:跳过某些步骤、实现条件分支
### `jump` — 提取失败时降级
- 位置:var 条目级别
- 触发条件:该条目的提取失败且未设置 default
- 效果:跳转到指定步骤号(0-based),当前步骤剩余 var 不再执行
- 用途:主数据源失败时自动切换到备数据源
### `abort` — 提取失败时终止
- 位置:var 条目级别
- 触发条件:该条目的提取失败
- 效果:终止整个工具执行,返回错误信息给 LLM
- 用途:关键变量(如认证 token)获取失败时立即报错
---
## `cache` — 键级缓存
当步骤设置 `cache: true` 时,该步骤提取的所有变量会被缓存。
缓存行为:
- **命中**:缓存中存在该步骤所有 var.name → 跳过 HTTP 请求,直接返回缓存值
- **未命中**:任意一个 var.name 缺失 → 发起 HTTP 请求,提取所有变量,全部存入缓存
- **生命周期**:单次工具调用期间(内存级),工具返回后释放
- **缓存键**:由步骤的 `url + method + headers + body` 模板展开后的哈希确定
典型用法:第一步获取认证 token 并缓存,后续多步复用 token 时不会重复请求认证接口。
---
## 模板语法
以下字段支持模板语法:`url``headers` 的值、`body``result`
### 三种引用
| 语法 | 来源 | 示例 |
|------|------|------|
| `${param:key}` | LLM 传入的工具参数 | `${param:city}``${param:order_id}` |
| `${env:VAR}` | 系统环境变量 | `${env:API_KEY}``${env:BASE_URL}` |
| `${stepN:var_name}` | 第 N 个步骤提取的变量,N 从 0 开始 | `${step0:token}``${step1:status}` |
### 约束
1. 禁止嵌套引用:`${step0:${env:KEY}}` 不允许
2. 禁止原始步骤引用:`${step:0}` 不允许,必须先通过 var 提取为命名变量
---
## `result` — 最终返回
步骤的 `result` 字段定义该工具最终返回给 LLM 的内容。
- 如果某步设置了 `result`,该步即为**终步**,执行后不再继续后续步骤
- 如果所有步骤都未设置 `result`,默认返回最后一步的原始响应体
- `result` 支持模板语法,可以拼接多个变量和文字
```yaml
# 返回单个变量
result: ${step1:status}
# 多行模板拼接
result: |
订单号: ${param:order_id}
状态: ${step1:status}
金额: ${step1:amount}
```
---
## 执行流程
```
1. 加载 YAML → 解析为 YamlToolDef
2. 解析 LLM 参数 → params HashMap
3. 加载 env 声明的环境变量 → env HashMap
4. 初始化 cache = {}
5. step_idx = 0
while step_idx < steps.len():
┌─ 当前步骤执行 ──────────────────────────────┐
│ │
│ [cache: true?] │
│ ├─ 缓存命中(所有 var 键都存在)→ 跳过HTTP │
│ └─ 缓存未命中 → 继续 │
│ │
│ 展开 url/headers/body 中的模板 │
│ ${param:key} → params["key"] │
│ ${env:VAR} → env["VAR"] │
│ ${stepN:var} → ctx.vars[(N, var)] │
│ │
│ HttpRequest 执行 → HttpResponse │
│ 保存到 ctx.steps[step_num] │
│ │
│ for each var: │
│ extract 语法分发: │
│ ├─ /.../ → 正则引擎,取捕获组1 │
│ └─ 其他 → 路径引擎(JSON→YAML 自动检测) │
│ ├─ 成功 → 存入 ctx.vars + step_vars │
│ └─ 失败 → │
│ ├─ abort:true → 终止,返回错误 │
│ ├─ default:x → 使用默认值 │
│ ├─ jump:N → step_idx = N, break │
│ └─ (无) → 空字符串 │
│ │
│ [cache: true?] → 将 step_vars 存入 cache │
│ │
│ [result 存在?] │
│ └─ 展开 result 模板 → 返回给 LLM → 结束 │
│ │
│ [next 存在?] │
│ ├─ next:N → step_idx = N │
│ └─ (无) → step_idx += 1 │
│ │
└──────────────────────────────────────────────┘
6. 循环结束 → 返回最后一步原始响应体
```
---
## 错误处理
| 场景 | 处理方式 |
|------|---------|
| YAML 文件格式错误 | 启动时打印警告,跳过该文件 |
| 模板引用了未定义的参数 | 替换为空字符串,记录警告日志 |
| 环境变量未设置 | 替换为空字符串,记录警告日志 |
| HTTP 请求失败(连接/超时) | 返回错误详情给 LLM |
| `var.abort: true` 且提取失败 | 终止,返回 "变量 xxx 提取失败" |
| `var.jump: N` 且提取失败 | 跳转到第 N 个步骤,N 从 0 开始,不报错 |
| `var.default` 且提取失败 | 使用默认值,不报错 |
| 路径语法错误(空段、非法索引) | 返回错误给 LLM,附带路径表达式 |
| 路径键不存在 / 索引越界 | 提取失败,触发 default/abort/jump |
| 正则语法错误 | 返回错误给 LLM,附带正则表达式 |
| 正则无匹配 | 提取失败,触发 default/abort/jump |
| 响应体非 JSON 也非 YAML(路径引擎) | 返回错误给 LLM |
| `next` / `jump` 目标步骤不存在 | 终止,返回配置错误 |
---
## 扩展预留
以下字段预留供后续版本实现:
| 字段 | 位置 | 用途 |
|------|------|------|
| `retry` | `steps[]` | 失败重试次数与退避策略 |
| `cache.ttl` | `steps[]` | 缓存有效期(秒) |
| `validate` | `params[]` | 参数级正则校验 |
| `sensitive` | `env[]` | 标记敏感环境变量,日志脱敏 |
| `loop` | `steps[]` | 条件循环(分页自动翻页) |
| `export` | `steps[]` | 将变量导出为环境变量供子进程使用 |
| `pre_request` | 顶层 | 工具级别的前置请求(全局认证) |
+10
View File
@@ -3,6 +3,10 @@ name = "service"
version = "0.1.0"
edition = "2024"
[[bin]]
name = "ias"
path = "src/main.rs"
[dependencies]
# ── 公共模块 ──
common = { path = "../common" }
@@ -21,6 +25,7 @@ tokio = { version = "1.0", features = [
"process",
"signal",
"io-util",
"net",
] } # 异步运行时核心
# ── HTTP 客户端 ──
reqwest = { version = "0.12", default-features = false, features = [
@@ -46,7 +51,12 @@ sqlx = { version = "0.9.0", features = [
# ── 错误处理 ──
anyhow = "1"
thiserror = "2"
# ── 日志管理 ──
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
# ── cli ──
clap = { version = "4.6.1", features = ["derive"] }
# ── 时间处理 ──
chrono = { version = "0.4", features = ["serde"] } # 日期时间
# ── 唯一标识 ──
uuid = { version = "1", features = ["v4", "serde"] }
@@ -0,0 +1,13 @@
CREATE TABLE ias_web_pages (
id BIGSERIAL PRIMARY KEY,
slug VARCHAR(64) NOT NULL UNIQUE,
page_type VARCHAR(64) NOT NULL,
title VARCHAR(255) NOT NULL,
summary TEXT,
content TEXT NOT NULL,
source_label VARCHAR(255),
metadata TEXT NOT NULL DEFAULT '{}',
created_at TIMESTAMP NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_ias_web_pages_type_slug ON ias_web_pages(page_type, slug);
+7 -1
View File
@@ -17,6 +17,7 @@ pub async fn start_loop(
account_id,
updates_buf,
} => {
state.runtime.record_wechat_event(account_id.clone()).await;
persist_updates_buf(&state.db, account_id, updates_buf).await;
}
WeChatEvent::Message {
@@ -24,6 +25,7 @@ pub async fn start_loop(
message,
updates_buf,
} => {
state.runtime.record_wechat_event(account_id.clone()).await;
persist_updates_buf(&state.db, account_id.clone(), updates_buf).await;
if message.msg_type != Some(WeixinMessage::TYPE_USER) {
@@ -43,7 +45,7 @@ pub async fn start_loop(
.clone()
.send(QueueMessage::Channel(ChannelMessage {
channel: "wechat".to_string(),
account_id: account_id,
account_id,
from_user_id: from_user_id.to_string(),
content: text.to_string(),
context: message.context_token,
@@ -52,6 +54,10 @@ pub async fn start_loop(
.unwrap();
}
WeChatEvent::PollError { account_id, error } => {
state
.runtime
.record_wechat_error(account_id.clone(), error.clone())
.await;
println!("微信账号轮询失败 account_id={account_id}: {error}");
}
}
+6 -4
View File
@@ -27,6 +27,10 @@ pub async fn init_wechat(state: Option<AppState>) -> Option<WeChatMultiAccountMa
.await
.unwrap();
}
state
.runtime
.replace_wechat_accounts(manager.account_ids())
.await;
start_loop(Arc::new(state.clone()), event_rx).await.unwrap();
return Some(manager);
@@ -67,7 +71,7 @@ pub async fn login_user(pool: &PgPool) {
user_name.trim().to_string()
};
let user = WechatAccountCreate {
account_id: account_id,
account_id,
token: account.token,
base_url: account.base_url,
user_id: account.user_id,
@@ -75,9 +79,7 @@ pub async fn login_user(pool: &PgPool) {
};
// todo 后续需要处理错误
UserRepositor::add_wechat_account(&pool, user)
.await
.unwrap();
UserRepositor::add_wechat_account(pool, user).await.unwrap();
}
}
Err(err) => {
+1 -1
View File
@@ -1,2 +1,2 @@
pub mod manager;
pub mod looper;
pub mod manager;
+24 -1
View File
@@ -12,7 +12,10 @@ pub struct Cli {
#[derive(Debug, Subcommand)]
pub enum Commands {
// 启动服务
Service,
Service {
#[command(subcommand)]
command: Option<ServiceAction>,
},
// 配置渠道
Channel {
#[command(subcommand)]
@@ -20,6 +23,26 @@ pub enum Commands {
},
}
#[derive(Debug, Clone, Subcommand)]
pub enum ServiceAction {
/// 前台运行服务
Run,
/// 后台启动服务
Start,
/// 停止后台服务
Stop,
/// 重启后台服务
Restart,
/// 查看服务运行状态
Status,
/// 查看后台服务日志
Logs {
/// 输出最后多少行日志
#[arg(short, long, default_value_t = 80)]
lines: usize,
},
}
#[derive(Debug, Subcommand)]
pub enum Channels {
// 微信渠道配置
+142
View File
@@ -0,0 +1,142 @@
use std::path::PathBuf;
use std::sync::Arc;
use serde::{Deserialize, Serialize};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use tokio::sync::watch;
use crate::core::runtime::{RuntimeSnapshot, RuntimeState};
#[derive(Debug, Clone)]
pub struct RuntimePaths {
pub dir: PathBuf,
pub pid_file: PathBuf,
pub socket_file: PathBuf,
pub state_file: PathBuf,
pub log_file: PathBuf,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ControlResponse {
Status { snapshot: RuntimeSnapshot },
Ack { message: String },
Error { message: String },
}
impl RuntimePaths {
pub fn new() -> Self {
let dir = std::env::var_os("XDG_RUNTIME_DIR")
.map(PathBuf::from)
.unwrap_or_else(|| {
let user = std::env::var("USER").unwrap_or_else(|_| "user".to_string());
std::env::temp_dir().join(format!("ias-{user}"))
})
.join("ias");
Self {
pid_file: dir.join("ias.pid"),
socket_file: dir.join("ias.sock"),
state_file: dir.join("ias.state.json"),
log_file: dir.join("ias.log"),
dir,
}
}
pub fn ensure_dir(&self) -> anyhow::Result<()> {
std::fs::create_dir_all(&self.dir)?;
Ok(())
}
}
pub async fn start_control_server(
runtime: Arc<RuntimeState>,
paths: RuntimePaths,
shutdown_tx: watch::Sender<bool>,
mut shutdown_rx: watch::Receiver<bool>,
) -> anyhow::Result<()> {
paths.ensure_dir()?;
if paths.socket_file.exists() {
if UnixStream::connect(&paths.socket_file).await.is_ok() {
anyhow::bail!(
"服务控制 socket 已存在且可连接: {}",
paths.socket_file.display()
);
}
std::fs::remove_file(&paths.socket_file)?;
}
let listener = UnixListener::bind(&paths.socket_file)?;
loop {
tokio::select! {
result = listener.accept() => {
let (stream, _) = result?;
let runtime = runtime.clone();
let shutdown_tx = shutdown_tx.clone();
tokio::spawn(async move {
if let Err(err) = handle_control_stream(stream, runtime, shutdown_tx).await {
eprintln!("处理服务控制命令失败: {err}");
}
});
}
changed = shutdown_rx.changed() => {
if changed.is_err() || *shutdown_rx.borrow() {
break;
}
}
}
}
Ok(())
}
pub async fn request_control(
paths: &RuntimePaths,
command: &str,
) -> anyhow::Result<ControlResponse> {
let mut stream = UnixStream::connect(&paths.socket_file).await?;
stream.write_all(command.as_bytes()).await?;
stream.write_all(b"\n").await?;
let mut reader = BufReader::new(stream);
let mut line = String::new();
reader.read_line(&mut line).await?;
let response = serde_json::from_str(line.trim())?;
Ok(response)
}
async fn handle_control_stream(
stream: UnixStream,
runtime: Arc<RuntimeState>,
shutdown_tx: watch::Sender<bool>,
) -> anyhow::Result<()> {
let mut reader = BufReader::new(stream);
let mut command = String::new();
reader.read_line(&mut command).await?;
let command = command.trim();
let response = match command {
"status" => ControlResponse::Status {
snapshot: runtime.snapshot().await,
},
"shutdown" => {
runtime.mark_stopping().await;
let _ = shutdown_tx.send(true);
ControlResponse::Ack {
message: "服务正在停止".to_string(),
}
}
_ => ControlResponse::Error {
message: format!("未知控制命令: {command}"),
},
};
let response = serde_json::to_string(&response)?;
let stream = reader.get_mut();
stream.write_all(response.as_bytes()).await?;
stream.write_all(b"\n").await?;
stream.flush().await?;
Ok(())
}
+3
View File
@@ -1,3 +1,6 @@
pub mod cli;
pub mod context;
pub mod control;
pub mod queue;
pub mod runtime;
pub mod supervisor;
+174 -13
View File
@@ -2,64 +2,225 @@ use crate::core::context::{
add_assistant_message, add_tool_message, add_user_message, snapshot_context,
};
use ai::core::send_message;
use common::queue::QueueMessage;
use common::queue::{ChannelMessage, QueueMessage};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{mpsc::Receiver, mpsc::Sender};
use tokio::task::JoinHandle;
use wechat::manager::WeChatMultiAccountManager;
struct TypingSession {
client_id: String,
refresh_task: JoinHandle<()>,
}
pub async fn create_queue(
sender: Sender<QueueMessage>,
mut receiver: Receiver<QueueMessage>,
manager: WeChatMultiAccountManager,
) -> anyhow::Result<()> {
tokio::spawn(async move {
) -> anyhow::Result<JoinHandle<()>> {
let manager = Arc::new(manager);
let handle = tokio::spawn(async move {
let mut typing_sessions: HashMap<String, TypingSession> = HashMap::new();
while let Some(msg) = receiver.recv().await {
let queue_sender = sender.clone();
match msg {
QueueMessage::Aissitant(channel, assistant) => {
let content = assistant.content.clone();
let reasoning = assistant.reasoning.clone().unwrap_or_default();
println!("\n\n收到llm回复:\n{}\n-------\n{}", reasoning, content);
if !content.trim().is_empty() || !reasoning.trim().is_empty() {
let has_tool_calls = assistant
.tool_calls
.as_ref()
.is_some_and(|calls| !calls.is_empty());
println!(
"\n\n收到llm回复:\n{}\n-------\n{}\n工具调用数:{}",
reasoning,
content,
assistant.tool_calls.as_ref().map_or(0, Vec::len)
);
if has_tool_calls {
refresh_typing(manager.clone(), &typing_sessions, &channel).await;
} else if !content.trim().is_empty() || !reasoning.trim().is_empty() {
let text = if !content.trim().is_empty() {
content.trim()
} else {
reasoning.trim()
};
let session_key = typing_key(&channel);
if let Some(session) = typing_sessions.remove(&session_key) {
session.refresh_task.abort();
manager
.send_text_with_client_id(
&channel.account_id,
&channel.from_user_id,
text,
channel.context.as_deref(),
&session.client_id,
)
.await
.unwrap();
} else {
manager
.send_text(
&channel.account_id,
&channel.from_user_id,
&text,
text,
channel.context.as_deref(),
)
.await
.unwrap();
}
} else if assistant
.tool_calls
.as_ref()
.is_none_or(|calls| calls.is_empty())
{
stop_typing(&mut typing_sessions, &channel);
}
add_assistant_message(assistant.content, assistant.tool_calls).await;
}
QueueMessage::Notice(channel, text) => {
let text = text.trim();
if !text.is_empty() {
println!("\n\n发送工具调用提醒:\n{}", text);
if let Err(err) = manager
.send_text(
&channel.account_id,
&channel.from_user_id,
text,
channel.context.as_deref(),
)
.await
{
eprintln!("发送工具调用提醒失败: {}", err);
}
}
}
QueueMessage::Channel(bundle) => {
println!(
"\n\n收到来自{}消息:\n{}",
bundle.from_user_id.clone(),
bundle.content.clone()
);
start_typing(manager.clone(), &mut typing_sessions, &bundle).await;
// todo 目前按单用户处理
add_user_message(bundle.content.clone()).await;
send_message(queue_sender, snapshot_context().await, bundle)
.await
.unwrap();
let messages = snapshot_context().await;
tokio::spawn(async move {
if let Err(err) = send_message(queue_sender, messages, bundle).await {
eprintln!("发送消息失败: {}", err);
}
});
}
QueueMessage::Tool(channel, tool_call_id, name, result) => {
println!("\n\n调用{}工具的结果:\n{}", name, result);
refresh_typing(manager.clone(), &typing_sessions, &channel).await;
add_tool_message(tool_call_id, result).await;
send_message(queue_sender, snapshot_context().await, channel)
.await
.unwrap();
let messages = snapshot_context().await;
tokio::spawn(async move {
if let Err(err) = send_message(queue_sender, messages, channel).await {
eprintln!("发送消息失败: {}", err);
}
});
}
}
}
});
Ok(())
Ok(handle)
}
async fn start_typing(
manager: Arc<WeChatMultiAccountManager>,
sessions: &mut HashMap<String, TypingSession>,
channel: &ChannelMessage,
) {
let key = typing_key(channel);
stop_typing(sessions, channel);
let client_id = match manager
.send_typing(
&channel.account_id,
&channel.from_user_id,
channel.context.as_deref(),
)
.await
{
Ok(client_id) => client_id,
Err(err) => {
eprintln!("设置微信输入中状态失败: {}", err);
return;
}
};
let account_id = channel.account_id.clone();
let from_user_id = channel.from_user_id.clone();
let context = channel.context.clone();
let refresh_client_id = client_id.clone();
let refresh_manager = manager.clone();
let refresh_task = tokio::spawn(async move {
loop {
tokio::time::sleep(Duration::from_secs(4)).await;
if let Err(err) = refresh_manager
.send_typing_with_client_id(
&account_id,
&from_user_id,
context.as_deref(),
&refresh_client_id,
)
.await
{
eprintln!("刷新微信输入中状态失败: {}", err);
return;
}
}
});
sessions.insert(
key,
TypingSession {
client_id,
refresh_task,
},
);
}
async fn refresh_typing(
manager: Arc<WeChatMultiAccountManager>,
sessions: &HashMap<String, TypingSession>,
channel: &ChannelMessage,
) {
let Some(session) = sessions.get(&typing_key(channel)) else {
return;
};
if let Err(err) = manager
.send_typing_with_client_id(
&channel.account_id,
&channel.from_user_id,
channel.context.as_deref(),
&session.client_id,
)
.await
{
eprintln!("刷新微信输入中状态失败: {}", err);
}
}
fn stop_typing(sessions: &mut HashMap<String, TypingSession>, channel: &ChannelMessage) {
if let Some(session) = sessions.remove(&typing_key(channel)) {
session.refresh_task.abort();
}
}
fn typing_key(channel: &ChannelMessage) -> String {
format!(
"{}:{}:{}",
channel.account_id,
channel.from_user_id,
channel.context.as_deref().unwrap_or_default()
)
}
// pub async fn send_to_queue(
+195
View File
@@ -0,0 +1,195 @@
use std::collections::HashMap;
use std::path::Path;
use std::process;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ServicePhase {
Starting,
Running,
Stopping,
Stopped,
Failed,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AccountRuntimePhase {
Running,
Retrying,
Stopped,
Failed,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AccountRuntimeStatus {
pub account_id: String,
pub phase: AccountRuntimePhase,
pub last_error: Option<String>,
pub last_event_at: Option<DateTime<Utc>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RuntimeSnapshot {
pub phase: ServicePhase,
pub pid: u32,
pub started_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
pub http_addr: String,
pub queue_alive: bool,
pub wechat_accounts: Vec<AccountRuntimeStatus>,
pub last_error: Option<String>,
}
struct RuntimeInner {
phase: ServicePhase,
started_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
http_addr: String,
queue_alive: bool,
wechat_accounts: HashMap<String, AccountRuntimeStatus>,
last_error: Option<String>,
}
pub struct RuntimeState {
inner: Mutex<RuntimeInner>,
}
impl RuntimeState {
pub fn new(http_addr: String) -> Self {
let now = Utc::now();
Self {
inner: Mutex::new(RuntimeInner {
phase: ServicePhase::Starting,
started_at: now,
updated_at: now,
http_addr,
queue_alive: false,
wechat_accounts: HashMap::new(),
last_error: None,
}),
}
}
pub async fn mark_running(&self) {
let mut inner = self.inner.lock().await;
inner.phase = ServicePhase::Running;
inner.updated_at = Utc::now();
inner.last_error = None;
}
pub async fn mark_stopping(&self) {
let mut inner = self.inner.lock().await;
inner.phase = ServicePhase::Stopping;
inner.queue_alive = false;
inner.updated_at = Utc::now();
}
pub async fn mark_stopped(&self) {
let mut inner = self.inner.lock().await;
inner.phase = ServicePhase::Stopped;
inner.queue_alive = false;
inner.updated_at = Utc::now();
}
pub async fn mark_failed(&self, error: impl Into<String>) {
let mut inner = self.inner.lock().await;
inner.phase = ServicePhase::Failed;
inner.queue_alive = false;
inner.updated_at = Utc::now();
inner.last_error = Some(error.into());
}
pub async fn set_queue_alive(&self, alive: bool) {
let mut inner = self.inner.lock().await;
inner.queue_alive = alive;
inner.updated_at = Utc::now();
}
pub async fn replace_wechat_accounts(&self, account_ids: Vec<String>) {
let mut inner = self.inner.lock().await;
let now = Utc::now();
inner.wechat_accounts = account_ids
.into_iter()
.map(|account_id| {
let status = AccountRuntimeStatus {
account_id: account_id.clone(),
phase: AccountRuntimePhase::Running,
last_error: None,
last_event_at: Some(now),
};
(account_id, status)
})
.collect();
inner.updated_at = now;
}
pub async fn record_wechat_event(&self, account_id: impl Into<String>) {
self.update_wechat_account(account_id.into(), AccountRuntimePhase::Running, None)
.await;
}
pub async fn record_wechat_error(
&self,
account_id: impl Into<String>,
error: impl Into<String>,
) {
self.update_wechat_account(
account_id.into(),
AccountRuntimePhase::Retrying,
Some(error.into()),
)
.await;
}
async fn update_wechat_account(
&self,
account_id: String,
phase: AccountRuntimePhase,
last_error: Option<String>,
) {
let mut inner = self.inner.lock().await;
let now = Utc::now();
let status = inner
.wechat_accounts
.entry(account_id.clone())
.or_insert_with(|| AccountRuntimeStatus {
account_id,
phase: AccountRuntimePhase::Running,
last_error: None,
last_event_at: None,
});
status.phase = phase;
status.last_error = last_error;
status.last_event_at = Some(now);
inner.updated_at = now;
}
pub async fn snapshot(&self) -> RuntimeSnapshot {
let inner = self.inner.lock().await;
let mut wechat_accounts = inner.wechat_accounts.values().cloned().collect::<Vec<_>>();
wechat_accounts.sort_by(|left, right| left.account_id.cmp(&right.account_id));
RuntimeSnapshot {
phase: inner.phase.clone(),
pid: process::id(),
started_at: inner.started_at,
updated_at: inner.updated_at,
http_addr: inner.http_addr.clone(),
queue_alive: inner.queue_alive,
wechat_accounts,
last_error: inner.last_error.clone(),
}
}
pub async fn write_snapshot(&self, path: &Path) -> anyhow::Result<()> {
let snapshot = self.snapshot().await;
let data = serde_json::to_vec_pretty(&snapshot)?;
std::fs::write(path, data)?;
Ok(())
}
}
+275
View File
@@ -0,0 +1,275 @@
use std::fs::{File, OpenOptions};
use std::io::{Read, Write};
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};
use crate::core::control::{ControlResponse, RuntimePaths, request_control};
use crate::core::runtime::{RuntimeSnapshot, ServicePhase};
pub async fn start_service() -> anyhow::Result<()> {
let paths = RuntimePaths::new();
paths.ensure_dir()?;
if let Ok(ControlResponse::Status { snapshot }) = request_control(&paths, "status").await {
println!("服务已经在运行,PID: {}", snapshot.pid);
print_snapshot(&snapshot);
return Ok(());
}
cleanup_stale_files(&paths)?;
let mut log = OpenOptions::new()
.create(true)
.append(true)
.open(&paths.log_file)?;
writeln!(log, "\n===== 启动后台服务 =====")?;
let exe = std::env::current_exe()?;
let stdout = log.try_clone()?;
let stderr = log;
let child = Command::new(exe)
.arg("service")
.arg("run")
.env("IAS_BACKGROUND", "1")
.stdin(Stdio::null())
.stdout(Stdio::from(stdout))
.stderr(Stdio::from(stderr))
.spawn()?;
println!("后台服务启动中,PID: {}", child.id());
wait_until_ready(&paths, Duration::from_secs(8)).await?;
println!("日志文件: {}", paths.log_file.display());
Ok(())
}
pub async fn stop_service() -> anyhow::Result<()> {
let paths = RuntimePaths::new();
paths.ensure_dir()?;
match request_control(&paths, "shutdown").await {
Ok(ControlResponse::Ack { message }) => println!("{message}"),
Ok(ControlResponse::Error { message }) => println!("停止失败: {message}"),
Ok(ControlResponse::Status { .. }) => println!("停止失败: 控制端返回了意外状态响应"),
Err(_) => {
if let Some(pid) = read_pid(&paths)? {
if process_is_alive(pid) {
println!("控制 socket 不可用,发送 SIGTERM 到 PID: {pid}");
send_sigterm(pid)?;
} else {
println!("服务未运行,清理过期运行文件");
cleanup_stale_files(&paths)?;
return Ok(());
}
} else {
println!("服务未运行");
return Ok(());
}
}
}
wait_until_stopped(&paths, Duration::from_secs(10)).await?;
cleanup_stale_files(&paths)?;
println!("服务已停止");
Ok(())
}
pub async fn restart_service() -> anyhow::Result<()> {
stop_service().await?;
start_service().await
}
pub async fn status_service() -> anyhow::Result<()> {
let paths = RuntimePaths::new();
paths.ensure_dir()?;
match request_control(&paths, "status").await {
Ok(ControlResponse::Status { snapshot }) => print_snapshot(&snapshot),
Ok(ControlResponse::Error { message }) => println!("状态查询失败: {message}"),
Ok(ControlResponse::Ack { message }) => println!("{message}"),
Err(_) => print_offline_status(&paths)?,
}
Ok(())
}
pub fn print_logs(lines: usize) -> anyhow::Result<()> {
let paths = RuntimePaths::new();
if !paths.log_file.exists() {
println!("日志文件不存在: {}", paths.log_file.display());
return Ok(());
}
let mut file = File::open(&paths.log_file)?;
let mut content = String::new();
file.read_to_string(&mut content)?;
let mut tail = content.lines().rev().take(lines).collect::<Vec<_>>();
tail.reverse();
for line in tail {
println!("{line}");
}
Ok(())
}
async fn wait_until_ready(paths: &RuntimePaths, timeout: Duration) -> anyhow::Result<()> {
let started = Instant::now();
loop {
if let Ok(ControlResponse::Status { snapshot }) = request_control(paths, "status").await {
println!("服务已启动");
print_snapshot(&snapshot);
return Ok(());
}
if started.elapsed() >= timeout {
anyhow::bail!(
"服务未在 {} 秒内就绪,请查看日志: {}",
timeout.as_secs(),
paths.log_file.display()
);
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
}
async fn wait_until_stopped(paths: &RuntimePaths, timeout: Duration) -> anyhow::Result<()> {
let started = Instant::now();
loop {
if request_control(paths, "status").await.is_err()
&& read_pid(paths)?.is_none_or(|pid| !process_is_alive(pid))
{
return Ok(());
}
if started.elapsed() >= timeout {
anyhow::bail!(
"服务未在 {} 秒内停止,请检查 PID 文件: {}",
timeout.as_secs(),
paths.pid_file.display()
);
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
}
pub fn write_pid_file(paths: &RuntimePaths) -> anyhow::Result<()> {
paths.ensure_dir()?;
std::fs::write(&paths.pid_file, std::process::id().to_string())?;
Ok(())
}
pub fn cleanup_runtime_files(paths: &RuntimePaths) {
if let Some(pid) = read_pid(paths).ok().flatten()
&& pid != std::process::id()
{
return;
}
let _ = std::fs::remove_file(&paths.pid_file);
let _ = std::fs::remove_file(&paths.socket_file);
}
fn cleanup_stale_files(paths: &RuntimePaths) -> anyhow::Result<()> {
if let Some(pid) = read_pid(paths)?
&& process_is_alive(pid)
{
anyhow::bail!(
"已有服务进程存在但控制 socket 不可用,PID: {}。请先执行 stop 或检查日志: {}",
pid,
paths.log_file.display()
);
}
let _ = std::fs::remove_file(&paths.pid_file);
let _ = std::fs::remove_file(&paths.socket_file);
Ok(())
}
fn print_offline_status(paths: &RuntimePaths) -> anyhow::Result<()> {
if let Some(pid) = read_pid(paths)?
&& process_is_alive(pid)
{
println!("服务进程存在但控制 socket 不可用,PID: {pid}");
println!("socket: {}", paths.socket_file.display());
println!("日志文件: {}", paths.log_file.display());
return Ok(());
}
println!("服务状态: stopped");
if paths.state_file.exists() {
println!("最后状态快照: {}", paths.state_file.display());
}
println!("日志文件: {}", paths.log_file.display());
Ok(())
}
fn print_snapshot(snapshot: &RuntimeSnapshot) {
println!("服务状态: {}", phase_name(&snapshot.phase));
println!("PID: {}", snapshot.pid);
println!("HTTP: {}", snapshot.http_addr);
println!("启动时间: {}", snapshot.started_at);
println!(
"队列: {}",
if snapshot.queue_alive {
"running"
} else {
"stopped"
}
);
if let Some(error) = &snapshot.last_error {
println!("最近错误: {error}");
}
if snapshot.wechat_accounts.is_empty() {
println!("微信账号: 0");
} else {
println!("微信账号:");
for account in &snapshot.wechat_accounts {
let last_event = account
.last_event_at
.map(|time| time.to_string())
.unwrap_or_else(|| "-".to_string());
let last_error = account.last_error.as_deref().unwrap_or("-");
println!(
" - {}: {:?}, last_event={}, last_error={}",
account.account_id, account.phase, last_event, last_error
);
}
}
}
fn phase_name(phase: &ServicePhase) -> &'static str {
match phase {
ServicePhase::Starting => "starting",
ServicePhase::Running => "running",
ServicePhase::Stopping => "stopping",
ServicePhase::Stopped => "stopped",
ServicePhase::Failed => "failed",
}
}
fn read_pid(paths: &RuntimePaths) -> anyhow::Result<Option<u32>> {
if !paths.pid_file.exists() {
return Ok(None);
}
let content = std::fs::read_to_string(&paths.pid_file)?;
let pid = content.trim().parse::<u32>()?;
Ok(Some(pid))
}
fn process_is_alive(pid: u32) -> bool {
std::path::Path::new("/proc").join(pid.to_string()).exists()
}
fn send_sigterm(pid: u32) -> anyhow::Result<()> {
let status = Command::new("kill")
.arg("-TERM")
.arg(pid.to_string())
.status()?;
if !status.success() {
anyhow::bail!("发送 SIGTERM 失败,PID: {pid}");
}
Ok(())
}
+29 -5
View File
@@ -1,20 +1,44 @@
use std::sync::Arc;
use anyhow::Ok;
use axum::routing::{get};
use anyhow::anyhow;
use axum::routing::get;
use axum::{Json, Router};
use tokio::sync::watch;
use crate::AppState;
use crate::web::routes::routes as web_routes;
// use crate::core::queue::send_to_queue;
pub async fn create_http(state: AppState, host: String) -> anyhow::Result<()> {
pub async fn create_http(
state: AppState,
host: String,
mut shutdown_rx: watch::Receiver<bool>,
) -> anyhow::Result<()> {
let app = Router::new()
// .route("/message", post(send_to_queue))
.route("/health", get(health))
.merge(web_routes())
.with_state(Arc::new(state));
let listener = tokio::net::TcpListener::bind(host.clone()).await?;
let listener = tokio::net::TcpListener::bind(&host).await.map_err(|err| {
if err.kind() == std::io::ErrorKind::AddrInUse {
anyhow!(
"HTTP 地址已被占用: {}。请停止已有服务,或修改 HOST 后重试",
host
)
} else {
anyhow!("HTTP 地址监听失败: {}: {}", host, err)
}
})?;
println!("服务启动在:http://{}", listener.local_addr()?);
axum::serve(listener, app).await?;
axum::serve(listener, app)
.with_graceful_shutdown(async move {
while shutdown_rx.changed().await.is_ok() {
if *shutdown_rx.borrow() {
break;
}
}
})
.await?;
Ok(())
}
+122 -18
View File
@@ -3,41 +3,59 @@ mod core;
mod http;
mod model;
mod repository;
mod web;
use clap::Parser;
use common::queue::QueueMessage;
use core::control::{ControlResponse, RuntimePaths, request_control, start_control_server};
use core::queue::create_queue;
use core::runtime::RuntimeState;
use core::supervisor::{
cleanup_runtime_files, print_logs, restart_service, start_service, status_service,
stop_service, write_pid_file,
};
use http::app::create_http;
use sqlx::{PgPool, postgres::PgPoolOptions};
use std::env;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::sync::{mpsc, watch};
use crate::channel::wechat::manager::{
delete_account, init_wechat, list_accounts, login_user, rename_account,
};
use crate::core::cli::{Channels, Cli, Commands, WechatAction};
use crate::core::cli::{Channels, Cli, Commands, ServiceAction, WechatAction};
#[derive(Clone)]
pub struct AppState {
pub sender: mpsc::Sender<QueueMessage>,
pub db: PgPool,
pub runtime: Arc<RuntimeState>,
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
dotenvy::dotenv().ok();
let db_url = env::var("DATABASE_URL").expect("必须设置DATABASE_URL");
let pool = PgPoolOptions::new()
.max_connections(10)
.connect(&db_url)
.await?;
init_logging();
let cli = Cli::parse();
match cli.command {
Commands::Service => {
service(pool).await.unwrap();
Commands::Service { command } => {
let action = command.unwrap_or(ServiceAction::Run);
match action {
ServiceAction::Run => {
let pool = connect_db().await?;
run_service(pool).await?;
}
Commands::Channel { command } => match command {
ServiceAction::Start => start_service().await?,
ServiceAction::Stop => stop_service().await?,
ServiceAction::Restart => restart_service().await?,
ServiceAction::Status => status_service().await?,
ServiceAction::Logs { lines } => print_logs(lines)?,
}
}
Commands::Channel { command } => {
let pool = connect_db().await?;
match command {
Channels::Wechat { command } => match command {
WechatAction::Login => login_user(&pool).await,
WechatAction::List => list_accounts(&pool).await,
@@ -46,28 +64,114 @@ async fn main() -> anyhow::Result<()> {
rename_account(&pool, account, name).await
}
},
},
}
}
}
Ok(())
}
async fn service(pool: PgPool) -> anyhow::Result<()> {
async fn connect_db() -> anyhow::Result<PgPool> {
let db_url = env::var("DATABASE_URL").map_err(|_| anyhow::anyhow!("必须设置DATABASE_URL"))?;
let pool = PgPoolOptions::new()
.max_connections(10)
.connect(&db_url)
.await?;
Ok(pool)
}
fn init_logging() {
let filter = tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));
let _ = tracing_subscriber::fmt().with_env_filter(filter).try_init();
}
async fn run_service(pool: PgPool) -> anyhow::Result<()> {
let name: String = env::var("NAME").unwrap_or_else(|_| "ias".to_string());
let host = env::var("HOST").unwrap_or_else(|_| "0.0.0.0:9003".to_string());
let paths = RuntimePaths::new();
paths.ensure_dir()?;
if let Ok(ControlResponse::Status { snapshot }) = request_control(&paths, "status").await {
anyhow::bail!("服务已在运行,PID: {}", snapshot.pid);
}
println!("{}已启动...", name);
let runtime = Arc::new(RuntimeState::new(host.clone()));
let (sender, receiver) = mpsc::channel::<QueueMessage>(100);
let state = AppState {
sender: sender.clone(),
db: pool,
runtime: runtime.clone(),
};
let manger = init_wechat(Some(state.clone())).await.unwrap();
create_queue(sender.clone(), receiver, manger)
let manger = init_wechat(Some(state.clone()))
.await
.unwrap();
.ok_or_else(|| anyhow::anyhow!("微信初始化失败:未加载到微信账号"))?;
let queue_task = create_queue(sender.clone(), receiver, manger).await?;
runtime.set_queue_alive(true).await;
// 启动http服务器,务必放在最下面执行
create_http(state.clone(), host.clone()).await.unwrap();
Ok(())
let (shutdown_tx, shutdown_rx) = watch::channel(false);
let mut control_task = tokio::spawn(start_control_server(
runtime.clone(),
paths.clone(),
shutdown_tx.clone(),
shutdown_tx.subscribe(),
));
tokio::select! {
result = &mut control_task => {
result??;
anyhow::bail!("服务控制通道已退出");
}
_ = tokio::time::sleep(std::time::Duration::from_millis(50)) => {}
}
write_pid_file(&paths)?;
let signal_task = tokio::spawn(wait_for_shutdown_signal(shutdown_tx.clone()));
runtime.mark_running().await;
runtime.write_snapshot(&paths.state_file).await?;
let result = create_http(state.clone(), host.clone(), shutdown_rx).await;
runtime.mark_stopping().await;
runtime.write_snapshot(&paths.state_file).await?;
queue_task.abort();
control_task.abort();
signal_task.abort();
if let Err(err) = &result {
runtime.mark_failed(err.to_string()).await;
runtime.write_snapshot(&paths.state_file).await?;
} else {
runtime.mark_stopped().await;
runtime.write_snapshot(&paths.state_file).await?;
}
cleanup_runtime_files(&paths);
result
}
async fn wait_for_shutdown_signal(shutdown_tx: watch::Sender<bool>) {
#[cfg(unix)]
{
use tokio::signal::unix::{SignalKind, signal};
let mut terminate = match signal(SignalKind::terminate()) {
Ok(signal) => signal,
Err(err) => {
eprintln!("监听 SIGTERM 失败: {err}");
return;
}
};
tokio::select! {
_ = tokio::signal::ctrl_c() => {}
_ = terminate.recv() => {}
}
}
#[cfg(not(unix))]
{
let _ = tokio::signal::ctrl_c().await;
}
let _ = shutdown_tx.send(true);
}
+3
View File
@@ -0,0 +1,3 @@
pub mod model;
pub mod repository;
pub mod routes;
+59
View File
@@ -0,0 +1,59 @@
use chrono::NaiveDateTime;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sqlx::prelude::FromRow;
#[derive(Debug, Clone, Serialize, FromRow)]
pub struct WebPage {
pub id: i64,
pub slug: String,
pub page_type: String,
pub title: String,
pub summary: Option<String>,
pub content: String,
pub source_label: Option<String>,
pub metadata: String,
pub created_at: Option<NaiveDateTime>,
}
#[derive(Debug, Clone)]
pub struct NewWebPage {
pub slug: String,
pub page_type: String,
pub title: String,
pub summary: Option<String>,
pub content: String,
pub source_label: Option<String>,
pub metadata: String,
}
#[derive(Debug, Deserialize)]
pub struct CreateWebPageRequest {
pub page_type: Option<String>,
pub title: String,
pub summary: Option<String>,
pub content: String,
pub source_label: Option<String>,
pub metadata: Option<Value>,
}
#[derive(Debug, Deserialize)]
pub struct CreateMailPageRequest {
pub from: String,
pub subject: String,
pub reminder: Option<String>,
pub content: String,
pub received_at: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct CreateWebPageResponse {
pub success: bool,
pub id: i64,
pub slug: String,
pub page_type: String,
pub title: String,
pub summary: Option<String>,
pub path: String,
pub url: String,
}
+56
View File
@@ -0,0 +1,56 @@
use sqlx::PgPool;
use crate::web::model::{NewWebPage, WebPage};
pub struct WebPageRepository;
impl WebPageRepository {
pub async fn create(pool: &PgPool, page: NewWebPage) -> anyhow::Result<WebPage> {
let row = sqlx::query_as::<_, WebPage>(
r#"
INSERT INTO ias_web_pages (
slug,
page_type,
title,
summary,
content,
source_label,
metadata
)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id, slug, page_type, title, summary, content, source_label, metadata, created_at
"#,
)
.bind(page.slug)
.bind(page.page_type)
.bind(page.title)
.bind(page.summary)
.bind(page.content)
.bind(page.source_label)
.bind(page.metadata)
.fetch_one(pool)
.await?;
Ok(row)
}
pub async fn find_by_type_and_slug(
pool: &PgPool,
page_type: &str,
slug: &str,
) -> anyhow::Result<Option<WebPage>> {
let row = sqlx::query_as::<_, WebPage>(
r#"
SELECT id, slug, page_type, title, summary, content, source_label, metadata, created_at
FROM ias_web_pages
WHERE page_type = $1 AND slug = $2
"#,
)
.bind(page_type)
.bind(slug)
.fetch_optional(pool)
.await?;
Ok(row)
}
}
+479
View File
@@ -0,0 +1,479 @@
use std::sync::Arc;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::Html;
use axum::routing::{get, post};
use axum::{Json, Router};
use serde_json::{Value, json};
use uuid::Uuid;
use crate::AppState;
use crate::web::model::{
CreateMailPageRequest, CreateWebPageRequest, CreateWebPageResponse, NewWebPage, WebPage,
};
use crate::web::repository::WebPageRepository;
type ApiError = (StatusCode, Json<Value>);
type HtmlError = (StatusCode, Html<String>);
pub fn routes() -> Router<Arc<AppState>> {
Router::new()
.route("/web/pages", post(create_page))
.route("/web/mail", post(create_mail_page))
.route("/mail/{slug}", get(show_mail_page))
.route("/web/{page_type}/{slug}", get(show_typed_page))
}
async fn create_page(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
Json(request): Json<CreateWebPageRequest>,
) -> Result<Json<CreateWebPageResponse>, ApiError> {
let page_type = normalize_key(request.page_type.as_deref().unwrap_or("page"), "page")?;
let title = required_text(&request.title, "title")?;
let content = required_text(&request.content, "content")?;
let metadata = request.metadata.unwrap_or_else(|| json!({})).to_string();
let page = NewWebPage {
slug: generate_slug(),
page_type,
title,
summary: optional_text(request.summary),
content,
source_label: optional_text(request.source_label),
metadata,
};
let page = WebPageRepository::create(&state.db, page)
.await
.map_err(api_internal_error)?;
Ok(Json(response_for_page(&page, &headers)))
}
async fn create_mail_page(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
Json(request): Json<CreateMailPageRequest>,
) -> Result<Json<CreateWebPageResponse>, ApiError> {
let from = required_text(&request.from, "from")?;
let subject = required_text(&request.subject, "subject")?;
let content = required_text(&request.content, "content")?;
let reminder = optional_text(request.reminder);
let received_at = optional_text(request.received_at);
let summary = reminder
.clone()
.unwrap_or_else(|| format!("您收到了一份来自 {from} 的邮件提醒"));
let metadata = json!({
"from": from,
"subject": subject,
"reminder": reminder,
"received_at": received_at,
})
.to_string();
let page = NewWebPage {
slug: generate_slug(),
page_type: "mail".to_string(),
title: subject,
summary: Some(summary),
content,
source_label: Some(from),
metadata,
};
let page = WebPageRepository::create(&state.db, page)
.await
.map_err(api_internal_error)?;
Ok(Json(response_for_page(&page, &headers)))
}
async fn show_mail_page(
State(state): State<Arc<AppState>>,
Path(slug): Path<String>,
) -> Result<Html<String>, HtmlError> {
let page = WebPageRepository::find_by_type_and_slug(&state.db, "mail", &slug)
.await
.map_err(html_internal_error)?
.ok_or_else(html_not_found)?;
Ok(Html(render_page(&page)))
}
async fn show_typed_page(
State(state): State<Arc<AppState>>,
Path((page_type, slug)): Path<(String, String)>,
) -> Result<Html<String>, HtmlError> {
let page_type = normalize_key(&page_type, "page").map_err(|_| html_not_found())?;
let page = WebPageRepository::find_by_type_and_slug(&state.db, &page_type, &slug)
.await
.map_err(html_internal_error)?
.ok_or_else(html_not_found)?;
Ok(Html(render_page(&page)))
}
fn response_for_page(page: &WebPage, headers: &HeaderMap) -> CreateWebPageResponse {
let path = public_path(page);
let url = format!("{}{}", request_base_url(headers), path);
CreateWebPageResponse {
success: true,
id: page.id,
slug: page.slug.clone(),
page_type: page.page_type.clone(),
title: page.title.clone(),
summary: page.summary.clone(),
path,
url,
}
}
fn public_path(page: &WebPage) -> String {
if page.page_type == "mail" {
format!("/mail/{}", page.slug)
} else {
format!("/web/{}/{}", page.page_type, page.slug)
}
}
fn request_base_url(headers: &HeaderMap) -> String {
if let Ok(base_url) = std::env::var("WEB_BASE_URL")
&& !base_url.trim().is_empty()
{
return base_url.trim().trim_end_matches('/').to_string();
}
let host = headers
.get("host")
.and_then(|value| value.to_str().ok())
.filter(|value| !value.trim().is_empty())
.unwrap_or("127.0.0.1:9003");
let proto = headers
.get("x-forwarded-proto")
.and_then(|value| value.to_str().ok())
.filter(|value| !value.trim().is_empty())
.unwrap_or("http");
format!("{}://{}", proto.trim(), host.trim())
}
fn generate_slug() -> String {
Uuid::new_v4().simple().to_string()
}
fn required_text(value: &str, field: &str) -> Result<String, ApiError> {
let value = value.trim();
if value.is_empty() {
return Err(api_bad_request(format!("{field} 不能为空")));
}
Ok(value.to_string())
}
fn optional_text(value: Option<String>) -> Option<String> {
value
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
}
fn normalize_key(value: &str, field: &str) -> Result<String, ApiError> {
let normalized = value.trim().to_ascii_lowercase();
if normalized.is_empty() || normalized.len() > 64 {
return Err(api_bad_request(format!("{field} 长度必须在 1 到 64 之间")));
}
if !normalized
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_')
{
return Err(api_bad_request(format!(
"{field} 只能包含字母、数字、下划线或连字符"
)));
}
Ok(normalized)
}
fn render_page(page: &WebPage) -> String {
let summary = page
.summary
.as_deref()
.map(|summary| format!(r#"<p class="summary">{}</p>"#, escape_html(summary)))
.unwrap_or_default();
let source = page
.source_label
.as_deref()
.map(|source| {
format!(
r#"<span class="meta-item">来源:{}</span>"#,
escape_html(source)
)
})
.unwrap_or_default();
let created_at = page
.created_at
.map(|time| time.format("%Y-%m-%d %H:%M:%S").to_string())
.unwrap_or_else(|| "-".to_string());
let metadata = render_metadata(&page.metadata);
let content = render_text_content(&page.content);
let page_type = escape_html(&page.page_type);
format!(
r#"<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{title}</title>
<style>
:root {{
color-scheme: light;
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
background: #f6f7f9;
color: #1f2328;
}}
body {{
margin: 0;
min-height: 100vh;
background: #f6f7f9;
}}
main {{
width: min(880px, calc(100% - 32px));
margin: 0 auto;
padding: 40px 0 56px;
}}
.eyebrow {{
display: inline-flex;
align-items: center;
min-height: 28px;
padding: 0 10px;
border: 1px solid #d8dee4;
border-radius: 999px;
background: #ffffff;
color: #57606a;
font-size: 13px;
line-height: 1;
}}
h1 {{
margin: 18px 0 10px;
font-size: clamp(28px, 4vw, 42px);
line-height: 1.15;
font-weight: 720;
letter-spacing: 0;
}}
.summary {{
margin: 0 0 18px;
color: #57606a;
font-size: 17px;
line-height: 1.65;
}}
.meta {{
display: flex;
flex-wrap: wrap;
gap: 8px 16px;
margin: 0 0 28px;
color: #6e7781;
font-size: 14px;
}}
.content {{
border: 1px solid #d8dee4;
border-radius: 8px;
background: #ffffff;
padding: 24px;
font-size: 16px;
line-height: 1.8;
white-space: normal;
overflow-wrap: anywhere;
}}
.details {{
margin: 24px 0;
border-top: 1px solid #d8dee4;
border-bottom: 1px solid #d8dee4;
padding: 14px 0;
}}
dl {{
display: grid;
grid-template-columns: minmax(88px, 160px) 1fr;
gap: 10px 16px;
margin: 0;
font-size: 14px;
line-height: 1.6;
}}
dt {{
color: #6e7781;
}}
dd {{
margin: 0;
color: #24292f;
overflow-wrap: anywhere;
}}
@media (max-width: 640px) {{
main {{
width: min(100% - 24px, 880px);
padding-top: 28px;
}}
.content {{
padding: 18px;
}}
dl {{
grid-template-columns: 1fr;
gap: 2px 0;
}}
}}
</style>
</head>
<body>
<main>
<span class="eyebrow">{page_type}</span>
<h1>{title}</h1>
{summary}
<div class="meta">{source}<span class="meta-item">创建时间:{created_at}</span></div>
{metadata}
<article class="content">{content}</article>
</main>
</body>
</html>"#,
title = escape_html(&page.title),
summary = summary,
source = source,
created_at = escape_html(&created_at),
metadata = metadata,
content = content,
page_type = page_type
)
}
fn render_metadata(metadata: &str) -> String {
let Ok(Value::Object(values)) = serde_json::from_str::<Value>(metadata) else {
return String::new();
};
let items = values
.into_iter()
.filter_map(|(key, value)| {
let value = metadata_value_to_text(value)?;
Some(format!(
"<dt>{}</dt><dd>{}</dd>",
escape_html(metadata_label(&key)),
escape_html(&value)
))
})
.collect::<Vec<_>>()
.join("");
if items.is_empty() {
String::new()
} else {
format!(r#"<section class="details"><dl>{items}</dl></section>"#)
}
}
fn metadata_label(key: &str) -> &str {
match key {
"from" => "发件人",
"subject" => "主题",
"reminder" => "提醒",
"received_at" => "收件时间",
other => other,
}
}
fn metadata_value_to_text(value: Value) -> Option<String> {
match value {
Value::Null => None,
Value::String(value) => {
let value = value.trim().to_string();
(!value.is_empty()).then_some(value)
}
Value::Array(values) if values.is_empty() => None,
Value::Object(values) if values.is_empty() => None,
other => Some(other.to_string()),
}
}
fn render_text_content(content: &str) -> String {
content
.lines()
.map(|line| {
if line.trim().is_empty() {
"<br>".to_string()
} else {
escape_html(line)
}
})
.collect::<Vec<_>>()
.join("<br>\n")
}
fn escape_html(value: &str) -> String {
let mut escaped = String::with_capacity(value.len());
for ch in value.chars() {
match ch {
'&' => escaped.push_str("&amp;"),
'<' => escaped.push_str("&lt;"),
'>' => escaped.push_str("&gt;"),
'"' => escaped.push_str("&quot;"),
'\'' => escaped.push_str("&#39;"),
_ => escaped.push(ch),
}
}
escaped
}
fn api_bad_request(message: impl Into<String>) -> ApiError {
(
StatusCode::BAD_REQUEST,
Json(json!({
"success": false,
"message": message.into(),
})),
)
}
fn api_internal_error(error: anyhow::Error) -> ApiError {
eprintln!("创建 web 页面失败: {error}");
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({
"success": false,
"message": "创建 web 页面失败",
})),
)
}
fn html_not_found() -> HtmlError {
(
StatusCode::NOT_FOUND,
Html(render_error_page("页面不存在", "没有找到对应的内容页面")),
)
}
fn html_internal_error(error: anyhow::Error) -> HtmlError {
eprintln!("读取 web 页面失败: {error}");
(
StatusCode::INTERNAL_SERVER_ERROR,
Html(render_error_page("页面暂时不可用", "读取页面内容失败")),
)
}
fn render_error_page(title: &str, message: &str) -> String {
format!(
r#"<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{title}</title>
</head>
<body style="margin:0;font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;background:#f6f7f9;color:#1f2328;">
<main style="width:min(760px,calc(100% - 32px));margin:0 auto;padding:48px 0;">
<h1 style="font-size:32px;line-height:1.2;margin:0 0 12px;">{title}</h1>
<p style="font-size:16px;line-height:1.7;margin:0;color:#57606a;">{message}</p>
</main>
</body>
</html>"#,
title = escape_html(title),
message = escape_html(message)
)
}
+47
View File
@@ -0,0 +1,47 @@
name: anysearch_search
desc: 使用 AnySearch 统一搜索 API 搜索互联网、文档、学术、代码等信息,返回结果列表和搜索元数据。默认匿名访问。
params:
- name: query
type: string
required: true
description: 搜索关键词或问题
- name: max_results
type: integer
required: false
default: "10"
description: 返回结果数量,范围 1-20,默认 10
steps:
- url: https://api.anysearch.com/v1/search
method: POST
headers:
Content-Type: application/json
body: |
{
"query": "${param:query}",
"max_results": ${param:max_results},
"format": "json"
}
timeout: 30
var:
- name: code
extract: code
default: "-1"
- name: message
extract: message
default: ""
- name: request_id
extract: request_id
default: ""
- name: results
extract: data.results
default: "[]"
- name: total_results
extract: data.metadata.total_results
default: "0"
- name: search_time_ms
extract: data.metadata.search_time_ms
default: "0"
result: |
AnySearch code=${step0:code}, message=${step0:message}, request_id=${step0:request_id}
total_results=${step0:total_results}, search_time_ms=${step0:search_time_ms}
results=${step0:results}
+21
View File
@@ -0,0 +1,21 @@
name: get_weather
desc: 查询指定城市实时天气
params:
- name: city
type: string
required: true
description: 城市名称
env:
- WEATHER_API_KEY
steps:
- url: https://api.weather.com/current?q=${param:city}&key=${env:WEATHER_API_KEY}
method: GET
var:
- name: city_name
extract: location.name
- name: temp
extract: current.temp_c
- name: condition
extract: current.condition.text
result: |
${step0:city_name} 当前温度 ${step0:temp}°C${step0:condition}
+63 -5
View File
@@ -63,7 +63,7 @@ impl WeChatClient {
.unwrap_or_else(|_| "https://ilinkai.weixin.qq.com".to_string())
}
/// 版本号编码:0x00MMNNPP
const CLIENT_VERSION: u32 = 0x000100_00; // 1.0.0
const CLIENT_VERSION: u32 = 0x0001_0000; // 1.0.0
pub fn new(base_url: Option<String>) -> Self {
// 优先级: 显式参数 > WEIXIN_BASE_URL 环境变量 > 默认值
@@ -363,14 +363,72 @@ impl WeChatClient {
context_token: Option<&str>,
) -> Result<String, String> {
let client_id = uuid::Uuid::new_v4().to_string();
self.send_text_with_client_id(to, text, context_token, &client_id)
.await
}
/// 发送“输入中”状态,返回用于最终完成消息的 client_id。
pub async fn send_typing(
&self,
to: &str,
context_token: Option<&str>,
) -> Result<String, String> {
let client_id = uuid::Uuid::new_v4().to_string();
self.send_typing_with_client_id(to, context_token, &client_id)
.await?;
Ok(client_id.to_string())
}
/// 用已有 client_id 刷新“输入中”状态。
pub async fn send_typing_with_client_id(
&self,
to: &str,
context_token: Option<&str>,
client_id: &str,
) -> Result<String, String> {
self.send_bot_message(
to,
client_id,
WeixinMessage::STATE_TYPING,
None,
context_token,
)
.await
}
/// 使用已有 client_id 完成文本消息。
pub async fn send_text_with_client_id(
&self,
to: &str,
text: &str,
context_token: Option<&str>,
client_id: &str,
) -> Result<String, String> {
self.send_bot_message(
to,
client_id,
WeixinMessage::STATE_FINISH,
Some(vec![MessageItem::text(text)]),
context_token,
)
.await
}
async fn send_bot_message(
&self,
to: &str,
client_id: &str,
message_state: i32,
item_list: Option<Vec<MessageItem>>,
context_token: Option<&str>,
) -> Result<String, String> {
let msg = WeixinMessage {
from_user_id: Some(String::new()),
to_user_id: Some(to.to_string()),
client_id: Some(client_id.clone()),
client_id: Some(client_id.to_string()),
msg_type: Some(WeixinMessage::TYPE_BOT),
message_state: Some(2), // FINISH
item_list: Some(vec![MessageItem::text(text)]),
message_state: Some(message_state),
item_list,
context_token: context_token.map(|s| s.to_string()),
..Default::default()
};
@@ -387,7 +445,7 @@ impl WeChatClient {
.await
.map_err(|e| format!("发送消息网络错误: {}", e))?;
Ok(client_id)
Ok(client_id.to_string())
}
/// 设置 auth 状态(从持久化恢复)
+56 -2
View File
@@ -47,7 +47,7 @@ pub enum WeChatEvent {
/// 收到一条微信消息。
Message {
account_id: String,
message: WeixinMessage,
message: Box<WeixinMessage>,
updates_buf: String,
},
/// 某个账号的轮询发生错误。管理器会继续重试,除非任务被停止。
@@ -145,7 +145,7 @@ impl WeChatMultiAccountManager {
for message in resp.msgs {
let event = WeChatEvent::Message {
account_id: account_id.clone(),
message,
message: Box::new(message),
updates_buf: updates_buf.clone(),
};
if event_tx.send(event).await.is_err() {
@@ -230,6 +230,60 @@ impl WeChatMultiAccountManager {
runtime.client.send_text(to, text, context_token).await
}
/// 使用指定账号发送输入中状态,返回后续完成消息复用的 client_id。
pub async fn send_typing(
&self,
account_id: &str,
to: &str,
context_token: Option<&str>,
) -> Result<String, String> {
let runtime = self
.accounts
.get(account_id)
.ok_or_else(|| format!("微信账号未启动: {}", account_id))?;
runtime.client.send_typing(to, context_token).await
}
/// 使用指定账号刷新输入中状态。
pub async fn send_typing_with_client_id(
&self,
account_id: &str,
to: &str,
context_token: Option<&str>,
client_id: &str,
) -> Result<String, String> {
let runtime = self
.accounts
.get(account_id)
.ok_or_else(|| format!("微信账号未启动: {}", account_id))?;
runtime
.client
.send_typing_with_client_id(to, context_token, client_id)
.await
}
/// 使用指定账号完成已有输入中消息。
pub async fn send_text_with_client_id(
&self,
account_id: &str,
to: &str,
text: &str,
context_token: Option<&str>,
client_id: &str,
) -> Result<String, String> {
let runtime = self
.accounts
.get(account_id)
.ok_or_else(|| format!("微信账号未启动: {}", account_id))?;
runtime
.client
.send_text_with_client_id(to, text, context_token, client_id)
.await
}
}
impl Drop for WeChatMultiAccountManager {
+8
View File
@@ -30,6 +30,12 @@ impl BaseInfo {
}
}
impl Default for BaseInfo {
fn default() -> Self {
Self::new()
}
}
// ─── 消息类型 ───
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -232,6 +238,8 @@ impl WeixinMessage {
pub const TYPE_NONE: i32 = 0;
pub const TYPE_USER: i32 = 1;
pub const TYPE_BOT: i32 = 2;
pub const STATE_TYPING: i32 = 1;
pub const STATE_FINISH: i32 = 2;
/// 获取文本内容(如果有纯文本 item)
pub fn text_content(&self) -> Option<&str> {