如何开发具有自动互联网搜索功能的免费人工智能代理

使用 Groq 的免费 API 和 LangGraph 开发一个 AI 代理，以回答用户的问题并根据问题自动调用互联网搜索

如果您不是 Medium 会员，您可以通过此链接阅读完整故事。

生成性人工智能中的代理 AI 方法正在取得进展，并发现多个潜在应用。AI 代理就像数字助手，可以代表用户执行多项任务。它们可以理解用户的问题、整体目标，推理并决定选择最佳行动，并相互沟通以完成给定任务。例如，一个代理系统可以从您的简历中提取关键信息，另一个代理可以利用这些信息进行全面的互联网搜索，以找到与您的技能和专业匹配的相关工作，而另一个代理可以代表您申请这些工作。

可以利用多种代理框架来开发代理系统。一些著名的代理框架包括 LangGraph、AutoGen 和 CrewAI。实验这些框架需要访问专有的大型语言模型（LLM）的 API，例如 GPT-4o、Claude-3.5、Gemini-2.0 等，或本地运行的开源模型，如 Llama-3.3、Mistral、Phi 等。虽然开源模型可以在本地运行，并且不需要购买任何 API 密钥，但需要一台配备通用图形处理单元（GPU）的强大计算机，以便在合理的时间内从这些模型获得响应。

在我接下来的文章中，我将解释如何使用 Grok 的免费 API 开始使用他们强大的 LLM 进行编码。

在当前文章中，我将解释：

• 如何使用“Groq”的免费 API 密钥在 Groq Cloud 上运行开源模型。
• 如何使用 LangChain 和 LangGraph 开发一个简单的 AI 代理，具有简单的 GUI，能够从其知识库回答用户的问题，并自动调用基于 AI 的网络搜索来回答需要最新信息的问题。

这个 AI 代理可以扩展以处理更复杂的任务和用例。该应用程序的代码位于我的 GitHub 仓库。

如果你喜欢这篇文章并想表达一些爱：

• 鼓掌 50 次——每一次的支持都比你想象的更有帮助！ 👏
• 关注 我在 Medium 上，并免费订阅以获取我的最新文章。 🫶
• 让我们在 LinkedIn 上联系。

Grok 与 Groq

请注意，Grok 和 Groq 是两个不同的 AI 产品。Groq 是一家开发了名为语言处理单元（Language Processing Units，LPUs）的定制 AI 芯片的公司，旨在以闪电般的速度运行大型语言模型（LLMs）。另一方面，Grok 是由 Elon Musk 的 xAI 公司 开发的 LLM 系列和聊天机器人。Grok 不是定制的 AI 芯片，而是使用传统 AI 架构的聊天机器人。

目前，Groq Cloud 提供对以下模型的免费 API 访问。完整代码可在我的 GitHub 仓库 中找到。

开始使用 Groq Cloud 的免费 API

要试验这些模型，您可以通过在 Groq Cloud 创建一个免费账户来获取免费的 API 密钥。以下是如何使用 Groq 的 API 和 ChatGroq 类访问 Groq Cloud 的开源模型 llama-3.3–70b-versatile 的方法。

from langchain_groq.chat_models import ChatGroq
import os

os.environ["GROQ_API_KEY"] = "GROQ_API_KEY" #replace it with your API key
## Initialize the Groq Llama Instant model
groq_llm = ChatGroq(model="llama-3.3–70b-versatile", temperature=0.7)
## Ask a simple question
question = "What is the capital of Finland?"
## Construct a simple prompt
prompt = (
    f"You are a helpful assistant.\n"
    f"Answer the following question clearly and concisely:\n\n"
    f"Question: {question}\n"
    f"Answer:"
)
## Get the response from the model
response = groq_llm.invoke(prompt)
## Print the response
print(response.content.strip())

AI Agents 和 LangGraph

LangGraph 是一个框架，用于设计 AI 代理的工作流程，以图形的形式定义行动或决策的顺序。例如，在我们的示例用例中，代理决定是根据 LLM 的内部知识回答问题，还是需要进行网络搜索以提供最新信息。

AI 代理是工作流程与决策逻辑的结合，能够智能地回答问题或执行其他需要分解为更简单子任务的复杂任务。

LangGraph 以图形的形式实现代理，具有以下主要组件：

• 节点： LangGraph 中的每个节点代表工作流程中的一个步骤（例如，决定查询是否需要互联网搜索或使用模型生成响应）。
• 边： 边连接节点，并定义决策和行动的流动。
• 状态： 它跟踪信息在图形中的移动，以便代理在每个步骤中使用正确的数据。

开发一个使用 LangGraph 的助手 AI 代理

在我们的用例中，AI 代理由以下组成：

1. 决策逻辑（路由函数） 用于分析用户的问题并决定查询是否应该进行网络搜索以获取最新信息，或者直接向 LLM 请求生成答案。

2. 动作（图节点） 代理可以在某个状态下采取的行动，例如，使用网络搜索获取最新信息，或直接使用 LLM 生成答案。

3. 执行工作流（状态图） 确定决策和行动的流程，包括起点和终点，以及从一个状态到另一个状态的转换。

4. 工具（LLM 和 Tavily） 由代理使用以处理响应生成。

在这个例子中，我使用了专为 AI 代理和 LLM 设计的 Tavily 搜索引擎。Tavily 提供免费的 API，可以通过在其网站上创建帐户获得。

让我们首先将 Tavily 和 Groq 的 API 密钥添加到环境变量中，并在 GraphState 类中定义我们的数据结构，以管理和跟踪 LangGraph 工作流中的信息流。它定义了在我们的状态图中节点之间传递的状态结构。

os.environ["TAVILY_API_KEY"]="TAVILY_API_KEY"
os.environ["GROQ_API_KEY"]="GROQ_API_KEY"

class GraphState(TypedDict):
    question: str # 用户问题
    generation: str  # 答案
    websearch_content: str  # 如果有，我们在这里存储 Tavily 搜索结果
    web_flag: str # 知道是否使用了网络搜索来回答问题

随后，我定义了一个路由函数，该函数将问题路由到两个工具之一：i) 网络搜索，或 ii) LLM。以下链条在 ChatPromptTemplate 中传递两个工具的名称和描述，并带有系统提示。如果选择了网络搜索，则 web_flag 状态变量被设置为 “True”，以便工作流中的其他状态相应地决定它们的行动。

def route_question(state: GraphState) -> str:
    question = state["question"]
    web_flag = state.get("web_flag", "False")
    tool_selection = {
    "websearch": (
        "需要最新统计数据、实时信息、最近新闻或当前更新的问题。"
    ),
    "generate": (
        "需要访问大型语言模型的一般知识，但不需要最新统计数据、实时信息、最近新闻或当前更新的问题。"
    )
    }

    SYS_PROMPT = """充当路由器，根据用户的问题选择特定工具或功能，使用以下规则：
                    - 分析给定的问题，并使用给定的工具选择字典，根据其描述和与问题的相关性输出相关工具的名称。 
                    - 字典的键是工具名称，值是其描述。 
                    - 仅输出工具名称，即确切的键，绝对不附加任何解释。 
                """

    # 定义 ChatPromptTemplate
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", SYS_PROMPT),
            ("human", """这是问题：
                        {question}
                        这是工具选择字典：
                        {tool_selection}
                        输出所需的工具。
                    """),
        ]
    )

    # 将输入传递给提示
    inputs = {
        "question": question,
        "tool_selection": tool_selection
    }

    # 调用链
    tool = (prompt | st.session_state.llm | StrOutputParser()).invoke(inputs)
    tool = re.sub(r"[\\'\"`]", "", tool.strip()) # 移除任何反斜杠和多余的空格
    if tool == "websearch":
        state["web_flag"] = "True"
    print(f"通过 {st.session_state.llm.model_name} 调用 {tool} 工具")
    return tool

两个状态函数，websearch 用于使用 Tavily 搜索引擎进行互联网搜索，generate 用于使用 LLM 创建最终响应，如下所示。web_flag 也由 websearch 函数设置为 “True”，该函数被 generate 函数访问，以相应地创建其响应。LLM 仅在上下文包含一些 URL 时引用响应，这意味着上下文是由网络搜索提供的。

#############################################################################
## Websearch 函数从 Tavily 获取上下文，存储在 state["websearch_content"]
#############################################################################
def websearch(state: GraphState) -> GraphState:
    """
    使用 Tavily 在网络上搜索问题，然后将结果附加到 `websearch_content` 中。
    """
    question = state["question"]
    
    if "tavily_client" not in st.session_state:
        st.session_state.tavily_client = TavilyClient()

    try:
        print("执行 Tavily 网络搜索...")
        search_result = st.session_state.tavily_client.get_search_context(
            query=question,
            search_depth="advanced",
            max_tokens=2048
        )
        # tavily_client 可能返回字符串或字典
        if isinstance(search_result, dict) and "documents" in search_result:
            # 合并所有文档内容
            docs = [doc.get("content", "") for doc in search_result["documents"]]
            state["websearch_content"] = "\n\n".join(docs)
            state["web_flag"] = "True"
        else:
            # 如果只是字符串或其他内容
            state["websearch_content"] = str(search_result)
            state["web_flag"] = "True"

    except Exception as e:
        print(f"Tavily 网络搜索期间出错: {e}")
        state["websearch_content"] = f"Tavily 错误: {e}"

    return state

#############################################################################
## 生成函数调用 Groq LLM，可选择包括 websearch 内容
#############################################################################
def generate(state: GraphState) -> GraphState:
    question = state["question"]
    context = state.get("websearch_content", "")
    web_flag = state.get("web_flag", "False")
    if "llm" not in st.session_state:
        raise RuntimeError("LLM 未初始化。请先调用 initialize_app。")

    prompt = (
        "您是一个提供有用信息的助手。\n\n"
        "如果有上下文，请根据上下文回答问题。"
        "如果问题没有上下文，请根据您的知识回答问题。"
        "如果 web_flag 为 'True'，请以以下格式引用上下文中的所有 URL: 来源: \n URL1\n URL2, ..."
        f"问题: {question}\n\n"
        f"上下文: {context}\n\n"
        f"web_flag: {web_flag}"
        "答案:"
        )
    try:
        response = st.session_state.llm.invoke(prompt)
        state["generation"] = response
    except Exception as e:
        state["generation"] = f"生成答案时出错: {str(e)}"

    return state

现在，让我们开发 LangGraph 管道。我们从我们的 GraphState 数据结构创建一个 StateGraph 对象，并向其添加两个节点：websearch 和 generate。我们将问题路由到 websearch 或 generate 函数。从 websearch 状态，下一状态转换将是 generate，然后是 END。工作流中的条件入口点定义了路由逻辑。

#############################################################################
## 构建 LangGraph 管道
#############################################################################
workflow = StateGraph(GraphState)
## 添加节点
workflow.add_node("websearch", websearch)
workflow.add_node("generate", generate)
## 我们将从 "route_question" 路由到 "websearch" 或 "generate"
## 然后从 "websearch" -> "generate" -> END
## 如果不需要搜索，则从 "generate" -> END 直接结束。
workflow.set_conditional_entry_point(
    route_question,  # 路由函数
    {
        "websearch": "websearch",
        "generate": "generate"
    }
)
workflow.add_edge("websearch", "generate")
workflow.add_edge("generate", END)

现在我们最终将开发一个 streamlit 用户界面，以便提问和获取响应。

import streamlit as st
from agent import initialize_app
import sys
import io

## 配置 Streamlit 页面布局
st.set_page_config(
    page_title="LangGraph 聊天机器人",
    layout="wide",
    initial_sidebar_state="expanded",
    page_icon="🤖"
)

## 初始化会话状态以存储消息
if "messages" not in st.session_state:
    st.session_state.messages = []

## 侧边栏布局
with st.sidebar:
    st.title("🤖 LangGraph 聊天机器人")

    # 如果模型的会话状态不存在，则初始化
    if "selected_model" not in st.session_state:
        st.session_state.selected_model = "llama-3.1-8b-instant"

    model_list = [
        "llama-3.1-8b-instant",
        "llama-3.3-70b-versatile",
        "llama3-70b-8192",
        "llama3-8b-8192",
        "mixtral-8x7b-32768",
        "gemma2-9b-it"
    ]

    st.session_state.selected_model = st.selectbox(
        "🤖 选择模型",
        model_list,
        key="model_selector",
        index=model_list.index(st.session_state.selected_model)
    )

    reset_button = st.button("🔄 重置对话", key="reset_button")
    if reset_button:
        st.session_state.messages = []

## 使用所选模型初始化 LangGraph 应用
app = initialize_app(model_name=st.session_state.selected_model)

## 标题和描述
st.title("📘 LangGraph 聊天界面")
st.markdown(
    """
    <div style="text-align: left; font-size: 18px; margin-top: 20px; line-height: 1.6;">
        🤖 <b>欢迎使用 LangGraph 聊天机器人!</b><br>
        我可以通过 AI 驱动的工作流程帮助您回答问题。
        <p style="margin-top: 10px;"><b>开始在下面输入您的问题，我将提供智能响应!</b></p>
    </div>
    """,
    unsafe_allow_html=True
)

## 显示对话历史
for message in st.session_state.messages:
    if message["role"] == "user":
        with st.chat_message("user"):
            st.markdown(f"**您:** {message['content']}")
    elif message["role"] == "assistant":
        with st.chat_message("assistant"):
            st.markdown(f"**助手:** {message['content']}")

## 新消息的输入框
if user_input := st.chat_input("在这里输入您的问题（最多 150 个字符）："):
    if len(user_input) > 150:
        st.error("您的问题超过 150 个字符。请缩短。")
    else:
        # 将用户的消息添加到会话状态并显示
        st.session_state.messages.append({"role": "user", "content": user_input})
        with st.chat_message("user"):
            st.markdown(f"**您:** {user_input}")

        # 捕获 agentic_rag.py 中的打印语句
        output_buffer = io.StringIO()
        sys.stdout = output_buffer  # 将 stdout 重定向到缓冲区

        try:
            with st.chat_message("assistant"):
                response_placeholder = st.empty()
                debug_placeholder = st.empty()
                streamed_response = ""

                # 在流式响应时显示加载指示器
                with st.spinner("思考中..."):
                    inputs = {"question": user_input}
                    for i, output in enumerate(app.stream(inputs)):
                        # 捕获中间打印消息
                        debug_logs = output_buffer.getvalue()
                        debug_placeholder.text_area(
                            "调试日志",
                            debug_logs,
                            height=100,
                            key=f"debug_logs_{i}"
                        )

                        if "generate" in output and "generation" in output["generate"]:
                            chunk = output["generate"]["generation"]

                            # 安全提取文本内容
                            if hasattr(chunk, "content"):  # 如果 chunk 是 AIMessage
                                chunk_text = chunk.content
                            else:  # 否则，转换为字符串
                                chunk_text = str(chunk)

                            # 将文本附加到流式响应中
                            streamed_response += chunk_text

                            # 用到目前为止的流式响应更新占位符
                            response_placeholder.markdown(f"**助手:** {streamed_response}")

                # 将最终响应存储在会话状态中
                st.session_state.messages.append({"role": "assistant", "content": streamed_response or "未生成响应。"})

        except Exception as e:
            # 处理错误并在对话历史中显示
            error_message = f"发生错误: {e}"
            st.session_state.messages.append({"role": "assistant", "content": error_message})
            with st.chat_message("assistant"):
                st.error(error_message)
        finally:
            # 恢复 stdout 到其原始状态
            sys.stdout = sys.__stdout__

我提供了一个选项，可以在运行时选择 Groq Cloud 上的多个可用模型。

可以看出，代理为上面截图中显示的两个查询决定了正确的操作/工具。第一个问题的答案来自LLM的内部知识。而第二个问题的答案则是通过网络搜索提供的。

该代理可以通过多种方式扩展，例如，添加更多功能以从可信来源进行搜索，并集成检索增强生成（RAG）以从特定文档中回答查询等。

这就是全部了，朋友们！ 如果你喜欢这篇文章，请多次点赞（👏），写下评论，并在Medium和LinkedIn上关注我。