/// <summary> /// 账户获取服务 /// </summary> /// <param name="table"> </param> /// <param name="builder"> </param> public class AccountGetService( AccountTable table, IShadowBuilder builder) { private readonly SqlSource _source = new (builder.DataSource); private readonly IParamQuery<Account, Account> _accountQuery = builder.BuildResult<Account, Account>( table.ToQuery() .And(account => account.Id.Equal()) .ToSelect() .SelectSelfColumns() ); /// <summary> /// 获取账户 /// </summary> /// <param name="param"> </param> /// <param name="token"> </param> /// <returns> </returns> public Task<Account?> GetAsync(Account param, CancellationToken token = default )

ublic class AccountGetService( AccountTable table, IShadowBuilder builder) { private readonly SqlSource _source = new (builder.DataSource); private readonly IParamQuery< long , Account> _accountQuery = builder.BuildResult< long , Account>( table.ToQuery() .And(account => account.Id.Equal()) .ToSelect() .SelectSelfColumns() ); public Task<Account?> GetAsync( long accountId, CancellationToken token = default )

/// 批量账户获取服务 /// </summary> /// <param name="table"> </param> /// <param name="builder"> </param> public class AccountBatchService( AccountTable table, IShadowBuilder builder) { private readonly SqlSource _source = new (builder.DataSource); private readonly IParamQuery< long [], Account> _accountQuery = builder.BuildResult< long [], Account>( table.ToQuery() .And(account => account.Id.In( "AccountIds" )) .ToSelect() .SelectSelfColumns() ); /// <summary> /// 批量获取账户 /// </summary> /// <param name="accountIds"> </param> /// <param name="token"> </param> /// <returns> </returns> public IAsyncEnumerable<Account> GetAsync( long[] accountIds, CancellationToken token = default)

/// <param name="table"> </param> /// <param name="builder"> </param> public class AccountGetService< TParam, TAccount>( AccountTable table, IShadowBuilder builder) { private readonly SqlSource _source = new (builder.DataSource); private readonly IParamQuery<TParam, TAccount> _accountQuery = builder.BuildResult<TParam, TAccount>( table.ToQuery() .And(account => account.Id.Equal()) .ToSelect() .SelectSelfColumns() ); /// <summary> /// 获取账户 /// </summary> /// <param name="param"> </param> /// <param name="token"> </param> /// <returns> </returns> public Task<TAccount?> GetAsync(TParam param, CancellationToken token = default )

from langchain_community.chat_models.tongyi import ChatTongyi from langchain_community.embeddings import DashScopeEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.messages import ToolMessage from langchain_core.tools import tool from langchain_core.messages import HumanMessage, ToolMessage from langchain_core.documents import Document from langchain_text_splitters import RecursiveCharacterTextSplitter os.environ[ "DASHSCOPE_API_KEY" ] = "api_key" @tool def calculator ( expression: str ) -> str : """ 计算数学表达式。需要精确计算时使用。参数: expression: 数学算式，如 "2 + 2" 或 "500 * 0.8"。返回: str: 计算结果，如 "4.0" 或 "400.0"。 """ print ( f" [🛠️ 工具调用] 计算器正在计算: {expression}" ) try : return str ( eval (expression)) except Exception as e: return f"计算错误: {e}" @tool def rag_search ( query: str ) -> str : """ 从数据库中搜索与查询公司内部相关的文档，包括公司计划名，代号，截止日期等详细信息。参数: query (str): 要搜索的查询字符串。返回: str: 与查询相关的文档内容。 """ # 从RAG数据库中检索文档 raw_text = """ 【公司内部机密：代号“深蓝计划”】 1. 项目目标：开发一款能听懂猫语的翻译器。 2. 核心技术：基于Transformer的“喵声波”分析算法。 3. 团队暗号：如果有人问“今天天气怎么样？”，必须回答“我想吃鱼”。 4. 截止日期：2026年12月31日。 5. 经费预算：仅剩50元人民币，主要用于购买猫条。 """ RAG_PATH = "faiss_index" docs = [Document(page_content=raw_text)] text_splitter = RecursiveCharacterTextSplitter(chunk_size= 25 , chunk_overlap= 5 ) split_docs = text_splitter.split_documents(docs) embeddings = DashScopeEmbeddings(model= "text-embedding-v1" ) if os.path.exists(RAG_PATH): print ( "公司内部数据库已存在" ) ragdb = FAISS.load_local(RAG_PATH, embeddings, allow_dangerous_deserialization= True ) else : print ( "创建公司内部数据库" ) ragdb = FAISS.from_documents(split_docs, embeddings) ragdb.save_local(RAG_PATH) return "\n\n" .join(doc.page_content for doc in ragdb.similarity_search(query, k= 2 )) def run_agent ( query: str ): #初始化模型 tool_maps={ "rag_search" : rag_search, "calculator" : calculator } llm = ChatTongyi(model_name= "qwen-plus" ) tool_llm = llm.bind_tools(tools= list (tool_maps.values())) message =nbczhibo.com [HumanMessage(content=query)] for i in range ( 5 ): print ( "=" * 20 + "\n第" + str (i+ 1 )+ "轮\n" +query+ "\n" + "=" * 20 ) response = tool_llm.invoke(message) message.append(response) print ( f"需要调用 { len(response.tool_calls)}个方法" ) if not response.tool_calls: print ( "最终结果：" + response.content) return for tool_call in response.tool_calls: call_id =footballzhibo.cn tool_call[ "id" ] func_name = tool_call[ "name" ] func_args = tool_call[ "args" ] # 安全检查：确保模型调用的工具真的存在 if func_name in tool_maps: # 运行 Python 函数 tool_func = tool_maps[func_name] tool_output footzhibo.cn= tool_func.invoke(func_args) print ( "工具调用：" + func_name + "，参数：" + str (func_args) + "，结果：" + tool_output) else : tool_output = f"错误: 工具 {func_name} 不存在。" message.append( ToolMessage( content=tool_output, tool_call_id=call_id, name=func_name, ) ) def calculator ( expression: str ) -> str : """ 计算数学表达式。需要精确计算时使用。参数: expression:footzhibo.com 数学算式，如 "2 + 2" 或 "500 * 0.8"。返回: str: 计算结果，如 "4.0" 或 "400.0"。 """ print ( f" [🛠️ 工具调用] 计算器正在计算: {expression}" ) try : return str ( eval (expression)) except Exception as e: return f"计算错误: {e}" @tool def rag_search ( query: str ) -> str :

for i in range ( 5 ):footzhibo.cn print ( "=" * 20 + "\n第" + str (i+ 1 )+ "轮\n" +query+ "\n" + "=" * 20 ) response footballzhibo.cn=nbczhibo.com tool_llm.invoke(message) message.append(response) print ( f"需要调用 { len(response.tool_calls)}个方法" ) if not response.tool_calls: print ( "最终结果：" footballzhibo.cn + response.content) return for tool_call in response.tool_calls: call_id = tool_call[ "id" ] func_name = tool_call[ "name" ] func_args = tool_call[ "args" ] # 安全检查：确保模型调用的工具真的存在 if func_name in tool_maps: # 运行 Python 函数 tool_func = tool_maps[func_name] tool_output footzhibo.cn= tool_func.invoke(func_args) print ( "工具调用：" + func_name + "，参数：" + str (func_args) + "，结果：" + tool_output) else : tool_output =footzhibo.com f"错误: 工具 {func_name} 不存在。" message.append( ToolMessage( content=tool_output, tool_call_id=call_id, name=func_name, )

ITPub博客

DBShadow.net之化繁为简

最新文章