Improve the timeliness of search results

When an agent invokes a search tool, the results may be outdated because of the search engine's update frequency and caching mechanism. For scenarios that are not sensitive to latency, you can scrape web content in real time to obtain more current results. The following example shows how to use LangChain to improve search timeliness.

1. Implement a search tool with enhanced timeliness. Because the ReadPage API has high latency, using this method to improve timeliness significantly increases the time required for a search.

import asyncio
import httpx
from langchain_core.tools import tool

async def iqs_search(query: str, max_results: int = 3) -> list[dict]:
    async with httpx.AsyncClient() as client:
        response = await client.post("https://cloud-iqs.aliyuncs.com/search/unified",
                                     headers={
                                         'X-API-Key': "<Your-IQS-ApiKey>"
                                     },
                                     json={
                                         'query': query,
                                         'engineType': "Generic",
                                     })
        if not response.is_success:
            return None

        pageItems = response.json().get("pageItems")
        return [
            {
                "link": item.get("link"),
                "title": item.get("title"),
                "snippet": item.get("snippet"),
                "publishedTime": item.get("publishedTime")
            } for item in pageItems[:max_results]
        ]


async def iqs_readpage(urls: list[str]) -> list[dict]:
    async def post_readpage(client, url):
        response = await client.post("https://cloud-iqs.aliyuncs.com/readpage/scrape",
                                     headers={
                                         'X-API-Key': "<Your-IQS-ApiKey>"
                                     },
                                     json={
                                         "url": url,
                                         "formats": ["markdown"],
                                         "timeout": 15000,
                                         "pageTimeout": 10000,
                                         "maxAge": 3600
                                     }, timeout=20000)
        if not response.is_success:
            return None

        page_result = response.json()
        data = page_result.get("data")
        if not data:
            return None
        return {
            "url": data.get("url"),
            "markdown": data.get("markdown")
        }

    tasks = []
    async with httpx.AsyncClient() as client:
        for url in urls:
            tasks.append(post_readpage(client, url))
        results = await asyncio.gather(*tasks)
    return results


@tool
async def search_freshness_boost(query: str):
    """Search the web for information.
    Args:
        query: Search query
    """
    page_results = await iqs_search(query, 3)
    links = [page.get("link") for page in page_results]
    read_results = await iqs_readpage(links)
    for idx, page_result in enumerate(page_results):
        if read_results[idx]:
            page_result['snippet'] = read_results[idx]["markdown"]
    return page_results

2. The agent uses the enhanced search tool.

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models import ChatOpenAI, ChatTongyi
from langchain.agents import AgentExecutor, create_tool_calling_agent


async def main():
    llm = ChatTongyi(
        api_key="<Bailian-ApiKey>",
        model="qwen-plus",
    )

    tools = [search_freshness_boost]
    agent = create_tool_calling_agent(llm, tools,
                                      ChatPromptTemplate.from_messages([
                                          ("system", "You are an AI assistant with real-time web search capabilities."),
                                          MessagesPlaceholder(variable_name="chat_history", optional=True),
                                          ("human", "{input}"),
                                          MessagesPlaceholder(variable_name="agent_scratchpad"),
                                      ])
                                      )

    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,
        handle_parsing_errors=True
    )
    result = await agent_executor.ainvoke({
        "input": "today's oil price"
    })
    print("\nFinal Answer:", result["output"])


if __name__ == "__main__":
    asyncio.run(main())