feat: 模块统一 ias- 前缀、独立版本与内网地址支持 (0.1.7)

- 所有 Rust crate 统一加 ias- 前缀:ai → ias-ai、common → ias-common、context → ias-context、mail → ias-mail、service → ias-service、wechat → ias-wechat
- 新增 ias-main 模块承载 CLI 入口和二进制打包
- 各模块开始独立维护版本号,ias-main 0.1.7 代表整体版本
- 新增 IA_WEB_INTERNAL_URL 环境变量,版本通知同时发送公网和内网链接
- 将已有升级日志翻译为中文
- ias_upgrade_logs 新增 modules JSONB 字段
- 新增 VERSION 文件和 CHANGELOG.md
- 新增 AGENTS.md 项目规则文件
This commit is contained in:
2026-07-06 17:57:48 +08:00
parent ed7f89d44f
commit 25dd0daa8b
91 changed files with 5546 additions and 777 deletions
+156
View File
@@ -0,0 +1,156 @@
use regex::Regex;
use std::env;
use ias_common::model::ai::{
ChatCompletionRequestBody, ChatCompletionRequestExtra, ChatCompletionRequestMessage,
ChatCompletionRequestThinking, ChatCompletionResponseBody,
};
use ias_common::queue::{AssistantMessage, ChannelMessage, QueueMessage};
use crate::toolkit::{call_tools, get_tool, tool_notice_text};
pub async fn send_message(
sender: tokio::sync::mpsc::Sender<QueueMessage>,
messages: Vec<ChatCompletionRequestMessage>,
channel: ChannelMessage,
) -> anyhow::Result<()> {
let result = request_llm(messages).await;
match result {
Ok(result) => {
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();
sender
.send(QueueMessage::Aissitant(
channel.clone(),
AssistantMessage {
reasoning: Some(messsage_reasoning),
content: messsage_content,
tool_calls,
},
))
.await
.unwrap();
match choice.message.tool_calls {
Some(tool_calls) => {
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!("没有调用工具");
}
}
}
}
Err(err) => {
eprintln!("发送错误:{}", err);
}
}
Ok(())
}
pub async fn request_llm(
messages: Vec<ChatCompletionRequestMessage>,
) -> anyhow::Result<Option<ChatCompletionResponseBody>> {
let client = reqwest::Client::new();
let base_url =
env::var("PROVIDER_BASE_URL").map_err(|_| anyhow::anyhow!("必须设置PROVIDER_BASE_URL"))?;
let chat_re = Regex::new(r"(?i)(?:^|/)(?:v\d+/)?chat/completions/?$")?;
let version_re = Regex::new(r"(?i)(?:^|/)v\d+$")?;
let request_url = if chat_re.is_match(&base_url) {
base_url.to_string()
} else if version_re.is_match(&base_url) {
format!("{}/chat/completions", base_url)
} else {
format!("{}/v1/chat/completions", base_url)
};
let api_key =
env::var("PROVIDER_API_KEY").map_err(|_| anyhow::anyhow!("必须设置PROVIDER_API_KEY"))?;
let model_name = env::var("PROVIDER_MODEL").unwrap_or_else(|_| "deepseek-v4-flash".to_string());
let body = ChatCompletionRequestBody {
model: model_name,
messages,
stream: false,
tools: get_tool(),
extra_body: Some(ChatCompletionRequestExtra {
thinking: ChatCompletionRequestThinking {
thinking_type: "enabled".to_string(),
},
}),
};
let resp = client
.post(request_url)
.bearer_auth(api_key)
.json(&body)
.send()
.await;
match resp {
Ok(resp) => {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
// println!("response:\n{}", text);
if status.is_success() {
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);
eprintln!("body: {}", text);
}
}
Err(err) => {
eprintln!("请求发送失败: {}", err);
if err.is_timeout() {
eprintln!("原因: 请求超时");
}
if err.is_connect() {
eprintln!("原因: 连接失败");
}
if let Some(url) = err.url() {
eprintln!("url: {}", url);
}
}
}
Ok(None)
}
+4
View File
@@ -0,0 +1,4 @@
pub mod core;
pub mod toolkit;
pub mod tools;
pub mod workflow;
+311
View File
@@ -0,0 +1,311 @@
use crate::{
tools::date_time::get_system_date_time,
tools::memory::{delete_memory, query_memory, remember_memory, update_memory},
tools::web::create_web_page,
workflow::core::{call_flow, load_tools_define},
};
use ias_common::{
model::ai::{ChatCompletionFunction, ChatCompletionToolCall},
queue::{ChannelMessage, QueueMessage},
};
use serde_json::Value;
pub async fn call_tools(
channel: ChannelMessage,
tools: Vec<ChatCompletionToolCall>,
) -> 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(&channel, &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(),
tool_name,
tool_result,
)))
}
async fn execute_tool(
channel: &ChannelMessage,
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,
"remember_memory" => remember_memory(channel, &function.arguments).await,
"query_memory" => query_memory(channel, &function.arguments).await,
"update_memory" => update_memory(channel, &function.arguments).await,
"delete_memory" => delete_memory(channel, &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,
}
}
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> {
vec![
serde_json::json!({
"type": "function",
"function": {
"name": "query_date_time",
"description": "查看当前时间",
"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": "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": "remember_memory",
"description": "写入长期记忆。仅在用户明确要求记住,或信息对未来长期有稳定价值时使用;临时任务、一次性闲聊不要写入。记录结果会返回 id、创建时间和更新时间。",
"parameters": {
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "要长期保存的记忆内容,写成完整、可独立理解的一句话或短段落。"
},
"metadata": {
"type": "object",
"description": "可选结构化元数据,例如 {\"type\":\"preference\",\"source\":\"user\"}。"
},
"text": {
"type": "string",
"description": "可选。调用工具前发送给用户的提醒消息;为空则不发送。"
}
},
"required": ["content"]
}
}
}),
serde_json::json!({
"type": "function",
"function": {
"name": "query_memory",
"description": "查询当前用户的长期记忆。回答涉及用户偏好、长期事实、历史稳定信息时可以先查。返回包含 id、创建时间、更新时间和内容。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "可选关键词;为空时返回最近更新的长期记忆。"
},
"limit": {
"type": "integer",
"description": "返回数量,默认 20,最大 50。"
},
"text": {
"type": "string",
"description": "可选。调用工具前发送给用户的提醒消息;为空则不发送。"
}
},
"required": []
}
}
}),
serde_json::json!({
"type": "function",
"function": {
"name": "update_memory",
"description": "修改一条长期记忆。必须先知道要修改的记忆 id;修改后会返回新的更新时间。",
"parameters": {
"type": "object",
"properties": {
"id": {
"type": "integer",
"description": "长期记忆 id。"
},
"content": {
"type": "string",
"description": "修改后的完整记忆内容。"
},
"metadata": {
"type": "object",
"description": "可选结构化元数据;传入后会覆盖原 metadata。"
},
"text": {
"type": "string",
"description": "可选。调用工具前发送给用户的提醒消息;为空则不发送。"
}
},
"required": ["id", "content"]
}
}
}),
serde_json::json!({
"type": "function",
"function": {
"name": "delete_memory",
"description": "删除一条长期记忆。必须先知道要删除的记忆 id。",
"parameters": {
"type": "object",
"properties": {
"id": {
"type": "integer",
"description": "长期记忆 id。"
},
"text": {
"type": "string",
"description": "可选。调用工具前发送给用户的提醒消息;为空则不发送。"
}
},
"required": ["id"]
}
}
}),
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))
}
+241
View File
@@ -0,0 +1,241 @@
use anyhow::{Context, anyhow};
use ias_common::queue::ChannelMessage;
use serde::Deserialize;
use serde_json::{Value, json};
#[derive(Debug, Deserialize)]
struct RememberMemoryArgs {
content: String,
metadata: Option<Value>,
#[allow(dead_code)]
text: Option<String>,
}
#[derive(Debug, Deserialize)]
struct QueryMemoryArgs {
query: Option<String>,
limit: Option<i64>,
#[allow(dead_code)]
text: Option<String>,
}
#[derive(Debug, Deserialize)]
struct UpdateMemoryArgs {
id: i64,
content: String,
metadata: Option<Value>,
#[allow(dead_code)]
text: Option<String>,
}
#[derive(Debug, Deserialize)]
struct DeleteMemoryArgs {
id: i64,
#[allow(dead_code)]
text: Option<String>,
}
pub async fn remember_memory(channel: &ChannelMessage, arguments: &str) -> String {
match remember_memory_inner(channel, arguments).await {
Ok(result) => result,
Err(error) => format!("工具调用失败: 记录长期记忆失败: {error}"),
}
}
pub async fn query_memory(channel: &ChannelMessage, arguments: &str) -> String {
match query_memory_inner(channel, arguments).await {
Ok(result) => result,
Err(error) => format!("工具调用失败: 查询长期记忆失败: {error}"),
}
}
pub async fn update_memory(channel: &ChannelMessage, arguments: &str) -> String {
match update_memory_inner(channel, arguments).await {
Ok(result) => result,
Err(error) => format!("工具调用失败: 更新长期记忆失败: {error}"),
}
}
pub async fn delete_memory(channel: &ChannelMessage, arguments: &str) -> String {
match delete_memory_inner(channel, arguments).await {
Ok(result) => result,
Err(error) => format!("工具调用失败: 删除长期记忆失败: {error}"),
}
}
async fn remember_memory_inner(
channel: &ChannelMessage,
arguments: &str,
) -> anyhow::Result<String> {
let args: RememberMemoryArgs =
serde_json::from_str(arguments).context("解析 remember_memory 参数失败")?;
let content = required_text(&args.content, "content")?;
let payload = json_with_identity(
channel,
json!({
"content": content,
"metadata": args.metadata.unwrap_or_else(|| json!({})),
}),
);
let value = request_json("POST", "/context/memories", payload).await?;
let memory = value
.get("memory")
.ok_or_else(|| anyhow!("接口响应缺少 memory 字段"))?;
Ok(format!(
"长期记忆已记录:\n{}",
format_memory(memory).unwrap_or_else(|| memory.to_string())
))
}
async fn query_memory_inner(channel: &ChannelMessage, arguments: &str) -> anyhow::Result<String> {
let args: QueryMemoryArgs =
serde_json::from_str(arguments).context("解析 query_memory 参数失败")?;
let payload = json_with_identity(
channel,
json!({
"query": args.query,
"limit": args.limit.unwrap_or(20),
}),
);
let value = request_json("POST", "/context/memories/search", payload).await?;
let memories = value
.get("memories")
.and_then(Value::as_array)
.ok_or_else(|| anyhow!("接口响应缺少 memories 字段"))?;
if memories.is_empty() {
return Ok("没有查到匹配的长期记忆。".to_string());
}
let lines = memories
.iter()
.filter_map(format_memory)
.collect::<Vec<_>>()
.join("\n");
Ok(format!("查到 {} 条长期记忆:\n{}", memories.len(), lines))
}
async fn update_memory_inner(channel: &ChannelMessage, arguments: &str) -> anyhow::Result<String> {
let args: UpdateMemoryArgs =
serde_json::from_str(arguments).context("解析 update_memory 参数失败")?;
let content = required_text(&args.content, "content")?;
let payload = json_with_identity(
channel,
json!({
"content": content,
"metadata": args.metadata,
}),
);
let path = format!("/context/memories/{}", args.id);
let value = request_json("PATCH", &path, payload).await?;
let memory = value
.get("memory")
.ok_or_else(|| anyhow!("接口响应缺少 memory 字段"))?;
Ok(format!(
"长期记忆已更新:\n{}",
format_memory(memory).unwrap_or_else(|| memory.to_string())
))
}
async fn delete_memory_inner(channel: &ChannelMessage, arguments: &str) -> anyhow::Result<String> {
let args: DeleteMemoryArgs =
serde_json::from_str(arguments).context("解析 delete_memory 参数失败")?;
let payload = json_with_identity(channel, json!({}));
let path = format!("/context/memories/{}", args.id);
let value = request_json("DELETE", &path, payload).await?;
let deleted = value
.get("deleted")
.and_then(Value::as_bool)
.unwrap_or(false);
if deleted {
Ok(format!("长期记忆已删除:id={}", args.id))
} else {
Ok(format!("没有找到可删除的长期记忆:id={}", args.id))
}
}
fn json_with_identity(channel: &ChannelMessage, mut payload: Value) -> Value {
let Value::Object(map) = &mut payload else {
return payload;
};
map.insert("channel".to_string(), json!(channel.channel));
map.insert("account_id".to_string(), json!(channel.account_id));
map.insert("from_user_id".to_string(), json!(channel.from_user_id));
map.insert("context".to_string(), json!(channel.context));
payload
}
async fn request_json(method: &str, path: &str, payload: Value) -> anyhow::Result<Value> {
let endpoint = format!("{}{}", context_base_url(), path);
let api_key = context_api_key()?;
let client = reqwest::Client::new();
let builder = match method {
"POST" => client.post(&endpoint),
"PATCH" => client.patch(&endpoint),
"DELETE" => client.delete(&endpoint),
other => return Err(anyhow!("不支持的 HTTP 方法: {other}")),
};
let response = builder
.header("x-ias-context-token", api_key)
.json(&payload)
.send()
.await
.with_context(|| format!("请求 context 接口失败: {endpoint}"))?;
let status = response.status();
let body = response.text().await.unwrap_or_default();
if !status.is_success() {
return Err(anyhow!("接口返回失败: status={status}, body={body}"));
}
serde_json::from_str(&body).context("解析 context 接口响应失败")
}
fn format_memory(memory: &Value) -> Option<String> {
let id = memory.get("id")?.as_i64()?;
let content = memory.get("content")?.as_str()?.trim();
let created_at = memory
.get("created_at")
.and_then(Value::as_str)
.unwrap_or("-");
let updated_at = memory
.get("updated_at")
.and_then(Value::as_str)
.unwrap_or("-");
Some(format!(
"- id={} 创建时间={} 更新时间={} 内容={}",
id, created_at, updated_at, content
))
}
fn required_text(value: &str, field: &str) -> anyhow::Result<String> {
let value = value.trim();
if value.is_empty() {
return Err(anyhow!("{field} 不能为空"));
}
Ok(value.to_string())
}
fn context_base_url() -> String {
std::env::var("IA_CONTEXT_BASE_URL")
.or_else(|_| 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")
}
fn context_api_key() -> anyhow::Result<String> {
std::env::var("IA_CONTEXT_API_KEY")
.or_else(|_| std::env::var("IA_CONTEXT_TOKEN"))
.or_else(|_| std::env::var("ROUTER_API_KEY"))
.map(|value| value.trim().to_string())
.ok()
.filter(|value| !value.is_empty())
.ok_or_else(|| anyhow!("未配置 IA_CONTEXT_API_KEY/IA_CONTEXT_TOKEN/ROUTER_API_KEY"))
}
+3
View File
@@ -0,0 +1,3 @@
pub mod date_time;
pub mod memory;
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>,
}