in devel

agent/agent.py

@@ -203,11 +203,22 @@ class Agent:
         print(reply)
     """
 
-    SYSTEM_PROMPT = (
+    SYSTEM_PROMPT = """
         "你是一个智能助手，可以调用工具完成用户的任务。\n"
         "调用工具时请确保参数完整准确。\n"
         "工具调用结果会自动返回给你，请根据结果给出最终回答。"
-    )
+        
+        可用工具如下:
+        {tools}
+        
+        请严格按照以下格式进行回应:
+
+        Thought: 你的思考过程，用于分析问题、拆解任务和规划下一步行动。
+        Action: 你决定采取的行动，必须是以下格式之一:
+        - `{{tool_name}}[{{tool_input}}]`:调用一个可用工具。
+        - `Finish[最终答案]`:当你认为已经获得最终答案时。
+        - 当你收集到足够的信息，能够回答用户的最终问题时，你必须在Action:字段后使用 Finish[最终答案] 来输出最终答案。
+    """
 
     def __init__(
         self,
@@ -283,7 +294,7 @@ class Agent:
             logger.info(f"🔁 推理步骤 {step}/{self._max_steps}")
 
             # 构造 API 消息列表
-            messages = self._build_messages(loop_history)
+            messages = self._build_messages(tools, loop_history)
 
             # 调用 LLM
             llm_resp = self.llm.chat(
@@ -368,8 +379,9 @@ class Agent:
 
     # ── 消息构造 ──────────────────────────────────────────────
 
-    def _build_messages(self, history: list[Message]) -> list[dict]:
-        messages = [{"role": "system", "content": self.system_prompt}]
+    def _build_messages(self, tools, history: list[Message]) -> list[dict]:
+        prompt = self.system_prompt.format(tools="\n".join(f"{tool['name']}: {tool['description']}" for tool in tools))
+        messages = [{"role": "system", "content":prompt}]
         messages += [m.to_api_dict() for m in history]
         return messages
 

config.yaml

@@ -19,19 +19,19 @@ llm:
 database:
   type: "sqlite"
   url: "sqlite:///skills.db"
-skills_directory: "C:\\Users\\sonto\\Workspace\\base_agent\\skills\\OpenCLI\\skills"  # 新增：SKILL.md 文件所在目录
+skills_directory: "./"  # 新增：SKILL.md 文件所在目录
 # ── 本地 MCP Server 配置 ───────────────────────────────────────
 mcp:
-  server_name: "DemoMCPServer"
+  server_name: "MCPServer"
   transport: "stdio"
   host: "localhost"
   port: 3000
   # 本地注册的工具列表
   enabled_tools:
-    - calculator
+#    - calculator
     - web_search
-    - file_reader
-    - code_executor
+#    - file_reader
+#    - code_executor
 
 # ── 在线 MCP Skill 配置 ────────────────────────────────────────
 # 每一项代表一个远端 MCP Server，其暴露的所有工具将作为 skill 注册到 Agent
@@ -113,8 +113,7 @@ tools:
   web_search:
     max_results: 5
     timeout: 10
-    api_key: ""
-    engine: "mock"
+    api_key: "7917bef5e46044af5209fdb78518be98be394f3fe763bbce3fbb503280408bd9"
 
   file_reader:
     allowed_root: "./workspace"

envs

@@ -0,0 +1,7 @@
+LOG_DIR="./logs"                       # 日志目录
+
+# --- HEXSTRIKE_AI 配置项 ---
+HEXSTRIKE_AI_ENTRY="hexstrike_server:app"       # WSGI 入口，格式为 模块名:变量名
+HEXSTRIKE_AI_BIND_ADDRESS="127.0.0.1:8000"      # 监听地址
+HEXSTRIKE_AI_WORKERS=4                          # 工作进程数
+HEXSTRIKE_AI_PID_FILE="./hexstrike_ai.pid"          # PID 文件保存位置

install.sh

@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+
+SKILLS_DIR="./skills"
+
+function install  () {
+  name=$1
+  url=$2
+
+  if ! [ -d "${SKILLS_DIR}/$name" ]; then
+    git clone $url  ${SKILLS_DIR}/$name
+  fi
+  requirements_file="${SKILLS_DIR}/$name/requirements.txt"
+  if [ -f ${requirements_file} ]; then
+    pip install -r $requirements_file
+  fi
+}
+
+pip install -r ./requirements.txt
+install "hexstrike_ai" "https://github.com/0x4m4/hexstrike-ai.git"

requirements.txt

@@ -22,3 +22,4 @@ paramiko>=3.4.0             # SSH 连接
 pytest>=8.0.0
 pytest-asyncio>=0.23.0
 python-dotenv>=1.0.0        # 从 .env 文件加载环境变量
+gunicorn

run.sh

@@ -0,0 +1,36 @@
+#!/usr/bin/env bash
+
+if [ -f "./envs" ]; then
+  source ./envs
+fi
+
+# 创建日志目录
+if ! [ -d ${LOG_DIR} ]; then
+  mkdir -p $LOG_DIR
+fi
+
+function start_hexstrike_ai() {
+  if [ -f "$HEXSTRIKE_AI_PID_FILE" ]; then
+    echo "错误: hexstrike_ai 似乎已在运行 (PID: $(cat $HEXSTRIKE_AI_PID_FILE))"
+    exit 1
+  fi
+  # 启动 Gunicorn
+  echo "正在启动 hexstrike_ai"
+  cd skills/hexstrike_ai &&  gunicorn -w ${HEXSTRIKE_AI_WORKERS} \
+      -b ${HEXSTRIKE_AI_BIND_ADDRESS} \
+      --pid ${HEXSTRIKE_AI_PID_FILE} \
+      --access-logfile "$LOG_DIR/access.log" \
+      --error-logfile "$LOG_DIR/error.log" \
+      -D \
+      ${HEXSTRIKE_AI_ENTRY}
+
+  if [ $? -eq 0 ]; then
+      echo "启动成功！PID: $(cat $PID_FILE)"
+  else
+      echo "启动失败，请检查日志。"
+  fi
+}
+
+
+
+start_hexstrike_ai

stop.sh

@@ -0,0 +1,23 @@
+#!/bin/bash
+
+if [ -f "./envs" ]; then
+  source ./envs
+fi
+
+function stop() {
+  PID_FILE=$1
+  echo $PID_FILE
+  if [ -f "$PID_FILE" ]; then
+      PID=$(cat $PID_FILE)
+      echo "正在停止进程 $PID..."
+      kill $PID
+      # 循环等待进程结束并清理 PID 文件
+      while ps -p $PID > /dev/null; do sleep 1; done
+      rm $PID_FILE
+      echo "服务已停止。"
+  else
+      echo "未发现正在运行的服务 (未找到 PID 文件)。"
+  fi
+}
+
+stop ${HEXSTRIKE_AI_PID_FILE}

tools/web_search.py

@@ -1,13 +1,11 @@
 """
 tools/web_search.py
-网络搜索工具 —— 支持 mock / SerpAPI / Brave Search
+网络搜索工具 —— google搜索
 配置通过 settings.tools['web_search'] 读取
 """
 
-import json
-import time
-from dataclasses import dataclass, field
-
+from dataclasses import dataclass
+from serpapi import SerpApiClient
 from config.settings import settings
 from tools.base_tool import BaseTool
 from utils.logger import get_logger
@@ -42,127 +40,49 @@ class Tool(BaseTool):
             "query": {
                 "type":        "string",
                 "description": "搜索关键词或问题，例如: 'Python 3.12 新特性'",
-            },
-            "max_results": {
-                "type":        "integer",
-                "description": "返回结果数量（默认来自 config.yaml web_search.max_results）",
-            },
+            }
         },
         "required": ["query"],
     }
 
-    def execute(self, query: str = "", max_results: int | None = None, **_) -> str:
-        if not query or not query.strip():
-            return "❌ 参数错误: query 不能为空"
-
-        n      = max_results or _cfg('max_results', 5)
-        engine = _cfg('engine', 'mock')
-        logger.info(
-            f"🔍 搜索: {query}\n"
-            f"   引擎={engine}  max_results={n}  "
-            f"[config engine={_cfg('engine')} max_results={_cfg('max_results')}]"
-        )
-
-        match engine:
-            case "serpapi":
-                results = self._search_serpapi(query, n)
-            case "brave":
-                results = self._search_brave(query, n)
-            case _:
-                results = self._search_mock(query, n)
-
-        if not results:
-            return f"🔍 搜索 '{query}' 未找到相关结果"
-
-        lines = [f"🔍 搜索结果: {query}  (共 {len(results)} 条)", "─" * 50]
-        for r in results:
-            lines.append(str(r))
-        return "\n".join(lines)
-
-    # ── 搜索引擎实现 ──────────────────────────────────────────
-
-    @staticmethod
-    def _search_mock(query: str, n: int) -> list[SearchResult]:
-        """Mock 搜索（无需 API Key，用于测试）"""
-        return [
-            SearchResult(
-                title=f"搜索结果 {i + 1}: {query}",
-                url=f"https://example.com/result/{i + 1}",
-                snippet=(
-                    f"这是关于 '{query}' 的第 {i + 1} 条模拟搜索结果。"
-                    f"实际使用请在 config.yaml 中配置 engine: serpapi 或 brave。"
-                ),
-                rank=i + 1,
-            )
-            for i in range(n)
-        ]
-
-    def _search_serpapi(self, query: str, n: int) -> list[SearchResult]:
-        """SerpAPI 搜索"""
-        api_key = _cfg('api_key', '')
-        if not api_key:
-            logger.warning("⚠️  SerpAPI api_key 未配置，回退到 mock 模式")
-            return self._search_mock(query, n)
+    def execute(self, query: str = "", **_) -> str:
+        """
+            一个基于SerpApi的实战网页搜索引擎工具。
+            它会智能地解析搜索结果，优先返回直接答案或知识图谱信息。
+            """
+        print(f"🔍 正在执行 [SerpApi] 网页搜索: {query}")
         try:
-            import httpx
-            timeout = _cfg('timeout', 10)
-            resp = httpx.get(
-                "https://serpapi.com/search",
-                params={
-                    "q":       query,
-                    "num":     n,
-                    "api_key": api_key,
-                    "engine":  "google",
-                },
-                timeout=timeout,
-            )
-            resp.raise_for_status()
-            data    = resp.json()
-            organic = data.get("organic_results", [])
-            return [
-                SearchResult(
-                    title=r.get("title", ""),
-                    url=r.get("link", ""),
-                    snippet=r.get("snippet", ""),
-                    rank=i + 1,
-                )
-                for i, r in enumerate(organic[:n])
-            ]
-        except Exception as e:
-            logger.error(f"❌ SerpAPI 搜索失败: {e}，回退到 mock")
-            return self._search_mock(query, n)
+            api_key = _cfg("api_key")
+            if not api_key:
+                return "错误:ApiKey未配置。"
+
+            params = {
+                "engine": "google",
+                "q": query,
+                "api_key": api_key,
+                "gl": "cn",  # 国家代码
+                "hl": "zh-cn",  # 语言代码
+            }
+
+            client = SerpApiClient(params)
+            results = client.get_dict()
+
+            # 智能解析:优先寻找最直接的答案
+            if "answer_box_list" in results:
+                return "\n".join(results["answer_box_list"])
+            if "answer_box" in results and "answer" in results["answer_box"]:
+                return results["answer_box"]["answer"]
+            if "knowledge_graph" in results and "description" in results["knowledge_graph"]:
+                return results["knowledge_graph"]["description"]
+            if "organic_results" in results and results["organic_results"]:
+                # 如果没有直接答案，则返回前三个有机结果的摘要
+                snippets = [
+                    f"[{i + 1}] {res.get('title', '')}\n{res.get('snippet', '')}"
+                    for i, res in enumerate(results["organic_results"][:3])
+                ]
+                return "\n\n".join(snippets)
+
+            return f"对不起，没有找到关于 '{query}' 的信息。"
 
-    def _search_brave(self, query: str, n: int) -> list[SearchResult]:
-        """Brave Search API"""
-        api_key = _cfg('api_key', '')
-        if not api_key:
-            logger.warning("⚠️  Brave Search api_key 未配置，回退到 mock 模式")
-            return self._search_mock(query, n)
-        try:
-            import httpx
-            timeout = _cfg('timeout', 10)
-            resp = httpx.get(
-                "https://api.search.brave.com/res/v1/web/search",
-                params={"q": query, "count": n},
-                headers={
-                    "Accept":               "application/json",
-                    "Accept-Encoding":      "gzip",
-                    "X-Subscription-Token": api_key,
-                },
-                timeout=timeout,
-            )
-            resp.raise_for_status()
-            data    = resp.json()
-            web     = data.get("web", {}).get("results", [])
-            return [
-                SearchResult(
-                    title=r.get("title", ""),
-                    url=r.get("url", ""),
-                    snippet=r.get("description", ""),
-                    rank=i + 1,
-                )
-                for i, r in enumerate(web[:n])
-            ]
         except Exception as e:
-            logger.error(f"❌ Brave Search 失败: {e}，回退到 mock")
-            return self._search_mock(query, n)
+            return f"搜索时发生错误: {e}"