Upload 8 files

app.py

@@ -0,0 +1,210 @@
+import gradio as gr
+import os
+import time
+
+# ------------------------------------------------------------------
+# 1. 加载环境变量 (在Hugging Face Spaces中从Secrets加载)
+# ------------------------------------------------------------------
+# 使用os.environ.get()来安全地获取，如果没有设置，可以给一个默认值或报错
+# 这里假设你在Hugging Face Spaces的Secrets中设置了'GEMINI_API_KEY'
+api_key = os.environ.get("GEMINI_API_KEY")
+if not api_key:
+    print("警告：未找到 GEMINI_API_KEY。请在Hugging Face Spaces的Secrets中设置它。")
+    # 为了本地测试，可以从.env文件加载(需要安装python-dotenv)
+    # from dotenv import load_dotenv
+    # load_dotenv()
+    # api_key = os.environ.get("GEMINI_API_KEY")
+
+# ------------------------------------------------------------------
+# 2. 初始化后端 (这是整个系统的启动点)
+# ------------------------------------------------------------------
+print("--- 正在启动 AI 助理系统 ---")
+
+# 导入我们的核心模块
+# 使用try...except来处理可能的导入错误，这在调试时很有用
+try:
+    from database.setup import initialize_system
+    from core.agent import SmartAIAgent
+
+    print("核心模块导入成功。")
+except ImportError as e:
+    print(f"导入模块时出错: {e}")
+    print("请确保所有项目文件都已正确放置。")
+    # 如果导入失败，系统无法运行，这里可以抛出异常或退出
+    raise
+
+# 执行一次性的系统初始化（创建数据库、向量索引、加载工具等）
+# 这个函数应该被设计为幂等的，即多次运行不会产生副作用
+try:
+    # initialize_system() 函数将负责所有数据库和向量库的设置
+    # 它会返回一个已注册工具的列表或其他必要信息
+    registered_tools, tool_recommender = initialize_system()
+    print("系统数据库和工具推荐器初始化完成。")
+
+    # 创建 AI 智能体实例，并将推荐器和工具传递给它
+    # Agent需要知道所有可以执行的工具函数
+    agent = SmartAIAgent(
+        tool_recommender=tool_recommender,
+        registered_tools=registered_tools,
+        api_key=api_key,
+    )
+    print("AI 智能体核心已成功创建。")
+except Exception as e:
+    print(f"系统初始化过程中发生严重错误: {e}")
+    agent = None  # 标记Agent不可用
+    # 在Gradio界面上可以显示错误信息
+
+print("--- AI 助理系统已准备就绪 ---")
+
+# ------------------------------------------------------------------
+# 3. Gradio 事件处理函数
+# ------------------------------------------------------------------
+
+
+def handle_user_message(user_input, history):
+    """
+    当用户发送消息时，此函数首先被调用。
+    它将用户的消息添加到聊天历史记录中。
+    """
+    if not user_input.strip():
+        # 如果用户输入为空，不做任何事
+        return "", history
+    # 将用户消息和一条空的机器人消息占位符添加到历史
+    history.append((user_input, None))
+    # 返回空字符串以清空输入框，并返回更新后的历史记录
+    return "", history
+
+
+def generate_bot_response(history):
+    """
+    此函数以流式方式生成机器人的响应。
+    它调用Agent核心来处理最新的用户消息。
+    """
+    if agent is None:
+        # 如果Agent初始化失败，返回错误信息
+        history[-1][1] = "抱歉，AI助理系统初始化失败，无法提供服务。"
+        yield history
+        return
+
+    # 获取最新的用户问题
+    user_question = history[-1][0]
+
+    # 初始化一个空的机器人消息
+    bot_message = ""
+    history[-1][1] = bot_message
+
+    try:
+        # 调用Agent的流式处理方法
+        # agent.stream_run() 应该是一个生成器，逐步yield出思考过程和最终答案
+        for chunk in agent.stream_run(user_question):
+            # 将每个新的文本块追加到机器人消息中
+            bot_message += chunk
+            # 更新历史记录中最后一个元组的机器人部分
+            history[-1][1] = bot_message
+            # yield更新后的历史记录，Gradio会用它来刷新界面
+            yield history
+            # (可选) 增加一个微小的延迟，让流式效果更明显
+            time.sleep(0.01)
+
+    except Exception as e:
+        # 如果在处理过程中发生错误，将错误信息显示给用户
+        error_message = f"\n\n抱歉，处理您的请求时发生了错误：\n`{str(e)}`"
+        history[-1][1] += error_message
+        yield history
+
+
+# ------------------------------------------------------------------
+# 4. 创建 Gradio 界面
+# ------------------------------------------------------------------
+
+# 自定义CSS来美化界面
+custom_css = """
+/* 简单的CSS自定义，让界面更好看 */
+#chatbot .user {
+    background-color: #E0F7FA; /* 浅青色背景 */
+}
+#chatbot .bot {
+    background-color: #F1F8E9; /* 浅绿色背景 */
+}
+"""
+
+with gr.Blocks(
+    theme=gr.themes.Soft(primary_hue="teal", secondary_hue="lime"),
+    css=custom_css,
+    title="智能 AI 助理",
+) as demo:
+    gr.Markdown(
+        """
+        # 🚀 智能 AI 助理 Demo
+        ### (LangChain + LlamaIndex + Gemini)
+        这是一个演示如何结合 LangChain (智能体编排) 和 LlamaIndex (工具检索) 构建高级AI助理的Demo。
+        - **提问:** 在下面的文本框中输入你的问题。
+        - **观察:** 观察AI的思考过程，包括它如何推荐、选择和调用工具。
+        """
+    )
+
+    chatbot = gr.Chatbot(
+        [],
+        elem_id="chatbot",
+        label="聊天窗口",
+        bubble_full_width=False,
+        height=650,
+        avatar_images=(
+            None,
+            "https://raw.githubusercontent.com/gradio-app/gradio/main/guides/assets/logo.png",
+        ),  # 机器人用Gradio logo
+    )
+
+    with gr.Row():
+        text_input = gr.Textbox(
+            scale=4,
+            show_label=False,
+            placeholder="例如: '苹果公司(AAPL)今天的股价是多少？' 或 '关于AI的最新进展有什么新闻？'",
+            container=False,
+        )
+        submit_button = gr.Button("发送", variant="primary", scale=1, min_width=150)
+
+    # 示例问题，方便用户快速体验
+    gr.Examples(
+        examples=[
+            "苹果公司(AAPL)的股价是多少？",
+            "关于AI驱动的药物发现有什么最新新闻？",
+            "你好，你能做什么？",
+            "用Python写一个快速排序算法",  # 测试不使用工具的场景
+        ],
+        inputs=text_input,
+    )
+
+    # 定义事件的触发流程
+    # 当用户提交输入时（点击按钮或按回车）
+    submit_event = text_input.submit(
+        fn=handle_user_message,
+        inputs=[text_input, chatbot],
+        outputs=[text_input, chatbot],
+        queue=False,  # 立即执行，不排队
+    ).then(
+        fn=generate_bot_response,
+        inputs=[chatbot],
+        outputs=[chatbot],
+    )
+
+    submit_button.click(
+        fn=handle_user_message,
+        inputs=[text_input, chatbot],
+        outputs=[text_input, chatbot],
+        queue=False,
+    ).then(
+        fn=generate_bot_response,
+        inputs=[chatbot],
+        outputs=[chatbot],
+    )
+
+# ------------------------------------------------------------------
+# 5. 启动应用
+# ------------------------------------------------------------------
+if __name__ == "__main__":
+    # 使用 .queue() 来允许多个用户同时使用，这在Hugging Face Spaces上是推荐做法
+    demo.queue()
+    # .launch() 会启动Web服务器
+    # 在Hugging Face Spaces上，它会自动找到并运行这个
+    demo.launch(debug=True)  # debug=True可以在本地看到更详细的日志

core/agent.py

@@ -0,0 +1,159 @@
+# core/agent.py
+
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain.agents import AgentExecutor, create_json_chat_agent
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+from typing import List, Any
+import json
+
+from .tool_recommender import LlamaIndexToolRecommender
+from tools.tool_registry import get_tool_by_name
+
+# Agent的思考模板
+AGENT_PROMPT_TEMPLATE = """
+你是一个强大的AI助理。你的任务是理解用户的问题，并决定是否需要使用工具来回答。
+
+你有以下工具可用：
+{tools}
+
+如果需要使用工具，请严格按照以下JSON格式进行响应，不要包含任何其他文本或解释：
+{{
+  "tool": "要调用的工具名称",
+  "tool_input": {{ "参数1": "值1", "参数2": "值2" }}
+}}
+
+如果不需要使用任何工具，请直接回答用户的问题。
+
+这是对话历史：
+{chat_history}
+
+用户问题：{input}
+
+现在，请你思考并作出回应（JSON或直接回答）：
+"""
+
+
+class SmartAIAgent:
+    def __init__(
+        self,
+        tool_recommender: LlamaIndexToolRecommender,
+        registered_tools: List[Any],
+        api_key: str,
+    ):
+        self.tool_recommender = tool_recommender
+        self.registered_tools = registered_tools
+        self.llm = ChatGoogleGenerativeAI(
+            model="gemini-1.5-pro-latest",
+            google_api_key=api_key,
+            convert_system_message_to_human=True,  # 兼容性设置
+        )
+        self.chat_history = []
+        print("LangChain Agent已初始化，使用Gemini 1.5 Pro。")
+
+    def _format_tools_for_prompt(self, tools: List[dict]) -> str:
+        """将工具列表格式化为清晰的字符串，用于Prompt。"""
+        if not tools:
+            return "没有可用的工具。"
+
+        tool_strings = []
+        for tool in tools:
+            # 解析JSON字符串参数
+            params = json.loads(tool["parameters"])
+            param_str = ", ".join(
+                [f"{p_name}: {p_type}" for p_name, p_type in params.items()]
+            )
+            tool_strings.append(
+                f"- 工具名称: {tool['name']}\n  - 描述: {tool['description']}\n  - 参数: {param_str}"
+            )
+        return "\n".join(tool_strings)
+
+    def _format_chat_history(self) -> str:
+        """格式化聊天历史。"""
+        return "\n".join([f"{msg.type}: {msg.content}" for msg in self.chat_history])
+
+    def stream_run(self, user_input: str):
+        """
+        处理用户输入的流式方法。
+        这是一个生成器，会逐步yield出思考过程和结果。
+        """
+        # 1. 将用户输入添加到历史记录
+        self.chat_history.append(HumanMessage(content=user_input))
+        yield "🤔 正在分析您的问题...\n"
+
+        # 2. 调用工具推荐系统
+        yield "🔍 正在从工具库中推荐相关工具...\n"
+        recommended_tools_meta = self.tool_recommender.recommend_tools(user_input)
+
+        if not recommended_tools_meta:
+            yield "ℹ️ 未找到相关工具，将直接回答。\n"
+            recommended_tools_prompt = "没有推荐的工具。"
+        else:
+            tool_names = [t["name"] for t in recommended_tools_meta]
+            yield f"✅ 推荐工具: `{', '.join(tool_names)}`\n"
+            recommended_tools_prompt = self._format_tools_for_prompt(
+                recommended_tools_meta
+            )
+
+        # 3. 构建Agent Prompt，让LLM决策
+        yield "🧠 正在让AI大脑（Gemini Pro）决定如何行动...\n"
+        prompt = AGENT_PROMPT_TEMPLATE.format(
+            tools=recommended_tools_prompt,
+            chat_history=self._format_chat_history(),
+            input=user_input,
+        )
+
+        # 4. 调用LLM获取决策
+        llm_response = self.llm.invoke(prompt)
+        llm_decision_content = llm_response.content
+
+        # 5. 解析和执行决策
+        try:
+            # 尝试将LLM的响应解析为JSON
+            decision = json.loads(llm_decision_content)
+            tool_name = decision.get("tool")
+            tool_input = decision.get("tool_input")
+
+            yield f"💡 AI决策：调用工具 `{tool_name}`，参数为 `{tool_input}`\n"
+
+            # 查找并执行工具
+            tool_to_execute = get_tool_by_name(tool_name)
+            if tool_to_execute:
+                yield f"⚙️ 正在执行工具 `{tool_name}`...\n"
+                tool_output = tool_to_execute.invoke(tool_input)
+                yield f"📊 工具返回结果：\n---\n{str(tool_output)[:500]}...\n---\n"
+
+                # 将工具调用和结果添加到历史
+                self.chat_history.append(AIMessage(content=llm_decision_content))
+                self.chat_history.append(
+                    ToolMessage(content=str(tool_output), tool_call_id="N/A")
+                )  # 简单记录
+
+                # 6. 基于工具结果生成最终答案
+                yield "✍️ 正在根据工具结果生成最终回答...\n\n"
+                final_answer_prompt = f"""
+                基于以下对话历史和最新的工具结果，请为用户生成一个最终的、完整的、自然的回答。
+
+                对话历史：
+                {self._format_chat_history()}
+
+                请直接回答，不要提及你的思考过程。
+                """
+                final_answer_stream = self.llm.stream(final_answer_prompt)
+                full_final_answer = ""
+                for chunk in final_answer_stream:
+                    yield chunk.content
+                    full_final_answer += chunk.content
+
+                # 将最终答案添加到历史
+                self.chat_history.append(AIMessage(content=full_final_answer))
+
+            else:
+                yield f"❌ 错误：AI决策调用的工具 `{tool_name}` 不存在。\n"
+
+        except (json.JSONDecodeError, KeyError):
+            # 如果LLM的响应不是JSON，则认为是直接回答
+            yield "✅ AI决策：直接回答。\n\n"
+            # 直接将LLM的响应作为最终答案
+            yield llm_decision_content
+            self.chat_history.append(AIMessage(content=llm_decision_content))

core/tool_recommender.py

@@ -0,0 +1,92 @@
+# core/tool_recommender.py
+
+import sqlite3
+from pymilvus import MilvusClient
+from llama_index.embeddings.google import GooglePairedEmbeddings
+from typing import List, Dict
+
+
+class LlamaIndexToolRecommender:
+    """
+    使用LlamaIndex的嵌入模型和Milvus Lite进行工具推荐。
+    """
+
+    def __init__(self, milvus_client: MilvusClient, sqlite_db_path: str):
+        self.milvus_client = milvus_client
+        self.sqlite_db_path = sqlite_db_path
+        self.collection_name = "tool_embeddings"
+
+        try:
+            self.embed_model = GooglePairedEmbeddings(
+                model_name="models/text-embedding-004",
+                task_type="retrieval_query",  # 用于查询的嵌入
+            )
+        except Exception as e:
+            print(f"错误：无法初始化Google嵌入模型。请检查API Key。 - {e}")
+            raise
+
+        print("LlamaIndex工具推荐器已初始化。")
+
+    def recommend_tools(self, user_query: str, top_k: int = 3) -> List[Dict]:
+        """
+        根据用户查询，推荐最相关的top_k个工具。
+        返回一个包含工具元数据字典的列表。
+        """
+        print(f"\n[推荐系统] 收到查询: '{user_query}'")
+
+        # 1. 生成查询嵌入
+        query_embedding = self.embed_model.get_text_embedding(user_query)
+
+        # 2. 在Milvus中搜索相似的工具
+        search_results = self.milvus_client.search(
+            collection_name=self.collection_name,
+            data=[query_embedding],
+            limit=top_k,
+            output_fields=["id"],
+        )
+
+        if not search_results or not search_results[0]:
+            print("[推荐系统] 在Milvus中未找到相似工具。")
+            return []
+
+        # 3. 提取推荐的工具ID
+        recommended_ids = [hit.id for hit in search_results[0]]
+        print(f"[推荐系统] Milvus推荐的工具ID: {recommended_ids}")
+
+        # 4. 从SQLite中根据ID获取完整的工具元数据
+        with sqlite3.connect(self.sqlite_db_path) as conn:
+            cursor = conn.cursor()
+            # 使用IN子句一次性查询所有ID
+            placeholders = ",".join("?" for _ in recommended_ids)
+            cursor.execute(
+                f"SELECT name, description, parameters FROM tools WHERE id IN ({placeholders})",
+                recommended_ids,
+            )
+            tools_metadata = cursor.fetchall()
+
+        # 5. 将结果格式化为字典列表
+        # 注意：数据库返回的顺序可能与推荐顺序不同，需要重新排序
+        id_map = {tool_id: i for i, tool_id in enumerate(recommended_ids)}
+
+        formatted_tools = []
+        for name, description, parameters in tools_metadata:
+            # 找到这个工具在推荐列表中的原始ID
+            # 这是一个简化的查找，实际中可以做的更高效
+            for tool_id in recommended_ids:
+                cursor.execute("SELECT name FROM tools WHERE id = ?", (tool_id,))
+                if cursor.fetchone()[0] == name:
+                    formatted_tools.append(
+                        {
+                            "name": name,
+                            "description": description,
+                            "parameters": parameters,
+                            "original_rank": id_map.get(tool_id),
+                        }
+                    )
+
+        formatted_tools.sort(key=lambda x: x["original_rank"])
+        print(f"[推荐系统] 最终推荐的工具: {[t['name'] for t in formatted_tools]}")
+        return [
+            {k: v for k, v in t.items() if k != "original_rank"}
+            for t in formatted_tools
+        ]

database/setup.py

@@ -0,0 +1,205 @@
+# database/setup.py
+
+import os
+import sqlite3
+import json
+from pymilvus import MilvusClient, FieldSchema, CollectionSchema, DataType
+from llama_index.embeddings.google import GooglePairedEmbeddings
+
+# 导入你的工具注册表
+from tools.tool_registry import get_all_tools
+
+# --- 配置持久化路径 ---
+DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data")
+SQLITE_DB_PATH = os.path.join(DATA_DIR, "tools.metadata.db")
+MILVUS_DATA_PATH = os.path.join(
+    DATA_DIR, "milvus_data.db"
+)  # Milvus Lite 将数据存在一个文件中
+
+# --- Milvus Lite 配置 ---
+MILVUS_COLLECTION_NAME = "tool_embeddings"
+EMBEDDING_DIM = 768  # Google's text-embedding-004 model dimension
+
+# --- 全局变量，避免重复初始化 ---
+_db_initialized = False
+_milvus_initialized = False
+
+
+def initialize_system():
+    """
+    系统的主初始化函数。
+    它会创建目录、设置数据库和向量库，并加载工具。
+    这个函数是幂等的，即多次调用不会产生副作用。
+    """
+    global _db_initialized, _milvus_initialized
+
+    print("--- 开始系统初始化 ---")
+
+    # 1. 创建数据目录
+    os.makedirs(DATA_DIR, exist_ok=True)
+
+    # 2. 初始化SQLite数据库
+    if not _db_initialized:
+        _init_sqlite_db()
+        _db_initialized = True
+
+    # 3. 初始化Milvus Lite向量数据库
+    if not _milvus_initialized:
+        milvus_client = _init_milvus_lite()
+        _milvus_initialized = True
+    else:
+        milvus_client = MilvusClient(uri=MILVUS_DATA_PATH)
+
+    # 4. 获取所有工具定义
+    all_tools_definitions = get_all_tools()
+
+    # 5. 将工具元数据同步到SQLite
+    _sync_tools_to_sqlite(all_tools_definitions)
+
+    # 6. 将工具描述的嵌入同步到Milvus Lite
+    _sync_tool_embeddings_to_milvus(milvus_client)
+
+    # 7. 从LlamaIndex创建工具推荐器 (在这里创建并返回)
+    from core.tool_recommender import LlamaIndexToolRecommender
+
+    tool_recommender = LlamaIndexToolRecommender(
+        milvus_client=milvus_client, sqlite_db_path=SQLITE_DB_PATH
+    )
+
+    print("--- 系统初始化完成 ---")
+    return all_tools_definitions, tool_recommender
+
+
+def _init_sqlite_db():
+    """初始化SQLite数据库并创建表。"""
+    print(f"SQLite DB 路径: {SQLITE_DB_PATH}")
+    with sqlite3.connect(SQLITE_DB_PATH) as conn:
+        cursor = conn.cursor()
+        cursor.execute(
+            """
+            CREATE TABLE IF NOT EXISTS tools (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                name TEXT UNIQUE NOT NULL,
+                description TEXT NOT NULL,
+                parameters TEXT NOT NULL -- 存储JSON字符串
+            )
+        """
+        )
+        conn.commit()
+    print("SQLite DB 表已确认存在。")
+
+
+def _init_milvus_lite():
+    """初始化Milvus Lite并创建集合和索引。"""
+    print(f"Milvus Lite 数据路径: {MILVUS_DATA_PATH}")
+    client = MilvusClient(uri=MILVUS_DATA_PATH)
+
+    if not client.has_collection(collection_name=MILVUS_COLLECTION_NAME):
+        print(f"Milvus集合 '{MILVUS_COLLECTION_NAME}' 不存在，正在创建...")
+        fields = [
+            FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
+            FieldSchema(
+                name="embedding", dtype=DataType.FLOAT_VECTOR, dim=EMBEDDING_DIM
+            ),
+        ]
+        schema = CollectionSchema(fields, description="Tool embedding collection")
+        client.create_collection(collection_name=MILVUS_COLLECTION_NAME, schema=schema)
+
+        index_params = client.prepare_index_params()
+        index_params.add_index(
+            field_name="embedding",
+            index_type="AUTOINDEX",  # 让Milvus自动选择最佳索引
+            metric_type="L2",
+        )
+        client.create_index(
+            collection_name=MILVUS_COLLECTION_NAME, index_params=index_params
+        )
+        print("Milvus集合和索引创建完成。")
+    else:
+        print(f"Milvus集合 '{MILVUS_COLLECTION_NAME}' 已存在。")
+        # 确保集合已加载到内存中以供搜索
+        client.load_collection(collection_name=MILVUS_COLLECTION_NAME)
+
+    return client
+
+
+def _sync_tools_to_sqlite(tools_definitions):
+    """将工具定义同步到SQLite数据库。"""
+    print("正在同步工具元数据到SQLite...")
+    with sqlite3.connect(SQLITE_DB_PATH) as conn:
+        cursor = conn.cursor()
+        for tool in tools_definitions:
+            cursor.execute("SELECT id FROM tools WHERE name = ?", (tool.name,))
+            if cursor.fetchone() is None:
+                # 工具不存在，插入新工具
+                cursor.execute(
+                    "INSERT INTO tools (name, description, parameters) VALUES (?, ?, ?)",
+                    (tool.name, tool.description, json.dumps(tool.args)),
+                )
+                print(f"  - 新���工具到SQLite: {tool.name}")
+        conn.commit()
+    print("SQLite同步完成。")
+
+
+def _sync_tool_embeddings_to_milvus(milvus_client):
+    """计算并同步工具描述的嵌入到Milvus Lite。"""
+    print("正在同步工具嵌入到Milvus...")
+
+    # 1. 从SQLite获取所有工具
+    with sqlite3.connect(SQLITE_DB_PATH) as conn:
+        cursor = conn.cursor()
+        cursor.execute("SELECT id, description FROM tools")
+        all_tools_in_db = cursor.fetchall()
+
+    # 2. 获取Milvus中已存在的工具ID
+    try:
+        existing_milvus_ids_raw = milvus_client.query(
+            collection_name=MILVUS_COLLECTION_NAME,
+            filter="id > 0",
+            output_fields=["id"],
+        )
+        existing_milvus_ids = {item["id"] for item in existing_milvus_ids_raw}
+    except Exception:
+        existing_milvus_ids = set()
+
+    # 3. 找出需要计算嵌入的新工具
+    new_tools_to_embed = [
+        (tool_id, description)
+        for tool_id, description in all_tools_in_db
+        if tool_id not in existing_milvus_ids
+    ]
+
+    if not new_tools_to_embed:
+        print("所有工具嵌入已是最新，无需同步。")
+        return
+
+    print(f"发现 {len(new_tools_to_embed)} 个新工具需要生成嵌入...")
+
+    # 4. 初始化嵌入模型
+    try:
+        # 确保你的API Key已在环境中设置
+        embed_model = GooglePairedEmbeddings(
+            model_name="models/text-embedding-004",
+            task_type="retrieval_document",  # 用于存储的文档嵌入
+        )
+    except Exception as e:
+        print(f"错误：无法初始化Google嵌入模型。请检查API Key。 - {e}")
+        return
+
+    # 5. 生成嵌入并准备插入
+    tool_ids_to_insert = [tool[0] for tool in new_tools_to_embed]
+    descriptions_to_embed = [tool[1] for tool in new_tools_to_embed]
+
+    embeddings = embed_model.get_text_embedding_batch(
+        descriptions_to_embed, show_progress=True
+    )
+
+    data_to_insert = [
+        {"id": tool_id, "embedding": embedding}
+        for tool_id, embedding in zip(tool_ids_to_insert, embeddings)
+    ]
+
+    # 6. 插入到Milvus
+    milvus_client.insert(collection_name=MILVUS_COLLECTION_NAME, data=data_to_insert)
+    milvus_client.flush([MILVUS_COLLECTION_NAME])  # 确保数据写入
+    print(f"成功将 {len(data_to_insert)} 个新嵌入插入到Milvus。")

requirements.txt

@@ -0,0 +1,18 @@
+# Core
+gradio
+python-dotenv
+
+# LangChain & LlamaIndex
+langchain
+langchain-core
+langchain-google-genai
+llama-index
+llama-index-embeddings-google
+llama-index-llms-google
+
+# Vector DB
+pymilvus-lite
+
+# Tools
+requests
+beautifulsoup4

tools/news_tool.py

@@ -0,0 +1,42 @@
+# tools/news_tool.py
+
+import requests
+from bs4 import BeautifulSoup
+
+
+def search_latest_news(query: str) -> str:
+    """
+    使用requests和BeautifulSoup抓取DuckDuckGo搜索结果来模拟新闻搜索。
+    """
+    print(f"--- 正在执行工具: search_latest_news, 参数: {query} ---")
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"
+    }
+    url = f"https://html.duckduckgo.com/html/?q={query}"
+
+    try:
+        response = requests.get(url, headers=headers, timeout=5)
+        response.raise_for_status()
+
+        soup = BeautifulSoup(response.text, "html.parser")
+        results = soup.find_all("div", class_="result")
+
+        if not results:
+            return "没有找到相关的新闻报道。"
+
+        # 提取前三个结果的摘要
+        snippets = []
+        for result in results[:3]:
+            title_tag = result.find("a", class_="result__a")
+            snippet_tag = result.find("a", class_="result__snippet")
+            if title_tag and snippet_tag:
+                title = title_tag.text.strip()
+                snippet = snippet_tag.text.strip()
+                snippets.append(f"标题: {title}\n摘要: {snippet}\n")
+
+        return "\n---\n".join(snippets)
+
+    except requests.RequestException as e:
+        return f"搜索新闻时发生网络错误: {e}"
+    except Exception as e:
+        return f"解析新闻搜索结果时发生错误: {e}"

tools/stock_tool.py

@@ -0,0 +1,23 @@
+# tools/stock_tool.py
+
+import requests
+import random
+
+
+def get_stock_price(symbol: str) -> str:
+    """
+    模拟获取股票价格的函数。
+    在真实世界中，这里会调用一个真正的金融API。
+    """
+    print(f"--- 正在执行工具: get_stock_price, 参数: {symbol} ---")
+    symbol = symbol.upper()
+    # 模拟API调用
+    try:
+        # 这是一个模拟，实际应该调用如Alpha Vantage, Yahoo Finance等API
+        if symbol in ["AAPL", "GOOGL", "MSFT"]:
+            price = round(random.uniform(100, 500), 2)
+            return f"股票 {symbol} 的模拟实时价格是 ${price}。"
+        else:
+            return f"找不到股票代码为 {symbol} 的信息。"
+    except Exception as e:
+        return f"调用股票API时发生错误: {e}"

tools/tool_registry.py

@@ -0,0 +1,50 @@
+# tools/tool_registry.py
+
+from langchain_core.tools import tool
+from typing import List, Dict, Any
+
+# 导入你的实际工具函数
+from .stock_tool import get_stock_price
+from .news_tool import search_latest_news
+
+# 使用LangChain的@tool装饰器来定义工具，它会自动处理描述和参数结构
+# 这比手动构建字典更健壮
+
+
+# 使用 @tool 装饰器定义你的工具
+@tool
+def get_stock_price_tool(symbol: str) -> str:
+    """
+    获取指定股票代码（例如AAPL, GOOGL）的实时股票价格。
+    当用户询问特定公司的股价时使用此工具。
+    """
+    return get_stock_price(symbol)
+
+
+@tool
+def search_latest_news_tool(query: str) -> str:
+    """
+    根据关键词搜索最新的新闻报道。
+    当用户询问关于某个主题的最新动态、事件或新闻时使用此工具。
+    """
+    return search_latest_news(query)
+
+
+# 集中管理所有工具
+_all_tools = [
+    get_stock_price_tool,
+    search_latest_news_tool,
+]
+
+
+def get_all_tools() -> List[Any]:
+    """返回一个包含所有已定义工具的列表。"""
+    return _all_tools
+
+
+def get_tool_by_name(name: str) -> Any:
+    """根据名称查找并返回工具对象。"""
+    for tool_obj in _all_tools:
+        if tool_obj.name == name:
+            return tool_obj
+    return None