fix(rig-integration): implement ToolEmbeddingDyn trait for McpToolAdaptor (#99)

- Added ToolEmbeddingDyn trait for McpToolAdaptor to enable dynamic loading of McpTool
- Enhanced logging to record details

Co-authored-by: byeblack <byeblack@misaki.red>
This commit is contained in:
byeblack 2025-04-07 10:28:19 +08:00 • committed by GitHub
commit d6a26fcd31
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 58 additions and 28 deletions

View file

@ -23,4 +23,11 @@ anyhow = "1.0"
serde_json = "1" serde_json = "1"
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
toml = "0.8" toml = "0.8"
futures = "0.3" futures = "0.3"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = [
"env-filter",
"std",
"fmt",
] }
tracing-appender = "0.2"

View file

@ -2,12 +2,11 @@ use futures::StreamExt;
use rig::{ use rig::{
agent::Agent, agent::Agent,
message::Message, message::Message,
providers::deepseek::DeepSeekCompletionModel, streaming::{StreamingChat, StreamingChoice, StreamingCompletionModel},
streaming::{StreamingChat, StreamingChoice},
}; };
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
pub async fn cli_chatbot(chatbot: Agent<DeepSeekCompletionModel>) -> anyhow::Result<()> { pub async fn cli_chatbot<M: StreamingCompletionModel>(chatbot: Agent<M>) -> anyhow::Result<()> {
let mut chat_log = vec![]; let mut chat_log = vec![];
let mut output = BufWriter::new(tokio::io::stdout()); let mut output = BufWriter::new(tokio::io::stdout());
@ -27,6 +26,7 @@ pub async fn cli_chatbot(chatbot: Agent<DeepSeekCompletionModel>) -> anyhow::Res
} }
match chatbot.stream_chat(input, chat_log.clone()).await { match chatbot.stream_chat(input, chat_log.clone()).await {
Ok(mut response) => { Ok(mut response) => {
tracing::info!(%input);
chat_log.push(Message::user(input)); chat_log.push(Message::user(input));
stream_output_agent_start(&mut output).await?; stream_output_agent_start(&mut output).await?;
let mut message_buf = String::new(); let mut message_buf = String::new();

View file

@ -1,16 +1,30 @@
use rig::{ use rig::{
embeddings::EmbeddingsBuilder, embeddings::EmbeddingsBuilder,
providers::{ providers::{cohere, deepseek},
cohere,
deepseek::{self, DEEPSEEK_CHAT},
},
vector_store::in_memory_store::InMemoryVectorStore, vector_store::in_memory_store::InMemoryVectorStore,
}; };
use tracing_appender::rolling::{RollingFileAppender, Rotation};
pub mod chat; pub mod chat;
pub mod config; pub mod config;
pub mod mcp_adaptor; pub mod mcp_adaptor;
#[tokio::main] #[tokio::main]
async fn main() -> anyhow::Result<()> { async fn main() -> anyhow::Result<()> {
let file_appender = RollingFileAppender::new(
Rotation::DAILY,
"logs",
format!("{}.log", env!("CARGO_CRATE_NAME")),
);
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::from_default_env()
.add_directive(tracing::Level::INFO.into()),
)
.with_writer(file_appender)
.with_file(false)
.with_ansi(false)
.init();
let config = config::Config::retrieve("config.toml").await?; let config = config::Config::retrieve("config.toml").await?;
let openai_client = { let openai_client = {
if let Some(key) = config.deepseek_key { if let Some(key) = config.deepseek_key {
@ -27,7 +41,7 @@ async fn main() -> anyhow::Result<()> {
} }
}; };
let mcp_manager = config.mcp.create_manager().await?; let mcp_manager = config.mcp.create_manager().await?;
eprintln!( tracing::info!(
"MCP Manager created, {} servers started", "MCP Manager created, {} servers started",
mcp_manager.clients.len() mcp_manager.clients.len()
); );
@ -39,26 +53,16 @@ async fn main() -> anyhow::Result<()> {
.build() .build()
.await?; .await?;
let store = InMemoryVectorStore::from_documents_with_id_f(embeddings, |f| { let store = InMemoryVectorStore::from_documents_with_id_f(embeddings, |f| {
eprintln!("store tool {}", f.name); tracing::info!("store tool {}", f.name);
f.name.clone() f.name.clone()
}); });
let index = store.index(embedding_model); let index = store.index(embedding_model);
let dpsk = openai_client let dpsk = openai_client
.agent(DEEPSEEK_CHAT) .agent(deepseek::DEEPSEEK_CHAT)
.context(
r#"You are an assistant here to help the user to do some works.
When you need to use tools, you should select which tool is most appropriate to meet the user's requirement.
Follow these instructions closely.
1. Consider the user's request carefully and identify the core elements of the request.
2. Select which tool among those made available to you is appropriate given the context.
3. This is very important: never perform the operation yourself and never give me the direct result.
Always respond with the name of the tool that should be used and the appropriate inputs
in the following format:
Tool: <tool name>
Inputs: <list of inputs>"#,
)
.dynamic_tools(4, index, tool_set) .dynamic_tools(4, index, tool_set)
.build(); .build();
chat::cli_chatbot(dpsk).await?; chat::cli_chatbot(dpsk).await?;
Ok(()) Ok(())
} }

View file

@ -1,6 +1,6 @@
use std::collections::HashMap; use std::collections::HashMap;
use rig::tool::{ToolDyn as RigTool, ToolSet}; use rig::tool::{ToolDyn as RigTool, ToolEmbeddingDyn, ToolSet};
use rmcp::{ use rmcp::{
RoleClient, RoleClient,
model::{CallToolRequestParam, CallToolResult, Tool as McpTool}, model::{CallToolRequestParam, CallToolResult, Tool as McpTool},
@ -49,6 +49,8 @@ impl RigTool for McpToolAdaptor {
.map_err(rig::tool::ToolError::JsonError)?, .map_err(rig::tool::ToolError::JsonError)?,
}) })
.await .await
.inspect(|result| tracing::info!(?result))
.inspect_err(|error| tracing::error!(%error))
.map_err(|e| rig::tool::ToolError::ToolCallError(Box::new(e)))?; .map_err(|e| rig::tool::ToolError::ToolCallError(Box::new(e)))?;
Ok(convert_mcp_call_tool_result_to_string(call_mcp_tool_result)) Ok(convert_mcp_call_tool_result_to_string(call_mcp_tool_result))
@ -56,6 +58,22 @@ impl RigTool for McpToolAdaptor {
} }
} }
impl ToolEmbeddingDyn for McpToolAdaptor {
fn context(&self) -> serde_json::Result<serde_json::Value> {
serde_json::to_value(self.tool.clone())
}
fn embedding_docs(&self) -> Vec<String> {
vec![
self.tool
.description
.as_deref()
.unwrap_or_default()
.to_string(),
]
}
}
pub struct McpManager { pub struct McpManager {
pub clients: HashMap<String, RunningService<RoleClient, ()>>, pub clients: HashMap<String, RunningService<RoleClient, ()>>,
} }
@ -72,7 +90,7 @@ impl McpManager {
for result in results { for result in results {
match result { match result {
Err(e) => { Err(e) => {
eprintln!("Failed to get tool set: {:?}", e); tracing::error!(error = %e, "Failed to get tool set");
} }
Ok(tools) => { Ok(tools) => {
tool_set.add_tools(tools); tool_set.add_tools(tools);
@ -89,14 +107,15 @@ pub fn convert_mcp_call_tool_result_to_string(result: CallToolResult) -> String
pub async fn get_tool_set(server: ServerSink) -> anyhow::Result<ToolSet> { pub async fn get_tool_set(server: ServerSink) -> anyhow::Result<ToolSet> {
let tools = server.list_all_tools().await?; let tools = server.list_all_tools().await?;
let mut tool_set = ToolSet::default(); let mut tool_builder = ToolSet::builder();
for tool in tools { for tool in tools {
eprintln!("get tool: {}", tool.name); tracing::info!("get tool: {}", tool.name);
let adaptor = McpToolAdaptor { let adaptor = McpToolAdaptor {
tool: tool.clone(), tool: tool.clone(),
server: server.clone(), server: server.clone(),
}; };
tool_set.add_tool(adaptor); tool_builder = tool_builder.dynamic_tool(adaptor);
} }
let tool_set = tool_builder.build();
Ok(tool_set) Ok(tool_set)
} }