集成搜索API

Agent 平台

PAI 集成

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

• LangStudio 通过直观的交互式环境，简化企业级大模型应用的开发流程，参考文档 《联网搜索：Chat With Web Search 》快速构建具备联网搜索能力的 Agent。

• PAI-EAS Spot （一种经济高效的在线推理服务部署方案），部署 eas 时，参考文档 《RAG集成联网搜索搭建AI智能问答最佳实践》集成信息查询服务。

百炼集成

• 前往百炼控制台，创建自定义插件

• 参考 OpenAPI Specification，创建插件与工具

• 测试完成后发布插件，即可在百炼应用中使用 '通晓'。

Dify 集成

Dify-自定义工具

• 使用 OpenAPI Specification 中的 API 定义，创建自定义工具；

• 配置鉴权方法 ;

• 测试接口；

Dify-工作流

• 创建 '通晓' 工作流

• 添加工作流变量 'query'

• 添加节点 'HTTP 请求'

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

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

{%- 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 %}

• 添加节点 '结束'

• 测试并发布成工具

开发集成

在运行以下代码之前，请确认本地已安装 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.params_dict = {
            "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, self.params_dict, 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()