接入搜索 HTTP API

更新时间:
复制 MD 格式

本文介绍如何在集成信息查询服务-OpenAPI(http版本),包含Agent平台如 PAI、阿里云百炼、Dify 等、常见 Agent 开发框架 OpenAI 等

Agent 平台

PAI 集成

人工智能平台PAI是阿里云专为开发者打造的一站式机器学习平台。

阿里云百炼集成

填写插件名称信息查询服务插件描述为「'信息查询服务' 是一个强大的实时搜索 API,可提供来自多种搜索引擎、知识库集合的结构化数据。」,插件 URLhttps://cloud-iqs.aliyuncs.com/search。开启是否鉴权开关,鉴权类型选择服务级鉴权位置选择HeaderType选择bearerToken填写您的 IQS API Key。

  • 参考 联网搜索(搜文)API,以轻量版搜索为例创建工具

    • 工具名称:common-search

    • 工具描述:通用搜索接口提供增强的网络开放域的实时搜索能力,通过大模型优化与多数据源融合的技术,查询干净、准确、多样、高质量的结果。

    • 工具路径:https://cloud-iqs.aliyuncs.com/search/unified

    • 输入参数:

      • query:查询语句(必填参数)

      • engineType:搜索引擎类型(选填参数,业务透传,可参考文档中的参数自行配置)

请求方法选择 POST,提交方式选择 application/jsonquery 参数传参方式设为 大模型识别engineType 参数传参方式设为 业务透传。配置输出参数:参数名称 pageItems,参数描述为 搜索结果,类型为 String

  • 测试完成后发布插件,即可在阿里云百炼应用中使用 '信息查询服务' 插件,根据需求配置业务参数

在应用配置页面的技能区域,单击插件右侧的+按钮添加所需插件;然后在右侧对话预览面板顶部工具栏中,单击柱状图图标进入业务参数配置。

在应用调试界面单击入参变量配置,在弹出的对话框中找到 common-search 插件的 engineType 字段,填入 GenericAdvanced,然后单击确认

Dify 集成

Dify-自定义工具

单击创建自定义工具,在弹窗中填写工具名称,并在 Schema 区域粘贴 OpenAPI 3.1.0 格式的 JSON 定义(也可通过从 URL 中导入按钮导入)。系统自动解析 Schema 并在可用工具表格中列出工具名称、描述、请求方法及路径。

  • 配置鉴权方法 ;

鉴权方法弹窗中,鉴权类型选择 API Key,鉴权头部前缀选择 Custom,键(Key)填写 X-API-Key,值(Value)填写对应的 API 密钥,单击保存

  • 测试接口;

在自定义工具列表中选择目标工具(如通晓),打开测试 GetGenericSearch对话框,鉴权方法选择API Key,在参数区域填写参数名(如 query)及对应值,单击测试按钮,下方将返回 JSON 格式的结构化搜索结果。

Dify-工作流

  • 创建 '信息查询服务' 工作流

创建空白应用 对话框中,选择应用类型为 工作流,填写应用名称及描述,然后单击 创建

  • 添加工作流变量 'query'

设置字段类型为文本,最大长度为 100,并勾选必填,单击保存

  • 添加节点 'HTTP 请求'

配置 HTTP 请求 节点:请求方式选择 GET,填写目标 API 地址;在 HEADERS 区域添加 X-API-Key 并填入实际 API Key;在 PARAMS 区域将工作流变量 query 绑定为查询参数;根据需要设置超时时间与重试策略。

  • 添加节点 '代码执行' - HTTP 请求输出变量转换

配置 HTTP 请求 节点:请求方式为 GET,URL 为 https://cloud-iqs.aliyuncs.com/search/genericSearch,失败时重试 3 次。配置 代码执行 节点:输入变量 body 映射自 HTTP 请求的 body(String 类型),Python3 代码通过 import json 将 body 解析为 JSON 并赋值给输出变量 search_result(Object 类型)。

  • 添加节点 '模板转换' - 将结果转换为大模型文本

{%- if search_result and search_result.pageItems -%}
  {% for pageItem in search_result.pageItems %}
---
- Title: 
{{ pageItem.title }}
- URL: 
{{ pageItem.link }}
- Snippet: 
{{ pageItem.snippet }}
Text: 
{{ pageItem.mainText }}
---
  {% endfor %}
{% endif %}
  • 添加节点 '结束'

结束节点的配置面板中,添加输出变量 search_result,引用来源设置为总结搜索结果节点的 output,类型为 String

  • 测试并发布成工具

单击画布右上角的运行按钮启动测试,查看各节点的执行追踪记录及输出结果,确认工作流运行正常后再发布为工具。

单击右上角发布按钮展开下拉菜单,选择发布为工具

发布为工具弹窗中,填写名称为"通晓",工具调用名称generic_search工具描述为"'通晓' 是一个强大的实时搜索 API,可提供来自多种搜索引擎、知识库集合的结构化数据"。在工具入参表格中,参数名称为 query,类型为 text-input,方式选择LLM 填入,描述为 query to search标签隐私协议可按需填写。单击保存完成发布。

开发集成

在运行以下代码之前,请确认本地已安装 Python 环境以及相关依赖。

OpenAI

调用示例

此处向您展示通过 OpenAI 兼容接口,使用 function call 功能实现一个智能搜索 Agent。

from openai import OpenAI
from datetime import datetime
import json
import os
import requests
class TongxiaoSearch:
    def __init__(self):
        self.BACKEND = "https://cloud-iqs.aliyuncs.com"
        self.headers = {
            "X-API-Key": os.getenv('TONGXIAO_API_KEY')
        }
        self.timeout = 5
    def generic_search(self, query: str):
        url = f"{self.BACKEND}/search/genericSearch?query={query}"
        try:
            response = requests.get(url, headers=self.headers, timeout=self.timeout)
            if response.status_code != requests.codes.ok:
                raise RuntimeError(
                    f"GenericSearch error with error code: {response.status_code}"
                )
            search_resp = json.loads(response.text)
            return search_resp["pageItems"]
        except requests.HTTPError as e:
            raise RuntimeError(f"fail with: {url}, error: {e}")
# 定义工具列表,模型在选择使用哪个工具时会参考工具的name和description
tools = [
    {
        "type": "function",
        "function": {
            "name": "generic_search",
            "description": "提供开放域的实时搜索能力",
            "parameters": { 
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索问题(长度:>=2 and <=100)"
                    }
                }
            },
            "required": [
                "query"
            ]
        }
    }
]
# 信息查询服务-标准搜索
def generic_search(query: str):
    page_items = TongxiaoSearch().generic_search(query)
    page_results = []
    for page_item in page_items:
        if page_item.get('publishTime'):
            publish_date = datetime.fromtimestamp(page_item.get('publishTime') / 1000.0).date()
        else:
            publish_date = ''
        page_result = (
            f"Title: {page_item.get('title', '')}\n"
            f"URL: {page_item.get('link', '')}\n"
            f"Published Date: {publish_date}\n"
            f"Snippet: {page_item.get('snippet', '')}\n"
            f"Text: {page_item.get('mainText', '')}\n\n"
        )
        page_results.append(page_result)
    return "---\n".join(page_results)
client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
def get_response(messages):
    completion = client.chat.completions.create(
        model="qwen-plus",  # 此处以qwen-plus为例,可按需更换模型名称。模型列表:https://help.aliyun.com/zh/model-studio/getting-started/models
        messages=messages,
        tools=tools
        )
    return completion.model_dump()
def call_with_messages():
    messages = [
        {"role": "system", "content": "你是一个 AI 联网搜索引擎,旨在帮助用户在互联网上寻找并总结信息,避免不必要的闲聊,更加专注于内容。\n"
                                      "你回答之前务必先使用搜索工具(generic_search)获取信息。\n"
                                      f"今日日期:{datetime.now().strftime('%a, %b %d, %Y')}"}, 
        {"role": "user", "content": "阿里巴巴"}
    ]
    # 模型的第一轮调用
    i = 1
    first_response = get_response(messages)
    assistant_output = first_response['choices'][0]['message']
    print(f"\n第{i}轮大模型输出信息:{first_response}\n")
    if  assistant_output['content'] is None:
        assistant_output['content'] = ""
    messages.append(assistant_output)
    # 如果不需要调用工具,则直接返回最终答案
    if assistant_output['tool_calls'] == None:  # 如果模型判断无需调用工具,则将assistant的回复直接打印出来,无需进行模型的第二轮调用
        print(f"无需调用工具,我可以直接回复:{assistant_output['content']}")
        return
    # 如果需要调用工具,则进行模型的多轮调用,直到模型判断无需调用工具
    while assistant_output['tool_calls'] != None:
        tool_name = assistant_output['tool_calls'][0]['function']['name']
        if tool_name == 'generic_search':
            tool_info = {"name": "generic_search", "role":"tool"}
            # 提取位置参数信息
            query = json.loads(assistant_output['tool_calls'][0]['function']['arguments'])['query']
            tool_info['content'] = generic_search(query)
        else:
            tool_info = {"role":"tool", "name": tool_name, "content": ""}
        print(f"工具输出信息:\n{tool_info['content']}\n")
        messages.append(tool_info)
        assistant_output = get_response(messages)['choices'][0]['message']
        if  assistant_output['content'] is None:
            assistant_output['content'] = ""
        messages.append(assistant_output)
        i += 1
        print(f"第{i}轮大模型输出信息:{assistant_output}\n")
    print(f"最终答案:{assistant_output['content']}")
if __name__ == '__main__':
    call_with_messages()