当前位置：首页 > article >正文

使用LangChain构建第一个ReAct Agent

article 2025/2/23 5:29:09

使用LangChain构建第一个ReAct Agent

准备环境

使用Anaconda
安装python 3.10
安装langchain、langchain_openai、langchain_community （安装命令 pip install XXX）
申请DeepSeek API：https://platform.deepseek.com/api_keys（也可以用kimi：https://platform.moonshot.cn/console/api-keys）

代码

from langchain import hub
from langchain.agents import create_structured_chat_agent, AgentExecutor
from langchain.memory import ConversationBufferMemory
from langchain.schema import HumanMessage
from langchain.tools import BaseTool
from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, Tool

# 模型 DeepSeek
# model = ChatOpenAI(model="deepseek-chat",
#                    openai_api_key="xx", #填写自己DeepSeek的api
#                    openai_api_base="https://api.deepseek.com/v1")

# kimi
model = ChatOpenAI(model="moonshot-v1-8k",
                   openai_api_key="sk-xx", #填写自己kimi的api
                   openai_api_base="https://api.moonshot.cn/v1")

from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

api_wrapper = WikipediaAPIWrapper(top_k_results=3, doc_content_chars_max=2000)
tool = WikipediaQueryRun(api_wrapper=api_wrapper)

# 提示词，直接从langchain hub上下载，因为写这个ReAct机制的prompt比较复杂，直接用现成的。
prompt = hub.pull("hwchase17/structured-chat-agent")

print(prompt)

tools = [tool]
# 定义AI Agent
agent = create_structured_chat_agent(
    llm=model,
    tools=tools,
    prompt=prompt
)
# 使用Memory记录上下文
memory = ConversationBufferMemory(
    memory_key='chat_history',
    return_messages=True
)
# 定义AgentExecutor，必须使用AgentExecutor，才能执行代理定义的工具
agent_executor = AgentExecutor.from_agent_and_tools(
    agent=agent, tools=tools, memory=memory, verbose=True, handle_parsing_errors=True
)
# 测试使用到工具的场景
agent_executor.invoke({"input": "苹果手机一共发布了几代？"})

运行代码，得到以下结果：
在这里插入图片描述
注意：需要连接外网

国内版

如果无法连接外网，可以使用国内版
注册：https://open.bochaai.com/
创建key
在这里插入图片描述

import requests
from langchain.agents import initialize_agent, Tool, AgentType
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain import hub
from langchain.agents import create_structured_chat_agent, AgentExecutor
from langchain.memory import ConversationBufferMemory

OPENAI_API_KEY = "sk-xx" # kimi的key
BOCHA_API_KEY = "sk-xx" # 博查的key

# 定义Bocha Web Search工具
@tool
def bocha_websearch_tool(query: str, count: int = 10) -> str:
    """
    使用Bocha Web Search API 进行网页搜索。

    参数:
    - query: 搜索关键词
    - freshness: 搜索的时间范围
    - summary: 是否显示文本摘要
    - count: 返回的搜索结果数量

    返回:
    - 搜索结果的详细信息，包括网页标题、网页URL、网页摘要、网站名称、网站Icon、网页发布时间等。
    """
    
    url = 'https://api.bochaai.com/v1/web-search'
    headers = {
        'Authorization': f'Bearer {BOCHA_API_KEY}',  # 请替换为你的API密钥
        'Content-Type': 'application/json'
    }
    data = {
        "query": query,
        "freshness": "noLimit", # 搜索的时间范围，例如 "oneDay", "oneWeek", "oneMonth", "oneYear", "noLimit"
        "summary": True, # 是否返回长文本摘要
        "count": count
    }

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

    if response.status_code == 200:
        json_response = response.json()
        try:
            if json_response["code"] != 200 or not json_response["data"]:
                return f"搜索API请求失败，原因是: {response.msg or '未知错误'}"
            
            webpages = json_response["data"]["webPages"]["value"]
            if not webpages:
                return "未找到相关结果。"
            formatted_results = ""
            for idx, page in enumerate(webpages, start=1):
                formatted_results += (
                    f"引用: {idx}\n"
                    f"标题: {page['name']}\n"
                    f"URL: {page['url']}\n"
                    f"摘要: {page['summary']}\n"
                    f"网站名称: {page['siteName']}\n"
                    f"网站图标: {page['siteIcon']}\n"
                    f"发布时间: {page['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工具
bocha_tool = Tool(
    name="BochaWebSearch",
    func=bocha_websearch_tool,
    description="使用Bocha Web Search API 进行搜索互联网网页，输入应为搜索查询字符串，输出将返回搜索结果的详细信息，包括网页标题、网页URL、网页摘要、网站名称、网站Icon、网页发布时间等。"
)

# 提示词，直接从langchain hub上下载，因为写这个ReAct机制的prompt比较复杂，直接用现成的。
prompt = hub.pull("hwchase17/structured-chat-agent")

# model = ChatOpenAI(model="deepseek-chat",
#                    openai_api_key="1", #填写自己DeepSeek的key
#                    openai_api_base="https://api.deepseek.com/v1")
model = ChatOpenAI(model="moonshot-v1-8k",
                   openai_api_key=OPENAI_API_KEY, #填写自己kimi的key
                   openai_api_base="https://api.moonshot.cn/v1")
tools = [bocha_tool]
# 定义AI Agent
agent = create_structured_chat_agent(
    llm=model,
    tools=tools,
    prompt=prompt
)
# 使用Memory记录上下文
memory = ConversationBufferMemory(
    memory_key='chat_history',
    return_messages=True
)
# 定义AgentExecutor，必须使用AgentExecutor，才能执行代理定义的工具
agent_executor = AgentExecutor.from_agent_and_tools(
    agent=agent, tools=tools, memory=memory, verbose=True, handle_parsing_errors=True
)
# 测试使用到工具的场景
agent_executor.invoke({"input": "5090发售时间是什么时候？"})