LangChain 集成

import requests
from langchain.agents import initialize_agent, Tool, AgentType
from langchain_openai import ChatOpenAI

OPENAI_API_KEY = "YOUR-OPENAI-API-KEY"
ZIXIANG_API_KEY = "YOUR-ZIXIANG-API-KEY"

# 定义梓享搜索工具
@tool
def zixiang_websearch_tool(query: str, count: int = 10) -> str:
    """
    使用梓享 AI 搜索进行网页搜索。

    参数:
        - query: 搜索关键词
        - count: 返回的搜索结果数量

    返回:
        - 搜索结果的详细信息，包括网页标题、网页URL、网页摘要、网站名称、网站Icon、网页发布时间等。
    """

    url = 'https://search.aiserver.cloud/v1/api'
    headers = {
        'Authorization': f'Bearer {ZIXIANG_API_KEY}',
        'Content-Type': 'application/json'
    }
    data = {
        "query": query,
        "engine": "china",
        "max_results": count
    }

    response = requests.post(url, headers=headers, json=data)

    if response.status_code == 200:
        json_response = response.json()
        try:
            if json_response["code"] != 0 or not json_response["data"]:
                return f"搜索API请求失败，原因是: {json_response.get('msg', '未知错误')}"

            webpages = json_response["data"]["results"]
            if not webpages:
                return "未找到相关结果。"

            formatted_results = ""
            for idx, page in enumerate(webpages, start=1):
                formatted_results += (
                    f"引用: {idx}\n"
                    f"标题: {page['title']}\n"
                    f"URL: {page['url']}\n"
                    f"摘要: {page['content']}\n"
                    f"网站名称: {page.get('siteName', '')}\n"
                    f"网站图标: {page.get('siteIcon', '')}\n"
                    f"发布时间: {page.get('dateLastCrawled', '')}\n\n"
                )

            return formatted_results.strip()
        except Exception as e:
            return f"搜索API请求失败，原因是: 搜索结果解析失败 {str(e)}"
    else:
        return f"搜索API请求失败，状态码: {response.status_code}, 错误信息: {response.text}"

# 创建 LangChain 工具
zixiang_tool = Tool(
    name="ZixiangWebSearch",
    func=zixiang_websearch_tool,
    description="使用梓享 AI 搜索进行联网搜索，输入应为搜索查询字符串，输出将返回搜索结果的详细信息，包括网页标题、网页URL、网页摘要、网站名称、网站Icon、网页发布时间等。"
)

# 初始化模型和 Agent
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    openai_api_key=OPENAI_API_KEY
)

# 初始化代理，包含您的自定义工具
agent = initialize_agent(
    tools=[zixiang_tool],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 使用代理进行查询
user_question = "请告诉我2024年人工智能技术的最新进展有哪些亮点"
response = agent.run(user_question)
print(response)

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA

# 创建本地知识库工具
vectorstore = FAISS.load_local("./company_docs", OpenAIEmbeddings())
local_qa = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(temperature=0),
    retriever=vectorstore.as_retriever()
)

local_tool = Tool(
    name="LocalKnowledgeBase",
    func=local_qa.run,
    description="查询公司内部文档、产品手册、技术规范等固定知识"
)

# 创建 Agent（同时使用本地和互联网搜索）
agent = initialize_agent(
    tools=[local_tool, zixiang_tool],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# Agent 会自动选择合适的工具
agent.run("我们公司的技术架构是什么？")  # 使用本地知识库
agent.run("2024年行业最新技术趋势？")  # 使用互联网搜索

def zixiang_search_with_engine(query: str, engine: str = "china") -> str:
    """支持选择引擎的搜索功能"""
    url = "https://search.aiserver.cloud/v1/api"
    headers = {
        "Authorization": "Bearer YOUR-API-KEY",
        "Content-Type": "application/json"
    }
    data = {
        "engine": engine,  # china 或 global
        "query": query,
        "max_results": 5
    }

    response = requests.post(url, headers=headers, json=data)
    result = response.json()

    if result.get("code") == 0:
        results = result["data"]["results"]
        return "\n\n".join([
            f"标题: {item['title']}\n链接: {item['url']}\n摘要: {item['content']}"
            for item in results
        ])
    return f"搜索失败: {result.get('msg')}"

# 创建中文搜索工具
china_search_tool = Tool(
    name="ChinaSearch",
    func=lambda q: zixiang_search_with_engine(q, "china"),
    description="搜索中文互联网信息，适合查询国内内容"
)

# 创建全球搜索工具
global_search_tool = Tool(
    name="GlobalSearch",
    func=lambda q: zixiang_search_with_engine(q, "global"),
    description="搜索全球互联网信息，支持多语言"
)

# 使用两个工具
agent = initialize_agent(
    tools=[china_search_tool, global_search_tool],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)