# 导入依赖
import os
from tavily import TavilyClient
# 初始化客户端（环境变量读取密钥，规避硬编码风险）
client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
# 生产级全参数检索调用
result = client.search(
    query="2026 年大模型 RAG 检索增强生成有哪些新进展",  # 核心查询语句
    search_depth="advanced",  # 检索深度：basic快速检索，advanced深度抓取正文
    max_results=5,  # 返回结果数量，生产建议3-5条，平衡精度与token消耗
    include_answer=True,  # 开启内置汇总，直接生成可喂给LLM的精简上下文
    include_raw_content=False,  # 关闭原始HTML，仅保留清洗后文本，减少冗余
    include_domains=["arxiv.org", "zhihu.com", "github.com"],  # 限定权威域名检索
    exclude_domains=["csdn.net"],  # 拉黑低质量站点，过滤噪音
    topic="general",  # 检索类型：general通用、news实时新闻
    days=30,  # 时间窗口，仅检索近30天内容，适配时效性场景
)
# 打印Tavily内置汇总答案，可直接作为LLM上下文
print("=== Tavily 智能汇总答案 ===")
print(result["answer"])
# 逐条输出检索详情，包含分数、来源、摘要
print("\n=== 权威检索结果详情 ===")
for i, item in enumerate(result["results"], 1):
    print(f"\n[{i}] 标题：{item['title']}")
    print(f"    链接：{item['url']}")
    print(f"    可信度分数：{item['score']:.2f}")
    print(f"    内容摘要：{item['content'][:200]}...")

from tavily import TavilyClient
# 初始化客户端
client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
# 批量解析多个权威URL
resp = client.extract(urls=[
    "https://arxiv.org/abs/2312.10997",
    "https://www.tavily.com/documentation/api"
])
# 遍历输出纯净正文
for item in resp["results"]:
    print("检索链接：", item["url"])
    print("纯净正文长度：", len(item["raw_content"]))
    print("正文预览：", item["raw_content"][:300], "...\n")

import os
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain.agents import create_agent
# 1. 初始化大模型（腾讯混元，OpenAI 兼容协议，从环境变量读取密钥）
llm = ChatOpenAI(
    model="hy3-preview",
    api_key=os.environ.get("TENCENT_API_KEY", "sk-*"),
    base_url="https://tokenhub.tencentmaas.com/v1",
    temperature=0,
    max_tokens=1024,
)
# 2. 封装Tavily为Agent专属工具
#    工具类只认环境变量 TAVILY_API_KEY，或参数 tavily_api_key（优先显式传参，更直观）
tools = [
    TavilySearchResults(
        max_results=3,
        search_depth="advanced",
        tavily_api_key=os.environ.get("TAVILY_API_KEY", "tvly-dev-**"),
    )
]
# 3. 组装智能Agent（langchain 1.x 新 API：create_agent，内部基于 LangGraph）
#    系统提示词通过 system_prompt 传入，替代旧版 ChatPromptTemplate + agent_scratchpad
agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt="你是专业智能助手，遇到时效性、未知事实类问题，必须调用搜索工具查证后再回答，禁止编造信息。",
)
# 4. 测试实时性问题，触发自动联网检索
result = agent.invoke(
    {"messages": [{"role": "user", "content": "2026年大模型RAG领域最新研究成果有哪些？"}]}
)
# 5. 输出最终回答（最后一条 AI 消息）
print("\n===== 最终回答 =====")
print(result["messages"][-1].content)

# 安装依赖
# pip install llama-index llama-index-tools-tavily
from llama_index.core import VectorStoreIndex, Document
from llama_index.tools.tavily_reasoning.base import TavilyToolSpec
from llama_index.core.agent import FunctionCallingAgent
from llama_index.llms.openai import OpenAI
# 1. 构建本地私有知识库示例
local_docs = [Document(text="公司2026年出差报销制度：住宿上限500元/晚，交通实报实销")]
local_index = VectorStoreIndex.from_documents(local_docs)
local_query_engine = local_index.as_query_engine()
# 2. 初始化外网Tavily检索工具
tavily_tool = TavilyToolSpec(api_key=os.getenv("TAVILY_API_KEY")).to_tool_list()
# 3. 组装双数据源智能Agent
agent = FunctionCallingAgent.from_tools(
    [*tavily_tool, local_query_engine.as_tool()],
    llm=OpenAI(model="gpt-4o-mini"),
    verbose=True,
)
# 4. 混合场景测试：本地私有数据+外网公开数据
agent.chat("公司出差住宿标准是多少？对比2026年北京酒店均价水平如何？")

# 安装依赖
# pip install searxng-python trafilatura requests
import requests
import trafilatura
# 本地SearXNG服务地址（Docker一键部署）
SEARXNG_URL = "http://localhost:8080/search"
def searx_search(query: str, num_results: int = 5):
    """聚合百度、必应、搜狗多源中文检索"""
    resp = requests.get(
        SEARXNG_URL,
        params={"q": query, "format": "json", "language": "zh"},
        timeout=10,
    )
    return resp.json().get("results", [])[:num_results]
def extract_text(url: str) -> str:
    """中文网页精准正文清洗"""
    downloaded = trafilatura.fetch_url(url)
    if not downloaded:
        return ""
    return trafilatura.extract(
        downloaded,
        include_links=False,
        include_images=False,
        favor_recall=True  # 适配中文长文本抽取
    ) or ""
def my_tavily_like_search(query: str):
    """自建迷你中文检索工具，复刻Tavily核心能力"""
    results = searx_search(query)
    for item in results:
        item["content"] = extract_text(item["url"])
    # 过滤空内容噪音
    return [r for r in results if r["content"]]
# 实战调用
for r in my_tavily_like_search("RAG大模型最新综述论文"):
    print(f"标题：{r['title']}\n链接：{r['url']}\n摘要：{r['content'][:200]}\n")

from typing import List, Dict
class ChineseWebSearcher:
    """最小可运行中文联网检索骨架"""
    def __init__(self, llm_client, reranker):
        self.llm = llm_client       # 大模型用于查询改写
        self.reranker = reranker   # BGE重排模型用于结果打分
    def rewrite_query(self, user_question: str) -> List[str]:
        """将自然语言拆解为精准检索关键词"""
        prompt = f"将以下问题改写为3个适配搜索引擎的关键词组合：\n{user_question}"
        return self.llm.chat(prompt).split("\n")
    def rerank(self, query: str, docs: List[Dict]) -> List[Dict]:
        """基于BGE模型重排打分，过滤无效结果"""
        pairs = [(query, d["content"]) for d in docs]
        scores = self.reranker.compute_score(pairs)
        for d, s in zip(docs, scores):
            d["score"] = s
        return sorted(docs, key=lambda x: x["score"], reverse=True)
    def run(self, question: str, top_k: int = 5) -> List[Dict]:
        """完整检索链路执行"""
        queries = self.rewrite_query(question)
        all_docs = []
        # 多子查询并行检索
        for q in queries:
            all_docs.extend(self.search(q))
        # 结果去重
        seen = set()
        deduped_docs = []
        for doc in all_docs:
            if doc["url"] not in seen:
                seen.add(doc["url"])
                deduped_docs.append(doc)
        # 重排筛选
        return self.rerank(question, deduped_docs)[:top_k]

import hashlib
import time
class HybridRAG:
    """生产级混合检索RAG架构"""
    def __init__(self, vector_store, search_client, score_threshold=0.7):
        self.vs = vector_store         # 本地私有向量库
        self.search = search_client    # 联网检索客户端
        self.threshold = score_threshold  # 本地匹配分数阈值
        self._cache = {}               # 内存缓存
    def _cache_key(self, q: str) -> str:
        """生成查询唯一缓存key"""
        return hashlib.md5(q.encode()).hexdigest()
    def retrieve(self, question: str):
        # 1. 优先读取缓存，减少重复调用
        key = self._cache_key(question)
        if key in self._cache and time.time() - self._cache[key][0] < 600:
            return self._cache[key][1]
        # 2. 优先检索本地私有知识库
        local_hits = self.vs.similarity_search_with_score(question, k=5)
        best_score = local_hits[0][1] if local_hits else 0
        context = []
        # 3. 本地匹配度足够，直接复用本地结果，节省联网成本
        if best_score >= self.threshold:
            context = [doc.page_content for doc, _ in local_hits]
        else:
            # 4. 本地知识不足，联网检索兜底
            web_hits = self.search.search(
                question, max_results=3, search_depth="advanced"
            )["results"]
            # 过滤低分噪音数据
            context = [h["content"] for h in web_hits if h["score"] >= 0.5]
            # 本地+外网内容互补，提升回答完整性
            context += [doc.page_content for doc, _ in local_hits[:2]]
        # 5. 写入缓存，10分钟有效期
        self._cache[key] = (time.time(), context)
        return context
# 架构调用示例
# hybrid = HybridRAG(vector_store=my_vector_db, search_client=TavilyClient(...))
# context = hybrid.retrieve("2026杭州购房政策变化")