不需要RAG！在30分钟内构建一个问答AI智能体。万字长文，先收藏再看！

# 传统 RAG 流水线（简化版）def rag_pipeline(question):# 步骤 1：将问题转换为嵌入    question_embedding = embedding_model.encode(question)
# 步骤 2：搜索向量数据库    similar_chunks = vector_db.similarity_search(        question_embedding,        top_k=5)
# 步骤 3：从文档块构建上下文    context ="\n".join([chunk.content for chunk in similar_chunks])
# 步骤 4：生成答案    prompt = f"Context: {context}\n\nQuestion: {question}\nAnswer:"return llm.generate(prompt)

📁项目结构├── main.py           # 🌐 FastAPI 网络服务器（入口）├── qa_agent.py       # 🧠 核心代理逻辑（编排者）├── search_tool.py    # 🔍 搜索功能├── scraping_tool.py  # 🕷️ 网页抓取功能└── sites_data.csv    # 📋 域名配置

"""特定领域的Tavily搜索工具"""import loggingfrom typing importList,Optional,Type,Anyfrom pydantic importBaseModel,Field,ConfigDictfrom langchain.tools importBaseToolfrom langchain_google_genai importChatGoogleGenerativeAIfrom tavily importTavilyClient
logging.basicConfig(level=logging.INFO)logger = logging.getLogger(__name__)
classTavilySearchInput(BaseModel):    query: str =Field(description="包含相关关键词的搜索查询")    sites:List[str]=Field(        description="要搜索的网站域名（例如，['docs.langchain.com']）")    max_results:Optional[int]=Field(default=None, description="返回的最大结果数")    depth:Optional[str]=Field(default=None, description="搜索深度：'basic' 或 'advanced'")
classTavilyDomainSearchTool(BaseTool):"""使用 Tavily 搜索特定域名"""
    name: str ="search_documentation"    description: str ="""使用 Tavily 网络搜索特定文档网站。
必需参数：- query (字符串)：包含相关关键词的搜索查询-你想查找的内容- sites (列表)：要在其中搜索的网站域名（例如，['docs.langchain.com','fastapi.tiangolo.com']）
可选参数：- max_results (整数)：返回的最大搜索结果数（默认：10）- depth (字符串)：搜索深度-'basic'用于快速搜索，'advanced'用于全面搜索（默认：'basic'）
使用指南：1.从用户问题中创建富含关键词的搜索查询2.根据提到的技术选择相关网站域名3.使用'basic'深度进行快速回答，'advanced'进行深入研究4.根据答案的全面性需求调整 max_results
示例：-快速搜索：query="LangChain 自定义工具", sites=["docs.langchain.com"], depth="basic", max_results=5-全面搜索：query="FastAPI 认证中间件", sites=["fastapi.tiangolo.com"], depth="advanced", max_results=15
最佳实践：-在查询中包含技术术语和框架名称-根据问题上下文选择合适的域名-优先选择官方文档网站而非第三方来源-使用具体查询而非宽泛术语以获得更好结果"""    args_schema:Type[BaseModel]=TavilySearchInput
    tavily_client:Any=Field(default=None, exclude=True)    api_key: str =Field(exclude=True)    default_max_results:int=Field(default=10, exclude=True)    default_depth: str =Field(default="basic", exclude=True)    max_content_size:int=Field(default=10000, exclude=True)    enable_summarization:bool=Field(default=False, exclude=True)    summarizer_llm:Any=Field(default=None, exclude=True)
    model_config =ConfigDict(arbitrary_types_allowed=True)
def __init__(self,        api_key: str,        max_results:int=10,        depth: str ="basic",        max_content_size:int=10000,        enable_summarization:bool=False,        google_api_key:Optional[str]=None,):super().__init__(            api_key=api_key,            default_max_results=max_results,            default_depth=depth,            max_content_size=max_content_size,            enable_summarization=enable_summarization,            args_schema=TavilySearchInput,)
ifnot api_key:raiseValueError("TAVILY_API_KEY 是必需的")
object.__setattr__(self,"tavily_client",TavilyClient(api_key=api_key))
if enable_summarization and google_api_key:            summarizer = create_summarizer_llm(google_api_key)object.__setattr__(self,"summarizer_llm", summarizer)            logger.info("🧠 使用 Gemini Flash-Lite 启用搜索结果摘要")elif enable_summarization:            logger.warning("⚠️ 摘要功能已禁用：未提供 google_api_key")object.__setattr__(self,"enable_summarization",False)
        logger.info(            f"Tavily 搜索工具已初始化（摘要功能：{'启用' if self.enable_summarization else '禁用'}）")
    async def _search_async(self, query: str, sites:List[str], max_results:int=None, depth: str =None)-> str:"""异步执行给定参数的搜索"""try:            final_max_results = max_results orself.default_max_results            final_depth = depth orself.default_depth
            logger.info(f"🔍 搜索：'{query}' 在网站：{sites}")            logger.info(                f"📊 参数：max_results={final_max_results}, depth={final_depth}")
# 注意：TavilyClient 尚不支持异步方法，因此在线程中运行            search_results = await asyncio.to_thread(self.tavily_client.search,                query=query,                max_results=final_max_results,                search_depth=final_depth,                include_domains=sites,)
            logger.info(f"📥 收到 {len(search_results.get('results', []))} 个结果")
ifnot search_results.get("results"):                logger.warning("⚠️ 未返回搜索结果")return"未找到结果。请尝试不同的搜索查询或检查域名是否可访问。"
            formatted_results = format_search_results(                search_results["results"][:final_max_results],self.max_content_size)            final_result ="\n".join(formatted_results)
            logger.info(                f"✅ 处理了 {len(search_results['results'])} 个结果，返回 {len(final_result)} 个字符")
ifself.enable_summarization andself.summarizer_llm:try:                    logger.info("🧠 正在摘要结果...")                    summarized_result = await self._summarize_results_async(final_result, query)                    reduction = round((1- len(summarized_result)/ len(final_result))*100)                    logger.info(                        f"📊 摘要：{len(final_result)} → {len(summarized_result)} 字符（减少 {reduction}%）")return summarized_resultexceptExceptionas e:                    logger.error(                        f"❌ 摘要失败：{e}。返回原始结果。")
return final_result
exceptExceptionas e:            error_msg = f"❌ 搜索错误：{str(e)}"            logger.error(error_msg)return error_msg
    async def _summarize_results_async(self, search_results: str, original_query: str)-> str:"""使用 LLM 异步摘要搜索结果"""try:            prompt = create_summary_prompt(search_results, original_query)            response = await asyncio.to_thread(self.summarizer_llm.invoke, prompt)return response.contentexceptExceptionas e:            logger.error(f"LLM 摘要失败：{e}")return search_results
def _run(self, query: str, sites:List[str], max_results:int=None, depth: str =None)-> str:"""执行给定参数的搜索"""return asyncio.run(self._search_async(query, sites, max_results, depth))
    async def _arun(self, query: str, sites:List[str], max_results:int=None, depth: str =None)-> str:"""异步搜索版本"""return await self._search_async(query, sites, max_results, depth)

"""使用Chromium进行动态内容提取的网页抓取工具"""
import loggingimport asynciofrom typing importList,Type
from pydantic importBaseModel,Field,ConfigDictfrom langchain.tools importBaseToolfrom langchain_community.document_loaders importAsyncChromiumLoaderfrom langchain_community.document_transformers importBeautifulSoupTransformer
logging.basicConfig(level=logging.INFO)logger = logging.getLogger(__name__)
def get_default_tags()->List[str]:"""获取用于网页抓取的默认 HTML 标签"""return["p","li","div","a","span","h1","h2","h3","h4","h5","h6"]
classWebScrapingInput(BaseModel):    url: str =Field(description="要抓取的 URL")    tags_to_extract:List[str]=Field(        default_factory=get_default_tags, description="要提取的 HTML 标签")
classWebScrapingTool(BaseTool):"""当搜索结果不足时抓取网站内容"""
    name: str ="scrape_website"    description: str ="""使用 Chromium 浏览器抓取完整网站内容以进行全面页面提取。
必需参数：- url (字符串)：要抓取的完整 URL（必须包含 https:// 或 http://）
可选参数：- tags_to_extract (列表)：要提取内容的 HTML 标签默认：["p","li","div","a","span","h1","h2","h3","h4","h5","h6"]自定义示例：["pre","code"]用于代码示例，["table","tr","td"]用于表格
何时使用：-搜索结果不完整或不足-需要包括代码示例的完整页面内容-页面有搜索未捕捉的动态JavaScript内容-需要搜索未捕捉的特定格式或结构
示例：-基本抓取：url="https://docs.langchain.com/docs/modules/agents"-代码聚焦抓取：url="https://fastapi.tiangolo.com/tutorial/", tags_to_extract=["pre","code","p"]-表格提取：url="https://docs.python.org/3/library/", tags_to_extract=["table","tr","td","th"]
最佳实践：-仅在 search_documentation 提供的信息不足时使用-优先选择来自先前搜索结果的 URL 以确保相关性-使用特定标签提取以获得目标内容（处理更快）-注意：比搜索慢约3-10倍，为性能起见谨慎使用
限制：-内容会在配置的限制处截断以防止过多令牌使用-某些网站可能阻止自动化抓取-比搜索慢-仅在搜索不足时使用"""    args_schema:Type[BaseModel]=WebScrapingInput
    max_content_length:int=Field(default=10000, exclude=True)    model_config =ConfigDict(arbitrary_types_allowed=True)
def __init__(self, max_content_length:int=10000):super().__init__(            max_content_length=max_content_length, args_schema=WebScrapingInput)
    async def _process_scraping(self, url: str, tags_to_extract:List[str]=None, is_async:bool=True)-> str:"""同步和异步抓取的通用逻辑"""try:if tags_to_extract isNone:                tags_to_extract = get_default_tags()
            loader =AsyncChromiumLoader([url])
if is_async:                html_docs = await asyncio.to_thread(loader.load)else:                html_docs = loader.load()
ifnot html_docs:return f"无法从 {url} 加载内容"
            bs_transformer =BeautifulSoupTransformer()
if is_async:                docs_transformed = await asyncio.to_thread(                    bs_transformer.transform_documents,                    html_docs,                    tags_to_extract=tags_to_extract,)else:                docs_transformed = bs_transformer.transform_documents(                    html_docs,                    tags_to_extract=tags_to_extract,)
ifnot docs_transformed:return f"无法从 {url} 提取内容"
            content = docs_transformed[0].page_content
if len(content)>self.max_content_length:                content =(                    content[:self.max_content_length]+"\n\n... (内容已截断)")
return f"""**抓取的网站：**{url}**提取的内容：**
{content}
**注意：**完整的网站内容用于全面分析。"""
exceptExceptionas e:return f"网页抓取错误 {url}：{str(e)}"
def _run(self, url: str, tags_to_extract:List[str]=None)-> str:"""抓取网站内容"""return asyncio.run(self._process_scraping(url, tags_to_extract, is_async=False))
    async def _arun(self, url: str, tags_to_extract:List[str]=None)-> str:"""异步抓取版本"""return await self._process_scraping(url, tags_to_extract, is_async=True)

"""具有特定领域网络搜索能力的问答代理"""
import loggingimport pandas as pdfrom typing importList,Dict,Any,Optional
from langchain.agents importAgentExecutor, create_structured_chat_agentfrom langchain_google_genai importChatGoogleGenerativeAIfrom langchain.prompts importChatPromptTemplate,MessagesPlaceholderfrom langchain.schema importBaseMessage,HumanMessage,AIMessage
from search_tool importTavilyDomainSearchToolfrom scraping_tool importWebScrapingTool
logging.basicConfig(level=logging.INFO)logger = logging.getLogger(__name__)
def create_system_prompt(knowledge_sources_md: str, domains:List[str])-> str:"""创建包含知识来源的系统提示"""return f"""你是一个专门搜索特定文档网站的问答代理。
可用知识来源按类别/领域/主题划分，包含每个类别的网站和描述：{knowledge_sources_md}
指令：1.对于任何问题，始终首先使用 search_documentation 工具2.分析用户问题以确定相关的领域/主题/类别3.根据提到的技术/主题选择合适的网站4.如果搜索结果不足以完全回答问题，则在搜索结果中最相关的 URL 上使用 scrape_website 工具5.你只能回答关于可用知识来源的问题：{domains}6.如果问题超出可用知识来源，不要回答问题，并建议可以回答的主题
工具使用策略：-首先：使用 search_documentation 快速查找相关信息-其次：如果搜索结果不完整、不清晰或信息不足以回答问题，则在搜索结果中最有希望的 URL 上使用 scrape_website-为效率起见，始终优先搜索而非抓取，但在搜索结果无相关信息时始终使用抓取
规则：-保持帮助性和全面性-尽可能引用来源-仅在搜索结果无法提供答案时使用抓取-抓取时，选择来自先前搜索结果的最相关 URL
你可以使用以下工具：
{{tools}}
通过提供 action 键（工具名称）和 action_input 键（工具输入）使用 JSON blob 指定工具。
有效“action”值：“FinalAnswer”或{{tool_names}}
每次 $JSON_BLOB 仅提供一个动作，如下所示：

遵循以下格式：
问题：要回答的输入问题思考：考虑前后的步骤动作：

观察：动作结果...（重复思考/动作/观察 N 次）思考：我知道如何回应动作：

开始！提醒：始终以单个动作的有效 JSON blob 响应。如有需要使用工具。如果合适直接响应，如果不明确则要求澄清。格式为动作：```$JSON_BLOB```然后观察"""
classDomainQAAgent:"""根据用户查询搜索特定域的问答代理"""
def __init__(self,        csv_file_path: str ="sites_data.csv",        config:Optional[Dict[str,Any]]=None,):if config isNone:raiseValueError("配置是必需的")
self.config = configself.sites_df = load_sites_data(csv_file_path)self.llm = create_llm(config)self.search_tool = create_search_tool(config)self.scraping_tool = create_scraping_tool(config)self.chat_history:List[BaseMessage]=[]self.agent_executor =self._create_agent()
        logger.info(f"代理已初始化，包含 {len(self.sites_df)} 个网站")
def _create_agent(self)->AgentExecutor:"""创建带工具和提示的结构化聊天代理"""        knowledge_sources_md, domains = build_knowledge_sources_text(self.sites_df)        system_message = create_system_prompt(knowledge_sources_md, domains)
        prompt =ChatPromptTemplate.from_messages([("system", system_message),MessagesPlaceholder(variable_name="chat_history", optional=True),("human","{input}\n\n{agent_scratchpad}（提醒：无论如何都以 JSON blob 响应）""\n 重要：调用工具时保持 JSON blob 的相同格式，使用 action/action_input 字段，并在 action_input 字段中传递函数参数",),])
        agent = create_structured_chat_agent(            llm=self.llm, tools=[self.search_tool,self.scraping_tool], prompt=prompt)
returnAgentExecutor(            agent=agent,            tools=[self.search_tool,self.scraping_tool],            verbose=True,            max_iterations=10,# 限制迭代以防止无限循环            return_intermediate_steps=True,            handle_parsing_errors=True,# 优雅处理解析错误)
    async def achat(self, user_input: str)-> str:"""异步处理用户输入"""try:            logger.info(f"处理：{user_input}")
            agent_input ={"input": user_input,"chat_history":(self.chat_history[-5:]ifself.chat_history else[]),# 限制上下文窗口为 5}
# 异步调用代理执行器            response = await self.agent_executor.ainvoke(agent_input)            answer = response.get("output","无法处理您的请求。")
# 更新对话历史self.chat_history.extend([HumanMessage(content=user_input),AIMessage(content=answer)])
return answer
exceptExceptionas e:            error_msg = f"错误：{str(e)}"            logger.error(error_msg)return error_msg
def reset_memory(self):"""重置对话记忆"""self.chat_history.clear()        logger.info("记忆已重置")

"""特定领域的问答代理FastAPI应用程序
它从.env 文件读取环境变量并使用它们初始化问答代理。它有一个聊天端点，允许你与代理聊天。"""
import loggingimport osimport uuidfrom contextlib import asynccontextmanagerfrom typing importDict,Any
from fastapi importFastAPI,HTTPException,Cookie,Responsefrom pydantic importBaseModelimport uvicornfrom dotenv import load_dotenv
from qa_agent importDomainQAAgent
load_dotenv()logging.basicConfig(level=logging.INFO)logger = logging.getLogger(__name__)
def get_int_env(key: str,default:int)->int:"""从环境变量解析整数，提供回退"""try:returnint(os.getenv(key,default))exceptValueError:        logger.warning(f"无效的 {key}，使用默认值：{default}")returndefault
def get_float_env(key: str,default:float)->float:"""从环境变量解析浮点数，提供回退"""try:returnfloat(os.getenv(key,default))exceptValueError:        logger.warning(f"无效的 {key}，使用默认值：{default}")returndefault
def validate_api_keys():"""验证所需的 API 密钥是否存在"""    google_api_key = os.getenv("GOOGLE_API_KEY")    tavily_api_key = os.getenv("TAVILY_API_KEY")
ifnot google_api_key or google_api_key =="your_google_api_key_here":raiseValueError("GOOGLE_API_KEY 环境变量是必需的")
ifnot tavily_api_key or tavily_api_key =="your_tavily_api_key_here":raiseValueError("TAVILY_API_KEY 环境变量是必需的")
return google_api_key, tavily_api_key
def build_config()->Dict[str,Any]:"""从环境变量构建配置"""    google_api_key, tavily_api_key = validate_api_keys()
    search_depth = os.getenv("SEARCH_DEPTH","basic")if search_depth notin["basic","advanced"]:        logger.warning(f"无效的 SEARCH_DEPTH '{search_depth}'，使用默认值：basic")        search_depth ="basic"
return{"google_api_key": google_api_key,"tavily_api_key": tavily_api_key,"max_results": get_int_env("MAX_RESULTS",10),"search_depth": search_depth,"max_content_size": get_int_env("MAX_CONTENT_SIZE",10000),"max_scrape_length": get_int_env("MAX_SCRAPE_LENGTH",10000),"enable_search_summarization": os.getenv("ENABLE_SEARCH_SUMMARIZATION","false").lower()=="true","llm_temperature": get_float_env("LLM_TEMPERATURE",0.1),"llm_max_tokens": get_int_env("LLM_MAX_TOKENS",3000),"request_timeout": get_int_env("REQUEST_TIMEOUT",30),"llm_timeout": get_int_env("LLM_TIMEOUT",60),}
def log_config(config:Dict[str,Any]):"""美化打印配置（排除 API 密钥）"""    safe_config ={k: v for k, v in config.items()ifnot k.endswith("_api_key")}    logger.info("配置已加载：")for key, value in safe_config.items():        logger.info(f"  {key}: {value}")
def create_config()->Dict[str,Any]:"""创建并验证完整配置"""try:        config = build_config()        log_config(config)        logger.info("环境验证完成")return configexceptExceptionas e:        logger.error(f"环境验证失败：{str(e)}")raise
@asynccontextmanagerasync def lifespan(app:FastAPI):"""初始化和清理问答代理"""try:        logger.info("初始化问答代理...")        config = create_config()# 初始化空的会话存储而不是单个代理        app.state.user_sessions ={}        app.state.config = config        logger.info("会话存储初始化成功")exceptExceptionas e:        logger.error(f"无法初始化会话存储：{str(e)}")raise
yield
    logger.info("关闭会话存储...")# 清理所有代理实例if hasattr(app.state,'user_sessions'):for session_id in list(app.state.user_sessions.keys()):            logger.info(f"清理会话 {session_id}")        app.state.user_sessions.clear()
app =FastAPI(    title="特定领域问答代理 API",    description="使用 Tavily 和 LangChain 搜索特定域的问答代理",    version="1.0.0",    lifespan=lifespan,)
def get_or_create_agent(session_id: str)->DomainQAAgent:"""获取现有代理实例或为会话创建新实例"""ifnot hasattr(app.state,"user_sessions"):raiseHTTPException(status_code=500, detail="会话存储未初始化")
if session_id notin app.state.user_sessions:        logger.info(f"为会话 {session_id} 创建新代理实例")        app.state.user_sessions[session_id]=DomainQAAgent(config=app.state.config)
return app.state.user_sessions[session_id]
classChatRequest(BaseModel):    message: str    reset_memory:bool=False
classChatResponse(BaseModel):    response: str    status: str ="success"    session_id: str
@app.get("/health")async def health_check():"""带会话存储状态的健康检查"""return{"message":"特定领域问答代理 API 正在运行","status":"healthy","version":"1.0.0","active_sessions": len(app.state.user_sessions)if hasattr(app.state,"user_sessions")else0,}
@app.post("/chat", response_model=ChatResponse, summary="与问答代理聊天")async def chat(    request:ChatRequest,    response:Response,    session_id: str =Cookie(None)):"""通过问答代理处理用户问题"""# 如果不存在会话 ID，则生成新的ifnot session_id:        session_id = str(uuid.uuid4())        response.set_cookie(            key="session_id",            value=session_id,            httponly=True,            secure=True,            samesite="lax",            max_age=3600# 1 小时会话)
    logger.info(f"处理会话 {session_id} 的聊天请求")
# 获取或为此会话创建代理实例    agent = get_or_create_agent(session_id)
if request.reset_memory:        agent.reset_memory()        logger.info(f"会话 {session_id} 请求重置记忆")
    response_text = await agent.achat(request.message)    logger.info(f"成功处理会话 {session_id} 的聊天请求")
returnChatResponse(        response=response_text,        status="success",        session_id=session_id)
@app.post("/reset", summary="重置对话记忆")async def reset_memory(session_id: str =Cookie(None)):"""重置当前会话的对话记忆"""ifnot session_id:raiseHTTPException(status_code=400, detail="无活跃会话")
    agent = get_or_create_agent(session_id)    agent.reset_memory()    logger.info(f"通过端点为会话 {session_id} 重置记忆")return{"message":"对话记忆已重置","status":"success"}
if __name__ =="__main__":    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True, log_level="info")

python,python.langchain.com,官方LangChainPython文档，包括指南、教程和构建 LLM 应用的 API 参考web-development,developer.mozilla.org,涵盖 HTML、CSS、JavaScript和Web API 的全面Web开发文档

TAVILY_API_KEY=your_tavily_key_hereGEMINI_API_KEY=your_gemini_key_here

# 克隆并设置git clone https://github.com/javiramos1/qagent.gitcd qagent
make install# 配置你的 API 密钥cp .env.example .env# 添加你的 GOOGLE_API_KEY 和 TAVILY_API_KEY# 运行服务器make run

curl -X POST http://localhost:8000/chat \-H "Content-Type: application/json" \-d '{"message": "如何创建 LangChain 代理？"}'

{"status":"success","response":"要创建 LangChain 代理，你可以使用特定函数，如 `create_openai_functions_agent`、`create_react_agent` 和 `create_openai_tools_agent`。这些函数需要 LLM 和工具作为参数。推荐使用 LangGraph 构建代理，提供更多灵活性。还有一个 `create_python_agent` 函数可用。详情请参阅 LangChain 文档。"}

curl -X POST http://localhost:8000/chat \-H "Content-Type: application/json" \-d '{"message": "如何在 LlamaIndex 中使用 Ollama 的结构化输出，给我一个示例"}'

curl -X POST http://localhost:8000/chat \-H "Content-Type: application/json" \-d '{"message": "如何入侵数据库？"}'

{"status":"success","response":"我只能回答与我们可用文档来源相关的问题：AI 代理框架、AI 操作和 AI 数据框架。对于数据库安全问题，请咨询适当的安全文档或专家。"}