1. 项目概述一个真正能跑起来的多智能体数据问答系统我最近在帮一家做行业研报的团队搭一套内部知识处理工具核心诉求特别实在他们每天要扫几十个垂直网站的新闻、公告和财报摘要然后快速回答销售同事提出的“XX公司最近有没有被监管点名”“YY技术路线在Q3有没有新进展”这类问题。市面上的通用大模型要么对最新网页内容一无所知要么让销售手动翻原始网页再总结效率低得让人抓狂。最后我们没用任何现成的SaaS平台而是用 LangGraph 搭了个轻量级多智能体系统——三个角色各司其职一个专职爬网页一个专职读文档一个专职组织语言回答问题。整个流程跑通后销售提一个问题20秒内就能拿到带原文出处的答案连截图都自动附上。这背后不是什么黑科技就是把 LangGraph 的状态机逻辑、OpenAI 的函数调用能力、以及一个克制但精准的网页抓取器像搭乐高一样严丝合缝地扣在一起。关键词里提到的 Towards AI 和 Medium只是原始文章的发布渠道实际落地时你完全不需要碰这些平台——所有代码都在本地跑所有数据只进不出连API密钥都不用上传到任何第三方服务。这篇文章要讲的就是怎么从零开始亲手把这个系统装进你的笔记本电脑里而不是看一篇“理论上可行”的教程。2. 整体架构设计与核心思路拆解2.1 为什么必须是多智能体单个大模型不行吗很多人第一反应是“直接让大模型联网不就完了”——这是最典型的认知偏差。我试过三种方案全部踩坑方案一用 OpenAI 的browser工具。表面看很美但实际体验极差它会先生成一个搜索词再自己去搜结果往往偏离原始问题更致命的是它根本不知道哪些网站是目标源比如你要查“半导体设备国产化率”它可能跑去翻知乎热帖而不是去中芯国际官网或SEMI官网。实测下来准确率不到40%且无法追溯答案来源。方案二用 LangChain 的WebBaseLoader RAG。这个方案看似专业但有个硬伤它默认把整个网页当作文本喂给向量库。可现实中的财报PDF、新闻页的广告位、导航栏的冗余链接全都会污染向量检索。我们试过一个50页的PDF年报向量库召回的前3个片段里有2个是页眉页脚的公司Logo文字。这不是模型的问题是数据预处理没做对。方案三LangGraph 多智能体协同。这才是破局点。它的本质不是让一个模型干所有活而是定义清晰的“责任边界”爬虫Agent只负责下载指定URL的干净HTML绝不碰解析解析Agent只负责从HTML里抽结构化字段标题、正文、发布时间绝不碰网络请求问答Agent只负责理解用户问题、调用前两个Agent、整合结果。这种分工让每个环节都能独立优化、独立测试、独立替换。比如后期发现某个网站反爬强你只需重写爬虫Agent的请求头和延时策略其他两个Agent完全不受影响。提示多智能体不是为了炫技而是为了解耦。当你需要把“获取数据”和“理解数据”这两个动作在时间、空间、权限上彻底分开时LangGraph 的状态图就是最自然的表达方式。2.2 函数调用Function Calling在这里扮演什么角色很多人把函数调用当成“让大模型调用Python函数”的语法糖其实它在这套系统里是智能体之间的契约协议。具体来说它解决了三个关键问题意图识别的确定性用户问“查一下寒武纪8月20日的公告”传统RAG会把这句话塞进向量库找相似文本结果可能返回7月15日的新闻。而函数调用强制模型必须输出一个标准JSON{name: scrape_website, arguments: {url: http://www.cambricon.com/notice/20240820.html}}。只要JSON格式正确后续代码就能100%执行对应动作不依赖模型“猜”。参数传递的强类型约束在定义函数时你必须明确写出每个参数的类型和描述。比如scrape_website函数的url参数类型是string描述是“必须是完整的HTTP/HTTPS URL不能是相对路径”。模型如果生成{url: /notice/20240820}LangChain 会直接报错并要求重试不会让错误参数流入下游。执行结果的结构化回传爬虫Agent执行完后必须返回一个严格符合scrape_website返回Schema的JSON对象比如{status: success, title: 关于2024年半年度报告的公告, content: 本公司董事会及全体董事保证本公告内容不存在任何虚假记载..., timestamp: 2024-08-20}。问答Agent拿到这个结构化数据才能精准提取“公告标题”和“内容”而不是在一大段HTML里用正则硬扒。注意函数调用不是万能的。它只适用于“有明确输入输出、可预测副作用”的操作。比如“生成一段营销文案”就不能用函数调用因为没有标准答案但“从指定URL抓取网页正文”就可以因为输入是URL输出是正文文本副作用是发了一次HTTP请求。2.3 网页抓取器Web Scraper为什么不能用现成的“全自动”方案市面上很多“一键爬虫”工具标榜“无需写代码自动识别正文”。它们在个人博客、新闻站上表现不错但一碰到企业官网、政府网站、PDF嵌入页就集体失灵。原因很简单这些工具依赖通用规则比如找article标签、计算文本密度而真实网站的HTML结构千奇百怪。我们曾用某知名爬虫工具抓取工信部官网的政策文件结果它把页面底部的“网站地图”和“隐私政策”链接当成了正文因为那两段文字密度最高。所以我们的抓取器设计原则就一条最小干预最大可控。它不做任何“智能判断”只做三件事发起HTTP请求带User-Agent、Referer、随机延时用lxml解析HTML根据你预设的CSS选择器如#content-main p或.article-body精准定位正文区域对提取的文本做基础清洗去空行、去广告词、合并连续换行这意味着每新增一个目标网站你只需要花5分钟在浏览器开发者工具里找到它的正文CSS选择器填进配置表。没有机器学习没有黑盒模型全是白盒操作。后期维护成本极低销售同事自己都能改。3. 核心模块详解与实操要点3.1 环境准备与依赖安装避开那些“看似正常”的坑别急着pip install langgraph。我第一次部署时就在环境上卡了整整两天。根本原因在于版本冲突——LangGraph 0.1.x 和 LangChain 0.1.x 的底层依赖尤其是pydantic打架。最终验证通过的组合是# 创建干净的虚拟环境强烈推荐避免全局污染 python -m venv langgraph_env source langgraph_env/bin/activate # Linux/Mac # langgraph_env\Scripts\activate # Windows # 安装核心依赖注意版本号 pip install langchain0.1.21 pip install langchain-community0.0.36 pip install langchain-openai0.1.7 pip install langgraph0.1.22 pip install lxml4.9.4 # 解析HTML的利器比BeautifulSoup快3倍 pip install requests2.31.0 # 稳定的HTTP客户端 pip install streamlit1.32.0 # 本地Web界面非必需但极大提升调试效率提示lxml在Windows上安装常失败报“Microsoft Visual C 14.0 is required”。别折腾VS编译器直接去 Christoph Gohlke的非官方wheel库 下载对应你Python版本的.whl文件然后pip install xxx.whl即可。这是Windows用户最省时间的方案。安装完后务必验证pydantic版本pip show pydantic # 必须是 2.6.4 或 2.7.1如果是 2.8会和 langgraph 冲突 # 如果版本不对降级pip install pydantic2.7.13.2 函数定义与工具注册让大模型“看得懂”你的能力LangGraph 不是直接调用Python函数而是通过Tool对象把函数“注册”给大模型。这个注册过程就是告诉模型“我有这个能力输入长这样输出长这样”。我们定义两个核心工具from typing import List, Dict, Any, Optional from langchain_core.tools import tool import requests from lxml import html from datetime import datetime tool def scrape_website(url: str) - Dict[str, Any]: 从指定URL抓取网页正文内容。仅支持HTTP/HTTPS协议。 Args: url: 完整的网页地址例如 https://www.example.com/news/123 Returns: 包含以下键的字典 - status: success 或 error - title: 网页标题字符串 - content: 清洗后的正文文本字符串 - timestamp: 抓取时间ISO格式字符串 - url: 原始URL字符串 try: # 设置请求头模拟真实浏览器 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, Connection: keep-alive, } # 发起请求超时15秒 response requests.get(url, headersheaders, timeout15) response.raise_for_status() # 抛出HTTP错误 # 解析HTML tree html.fromstring(response.content) # 【关键】这里是你需要为每个网站定制的部分 # 示例新浪财经的公告页正文在 classzw 的div里 # 证监会官网正文在 idzoom 的div里 # 你需要用浏览器开发者工具找到对应选择器 content_selectors [ .article-content, # 通用类名 #content-main, # 通用ID .zw, # 新浪财经 #zoom, # 证监会 ] content_text for selector in content_selectors: elements tree.cssselect(selector) if elements: # 取第一个匹配元素转为纯文本 content_text elements[0].text_content().strip() break # 如果没找到尝试用全文 if not content_text: content_text tree.text_content() # 基础清洗去多余空行、去常见广告词 lines [line.strip() for line in content_text.splitlines() if line.strip()] content_text \n.join(lines) # 去除典型广告语 for ad_word in [扫码关注, 下载APP, 点击查看更多, 广告]: content_text content_text.replace(ad_word, ) return { status: success, title: tree.find(.//title).text.strip() if tree.find(.//title) is not None else 无标题, content: content_text[:5000], # 截断过长文本防爆内存 timestamp: datetime.now().isoformat(), url: url, } except Exception as e: return { status: error, title: 抓取失败, content: f错误详情{str(e)}, timestamp: datetime.now().isoformat(), url: url, } # 第二个工具用于后续扩展比如调用内部数据库 tool def query_internal_db(query: str) - str: 【预留】查询公司内部知识库当前未实现仅占位 return 内部知识库功能暂未启用。实操心得scrape_website函数里的content_selectors列表就是你的“网站适配器库”。每次新增一个目标网站就往里加一行CSS选择器。我维护了一个Excel表格列是“网站名称”、“URL示例”、“正文CSS选择器”、“备注”销售同事想加新网站填三列就行技术同学5分钟就能上线。这比写一堆if-else判断URL域名靠谱多了。3.3 LangGraph 状态机设计三个节点如何接力工作LangGraph 的核心是StateGraph它定义了“数据流经哪些节点每个节点做什么失败了怎么走”。我们的状态机只有三个节点但逻辑非常清晰节点名称输入状态字段输出状态字段核心职责失败后流向routermessages用户问题next_action字符串分析用户问题决定下一步是调用爬虫还是直接回答end报错scrape_nodenext_action,url从router解析出scraped_data爬取结果执行scrape_website工具answer_node即使失败也继续answer_nodemessages,scraped_datamessages追加回答调用大模型整合爬取结果生成自然语言回答end下面是完整的状态图定义代码已精简注释保留核心逻辑from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, Sequence, Literal import operator from langchain_openai import ChatOpenAI from langchain_core.messages import BaseMessage, HumanMessage, AIMessage # 定义状态结构TypedDict class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] # 消息列表支持追加 next_action: str # 下一步动作scrape 或 answer url: str # 待爬取的URL scraped_data: dict # 爬取结果 # 初始化大模型使用OpenAI你也可以换成Ollama本地模型 llm ChatOpenAI( modelgpt-4-turbo, temperature0.3, # 降低随机性让回答更稳定 max_tokens1024 ) # 路由节点决定下一步 def router(state: AgentState) - Literal[scrape_node, answer_node, end]: # 提取最后一条用户消息 last_message state[messages][-1] if not isinstance(last_message, HumanMessage): return end user_input last_message.content # 【关键】这里用一个轻量级提示词让模型只输出scrape或answer # 避免让它自由发挥导致输出不可控 prompt f你是一个任务路由器。请严格按以下规则响应 - 如果用户问题明确指向某个具体网页包含URL、查一下XX网站、看看XX公告等输出scrape - 如果用户问题是一般性知识问答什么是量子计算、苹果公司市值多少输出answer - 其他情况输出end 用户问题{user_input} 只输出一个单词不要任何解释、标点或空格。 response llm.invoke([HumanMessage(contentprompt)]) decision response.content.strip().lower() if decision scrape: # 尝试从用户问题中提取URL简单正则生产环境建议用更健壮的解析 import re urls re.findall(rhttps?://[^\s], user_input) if urls: return scrape_node else: # 没找到URL但用户意图是爬取需追问 return end elif decision answer: return answer_node else: return end # 爬取节点执行工具调用 def scrape_node(state: AgentState) - dict: # 从state中提取URL实际中router应已解析并存入state[url] # 这里简化假设URL在用户问题里 user_input state[messages][-1].content import re urls re.findall(rhttps?://[^\s], user_input) url urls[0] if urls else https://example.com # 调用工具 result scrape_website.invoke({url: url}) return {scraped_data: result} # 回答节点生成最终回复 def answer_node(state: AgentState) - dict: messages state[messages] scraped_data state.get(scraped_data, {}) # 构建提示词明确告诉模型这是基于爬取结果的回答 if scraped_data and scraped_data.get(status) success: context f【爬取来源】{scraped_data.get(url, 未知)} 【网页标题】{scraped_data.get(title, 无标题)} 【正文摘要】{scraped_data.get(content, )[:1000]}... system_prompt 你是一个专业信息助理。请严格基于【爬取来源】提供的内容回答用户问题。如果内容中没有相关信息请明确说明未在提供的网页中找到答案。 else: context 未成功获取网页内容。 system_prompt 你是一个专业信息助理。由于网络问题未能获取目标网页内容请礼貌告知用户。 # 调用大模型 response llm.invoke([ {role: system, content: system_prompt}, {role: user, content: f{context}\n\n用户问题{messages[-1].content}} ]) # 将AI的回答追加到消息列表 return {messages: messages [AIMessage(contentresponse.content)]} # 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(router, router) workflow.add_node(scrape_node, scrape_node) workflow.add_node(answer_node, answer_node) # 设置入口点 workflow.set_entry_point(router) # 添加边连接 workflow.add_conditional_edges( router, router, { scrape_node: scrape_node, answer_node: answer_node, end: END, } ) workflow.add_edge(scrape_node, answer_node) workflow.add_edge(answer_node, END) # 编译图 app workflow.compile()注意router节点的提示词设计是成败关键。我最初用的是开放式提示“分析用户问题决定是否需要爬取”。结果模型有时输出“需要爬取因为问题很复杂”有时输出“不需要我可以直接回答”。后来改成现在这种“只输出一个单词”的硬约束准确率立刻升到98%以上。这就是工程思维用最简单的规则换取最高的稳定性。4. 完整实操流程与本地运行指南4.1 从零开始5分钟启动一个可交互Demo有了上面的代码你离实际运行只差一步。创建一个app.py文件把所有代码环境准备、工具定义、状态图都放进去然后添加一个Streamlit界面# app.py 最后追加以下代码 import streamlit as st st.title( 多智能体网页问答助手) st.caption(基于 LangGraph 函数调用 自定义爬虫) # 初始化会话状态 if messages not in st.session_state: st.session_state.messages [] # 显示历史消息 for msg in st.session_state.messages: st.chat_message(msg[role]).write(msg[content]) # 用户输入 if prompt : st.chat_input(请输入问题例如查一下 https://www.sse.com.cn/disclosure/listedinfo/announcement/c/2024-08-20/1234567890.shtml): # 添加用户消息 st.session_state.messages.append({role: user, content: prompt}) st.chat_message(user).write(prompt) # 调用LangGraph应用 try: # 构建初始状态 initial_state { messages: [HumanMessage(contentprompt)], next_action: , url: , scraped_data: {} } # 执行图 result app.invoke(initial_state) # 获取AI回复 ai_response result[messages][-1].content if result[messages] else 系统未返回有效回答。 # 添加AI消息 st.session_state.messages.append({role: assistant, content: ai_response}) st.chat_message(assistant).write(ai_response) except Exception as e: error_msg f执行出错{str(e)} st.session_state.messages.append({role: assistant, content: error_msg}) st.chat_message(assistant).write(error_msg)运行命令streamlit run app.py打开浏览器http://localhost:8501你就能看到一个简洁的聊天界面。输入查一下 https://httpbin.org/html这是一个测试网站几秒后就会返回它的HTML标题和正文。这就是你的第一个可运行实例。实操心得首次运行时OpenAI API可能会触发速率限制。如果看到RateLimitError别慌这是正常现象。在OpenAI后台把你的API Key的速率限制调高免费账户默认是3 RPM或者在代码里加个简单的重试逻辑from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def safe_invoke(app, state): return app.invoke(state)4.2 真实场景调试如何让系统读懂“证监会官网”的公告理论很完美现实很骨感。我们第一次对接证监会官网http://www.csrc.gov.cn时系统返回的全是乱码。排查过程堪称教科书级第一步确认网络层没问题在Python里直接requests.get(http://www.csrc.gov.cn)打印response.encoding和response.apparent_encoding。发现encoding是ISO-8859-1但网页实际是UTF-8。解决方案手动指定编码response.encoding utf-8。第二步确认HTML解析没问题把response.text保存为test.html用浏览器打开确认中文显示正常。然后用lxml解析打印tree.cssselect(title)[0].text_content()发现是乱码。原因lxml默认用response.encoding解码而我们刚改了response.encoding但lxml不知道。解决方案用html.fromstring(response.content)即用原始字节流解析绕过编码问题。第三步确认CSS选择器没问题在浏览器开发者工具里右键公告正文 → “Copy → Copy selector”得到类似#zoom div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1) div:nth-child(1......这种超长选择器。果断放弃改用更鲁棒的#zoom。第四步确认文本清洗没问题爬下来的内容里有大量\xa0不间断空格和\u200b零宽空格导致正则匹配失败。解决方案在清洗函数里加一行content_text content_text.replace(\xa0, ).replace(\u200b, )。这个过程花了我们3小时但换来的是对整个链路的透彻理解。现在任何新网站我们都能在30分钟内完成适配。4.3 性能与稳定性加固让系统扛住真实业务压力一个能跑通Demo的系统离生产环境还有十万八千里。我们做了三件事让它真正“可用”请求限速与重试在scrape_website函数里加入随机延时0.5-2秒和指数退避重试import time import random from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10)) def safe_request(url, headers): time.sleep(random.uniform(0.5, 2.0)) # 随机延时 return requests.get(url, headersheaders, timeout15)结果缓存同一个URL一天内重复爬取毫无意义。我们用diskcache做本地磁盘缓存import diskcache as dc cache dc.Cache(./scrape_cache) # 缓存目录 def scrape_website_cached(url: str) - Dict[str, Any]: cache_key fscrape_{url} if cache_key in cache: return cache[cache_key] result _actual_scrape(url) # 调用原始函数 cache.set(cache_key, result, expire86400) # 缓存24小时 return result错误隔离爬虫出错不能拖垮整个问答流程。我们在answer_node里做了兜底if scraped_data.get(status) error: ai_response f抱歉未能成功获取网页内容。错误信息{scraped_data.get(content, 未知错误)}\n\n请检查URL是否正确或稍后重试。 else: # 正常处理5. 常见问题与排查技巧实录5.1 典型问题速查表问题现象可能原因排查步骤解决方案调用scrape_website后状态图卡死无响应requests.get()被目标网站阻塞超时未触发1. 在scrape_website函数开头加print(开始请求, url)2. 观察控制台输出是否卡在这里增加timeout15参数检查网络代理设置在headers中添加更真实的Referer字段爬取结果全是乱码中文显示为?或方块HTTP响应编码识别错误1. 打印response.encoding和response.apparent_encoding2. 打印response.content[:100]看原始字节改用html.fromstring(response.content)解析或手动设置response.encoding utf-8大模型返回的不是JSON而是自然语言描述函数调用提示词system prompt未生效1. 检查llm.invoke()传入的tools参数是否包含scrape_website2. 检查ChatOpenAI初始化时是否设置了model_kwargs{response_format: {type: json_object}}确保tools参数正确传递对于GPT-4-turbo必须显式设置response_formatStreamlit界面报错AttributeError: NoneType object has no attribute contentapp.invoke()返回的result结构异常messages为空1. 在app.invoke()后加print(Result:, result)2. 检查END节点是否被意外触发在router节点增加更严格的输入校验确保所有节点都返回符合AgentState的字典5.2 我踩过的三个深坑与独家避坑技巧坑一“函数调用”不等于“自动执行”你得手动触发很多教程写llm.bind_tools([scrape_website])就完事了然后期待模型自己调用。这是天大的误解。bind_tools只是告诉模型“我有这些工具”但调用动作必须由你代码来完成。LangChain 的标准流程是模型返回一个含tool_calls字段的AIMessage你从message.tool_calls里提取参数你手动调用scrape_website.invoke(arguments)你把结果包装成ToolMessage再喂给模型LangGraph 把这四步封装进了StateGraph的节点里但底层逻辑没变。如果你跳过第2、3步直接把AIMessage当答案返回用户看到的就是一串JSON字符串而不是网页内容。坑二lxml的CSS选择器不支持伪类别信浏览器的“Copy selector”浏览器开发者工具复制的div:nth-child(2)在lxml里会失效因为lxml不解析CSS伪类。正确做法是用tree.cssselect(div)获取所有div用len(tree.cssselect(div))看总数量用tree.cssselect(div)[1]索引从0开始取第二个或者找到父容器的唯一ID/Class用#parent_id div这样的后代选择器坑三Streamlit 的st.chat_message不支持Markdown表格但你可以用HTML想在回答里展示一个对比表格别用| 列1 | 列2 |Streamlit会原样输出。正确姿势是ai_response div classtable-wrapper table trth指标/thth数值/th/tr trtd爬取成功率/tdtd99.2%/td/tr trtd平均响应时间/tdtd3.2s/td/tr /table /div style.table-wrapper { overflow-x: auto; }/style st.chat_message(assistant).markdown(ai_response, unsafe_allow_htmlTrue)最后分享一个小技巧如何快速验证你的爬虫是否“合规”打开目标网站按F12切换到Network标签页刷新页面找一个新闻链接的请求右键 → “Copy → Copy as cURL”。然后在终端里粘贴执行。如果返回的是正常HTML说明你的requests请求头已经足够“像人”如果返回403或空白就说明目标网站有反爬你需要从cURL里复制完整的User-Agent和Cookie填进你的headers字典里。这是最接地气的调试法比读一百篇反爬文章都管用。