从零构建AI Agent:基于LangGraph的智能研究助手开发实战
1. 项目概述:从概念到实践的AI Agent之旅
最近“AI Agent”这个词火得不行,感觉身边搞技术的、做产品的、甚至一些创业者都在聊。但说实话,很多人对这个概念的理解还停留在“一个能自动执行任务的AI”这种模糊层面。今天,我就以一个过来人的身份,结合我最近从零搭建一个实际可用的AI Agent项目的经历,跟大家掰开揉碎了讲讲,到底什么是AI Agent,以及我们该如何亲手把它做出来。这不仅仅是调用几个API那么简单,它涉及到对AI能力边界的理解、任务拆解的思维、工具集成的技巧,以及如何让代码具备“思考”和“行动”的能力。
简单来说,AI Agent就是一个具备自主感知、决策、执行和反思能力的智能体。它不像传统的聊天机器人那样一问一答就结束,而是被赋予一个目标后,能够自己规划步骤、调用工具、处理信息,并持续朝着目标推进,直到任务完成或遇到无法逾越的障碍。比如,你可以让它“帮我分析一下上周的销售数据,找出异常点并生成一份报告”,它就会自己去数据库取数、用代码或工具分析、生成图表,最后整理成文档。这个过程,就是Agent在“工作”。
那么,谁适合来动手做这个呢?如果你是一名开发者,对Python有一定基础,对AI应用开发感兴趣,想从“调API”进阶到“造智能体”,那这篇内容就是为你准备的。即使你之前没接触过Agent框架,跟着步骤走,也能搭出一个有模有样的原型。当然,产品经理、技术爱好者也能从中理解Agent的核心逻辑,知道它能做什么、不能做什么,以及未来可能的发展方向。我们这次的目标很明确:不空谈理论,直接上手,用一个具体的例子,带你走完AI Agent从设计、开发到部署的完整闭环。
2. 核心设计思路:构建一个会“思考”的智能体
2.1 理解Agent的核心循环:感知、规划、行动、反思
在动手写代码之前,我们必须先建立起对AI Agent运行机制的正确认知。一个典型的Agent遵循一个核心循环,我习惯称之为“智能体工作流”。这个循环是它区别于简单脚本或规则引擎的关键。
感知(Perception) :这是起点。Agent需要接收外部的输入,这个输入可能来自用户的自然语言指令(如“帮我订一张明天北京飞上海的最便宜机票”),也可能来自系统事件、传感器数据或API回调。感知环节的核心是将这些多模态的、非结构化的输入,转化为Agent内部可以理解和处理的“状态”或“目标”。在现代Agent架构中,这通常由一个大语言模型(LLM)来完成,它负责理解用户的意图,并提取出关键的任务参数。
规划(Planning) :理解了“要做什么”之后,Agent不会立刻蛮干。一个成熟的Agent会进行任务分解和路径规划。它会把一个复杂目标拆解成一系列可执行的子任务。例如,“订机票”可以拆解为:1. 查询明天北京到上海的航班信息;2. 筛选出价格最低的选项;3. 确认用户身份和支付信息;4. 执行预订操作。规划能力的高低,直接决定了Agent处理复杂任务的智商上限。有些简单的Agent可能省略显式的规划步骤,但对于复杂任务,规划必不可少。
行动(Action) :规划好步骤,接下来就是执行。Agent通过调用各种“工具”(Tools)来与环境互动。工具可以是查询数据库的API、发送邮件的函数、操作浏览器的Selenium脚本,甚至是控制机械臂的SDK。行动环节是Agent能力的延伸,它的大脑(LLM)负责决策“何时调用何工具”,而工具则负责具体的“体力活”。设计良好的工具集是Agent实用性的基石。
反思(Reflection) :行动之后,Agent会观察结果,并进行评估。任务成功了吗?如果失败了,原因是什么?是工具调用出错,还是规划有误?基于反思,Agent可能会调整后续的规划,甚至重新理解最初的目标。这个环节让Agent具备了初步的“学习”和“纠错”能力。例如,在查询航班时如果返回“无直飞航班”,反思机制可能会触发重新规划,改为查询“中转航班”。
注意 :这个循环不是运行一次就结束的,而是一个持续迭代的过程。Agent会在这个循环中不断推进,直到任务被标记为完成或达到最大迭代次数。理解这个循环,是设计任何Agent系统的前提。
2.2 主流技术栈选型:框架、模型与工具
明确了设计思路,我们来看看具体用什么技术来实现。目前开源社区非常活跃,有几个框架已经形成了事实上的标准。
框架层(Framework) :这是Agent的“骨架”和“神经系统”。它提供了定义Agent、管理工具、运行循环的基础设施。
- LangChain / LangGraph :这可能是目前生态最丰富、社区最活跃的选择。LangChain提供了构建基于LLM应用的各类组件,而LangGraph特别擅长描述复杂的、有状态的、带循环的工作流,非常适合实现我们上面说的核心循环。它的抽象层次较高,能快速搭建原型,但深入定制时需要理解其内部状态管理。
- LlamaIndex :如果你构建的Agent核心需求是与私有数据深度交互(比如企业知识库问答、文档分析),LlamaIndex是更专注的选择。它在数据连接、索引、检索方面非常强大,可以轻松地将Agent与你的数据源结合。
- AutoGen(微软) :它采用了“多智能体对话”的范式。你可以创建多个具有不同角色(程序员、产品经理、测试员)的Agent,让它们通过对话协作完成任务。这种模式对于需要多角度推理的复杂任务(如代码评审、方案设计)特别有效。
- 简易自研 :对于功能极其简单或学习目的,你也可以不用框架,直接用OpenAI的Function Calling或类似功能,配合一个while循环和状态字典来实现最基础的Agent逻辑。这有助于理解底层原理。
模型层(LLM) :这是Agent的“大脑”。模型的选择决定了Agent的理解力、推理能力和成本。
- 云端大模型(API) : OpenAI GPT-4/GPT-4o/3.5-Turbo 是黄金标准,能力最强,开发最方便,但需要付费且数据出境。 Anthropic Claude 3 系列在长上下文和遵循指令方面表现优异。 国内大厂模型 (如DeepSeek、通义千问、文心一言、智谱GLM)的API也是可靠选择,需注意其函数调用(Function Calling)或工具使用(Tool Use)能力的支持程度和稳定性。
- 本地大模型(Local) :出于数据隐私、成本或网络考虑,你可以部署开源模型。 Qwen2.5系列 、 Llama 3系列 、 DeepSeek Coder (针对编程)等都是优秀的选择。你需要使用 Ollama 或 vLLM 等工具来部署和服务化这些模型,并在框架中配置本地API端点。本地部署对硬件(尤其是GPU显存)有要求,且推理速度可能较慢。
工具层(Tools) :这是Agent的“手和脚”。工具就是Python函数,用框架的装饰器包装一下,告诉LLM这个工具是干什么的、需要什么参数。
- 内置工具 :很多框架提供了一些常用工具,如网页搜索(DuckDuckGo)、维基百科查询、计算器、Python REPL(执行代码)等。
- 自定义工具 :这是Agent发挥价值的核心。你可以将任何能力封装成工具:查询MySQL数据库、调用企业内部CRM接口、发送Slack消息、生成图片、操作Excel文件等等。工具的设计要遵循“单一职责”原则,描述清晰准确,这样LLM才能正确调用。
我们的选择 :为了平衡学习成本、功能强大和社区支持,本教程将选择 LangChain(LangGraph) 作为主要框架,使用 OpenAI GPT-4o 或 DeepSeek最新版本 的API作为大脑(考虑到可访问性,示例代码会兼顾两者),并重点演示如何创建 自定义工具 。这是目前从入门到生产最平滑的一条路径。
2.3 定义我们的第一个Agent项目:智能研究助手
光讲理论太枯燥,我们定一个具体的目标:构建一个“智能研究助手”Agent。它的核心任务是: 根据用户提出的一个开放性问题,自动进行网络调研,收集、总结信息,并生成一份结构化的研究报告草稿。
这个项目看似简单,但涵盖了Agent设计的多个核心环节:
- 复杂任务理解 :用户的问题是开放的,比如“量子计算对现代密码学有哪些潜在威胁和机遇?”
- 多步骤规划 :Agent需要自己决定先去搜索哪些关键词、浏览哪些网页、如何交叉验证信息。
- 工具调用 :必须使用网页搜索和内容抓取工具。
- 信息整合与输出 :将收集到的碎片化信息,总结成连贯、有结构的文本。
通过实现这个Agent,你将完整走通感知、规划、行动、反思的全流程。接下来,我们就进入实战环节。
3. 开发环境搭建与核心工具创建
3.1 基础环境配置与依赖安装
工欲善其事,必先利其器。我们首先需要一个干净的Python环境。我强烈建议使用 Conda 或 venv 创建虚拟环境,避免包冲突。
# 使用 conda 创建环境(推荐)
conda create -n ai-agent python=3.10
conda activate ai-agent
# 或者使用 venv
python -m venv ai-agent-env
source ai-agent-env/bin/activate # Linux/Mac
# ai-agent-env\Scripts\activate # Windows
接下来安装核心依赖。我们将主要依赖 langchain 、 langchain-openai (或 langchain-community 对接其他模型)、 langgraph ,以及用于网页抓取的 beautifulsoup4 和 requests 。
pip install langchain langchain-openai langgraph beautifulsoup4 requests
# 如果你计划使用DeepSeek,可能需要安装 langchain-community 或专门的SDK
# pip install langchain-community
实操心得 :Python版本最好锁定在3.9或3.10,这是大多数AI库兼容性最好的版本。直接安装
langchain会包含一个比较全的基础套件,但有时某些组件更新不及时,你也可以选择安装更细分的包,如pip install langchain-core langchain-community。
3.2 构建两个关键自定义工具:搜索与抓取
我们的研究助手Agent需要与外界交互,核心工具就是 搜索 和 网页内容提取 。虽然LangChain有现成的工具(如 DuckDuckGoSearchRun ),但为了演示自定义工具的过程,并实现更可控的抓取(比如避免抓取到整站),我们自己来创建。
工具一:联网搜索工具(SearchTool) 这个工具接收一个查询字符串,调用一个公共搜索引擎(这里我们用DuckDuckGo的HTML接口模拟,实际生产环境可以考虑SerpAPI等付费服务),返回几条相关的摘要和链接。
import requests
from langchain.tools import tool
from typing import List, Dict
import urllib.parse
@tool
def search_web(query: str) -> str:
"""
使用搜索引擎查询网络信息。输入一个搜索查询词,返回相关的摘要和链接。
Args:
query: 搜索关键词,例如“量子计算 密码学 威胁 2024”
Returns:
一个格式化的字符串,包含搜索结果的标题、摘要和链接。
"""
# 注意:这是一个简化的示例。DuckDuckGo的即时答案API可能不稳定或受限。
# 生产环境请使用可靠的搜索API(如SerpAPI, Bing Search API)。
base_url = "https://api.duckduckgo.com/"
params = {
"q": query,
"format": "json",
"no_html": 1,
"skip_disambig": 1,
}
try:
response = requests.get(base_url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
results = []
# 提取抽象文本(AbstractText)
if data.get("AbstractText"):
results.append(f"摘要: {data['AbstractText']}")
# 提取相关主题(RelatedTopics)
for topic in data.get("RelatedTopics", [])[:3]: # 取前3条
if isinstance(topic, dict) and "Text" in topic and "FirstURL" in topic:
results.append(f"- {topic['Text']} [链接]({topic['FirstURL']})")
elif isinstance(topic, str):
results.append(f"- {topic}")
if results:
return "\n".join(results)
else:
return f"未找到关于 '{query}' 的直接结果。请尝试更换关键词。"
except Exception as e:
return f"搜索过程中出现错误:{str(e)}。请检查网络或稍后重试。"
工具二:智能网页抓取工具(ScrapeTool) 这个工具接收一个URL,抓取网页内容,但 不是 简单返回所有HTML文本。那样会包含大量噪音(导航栏、广告、脚本)。我们需要智能地提取主体内容。这里我们使用一个简单的启发式方法:找最大的文本连续块。更高级的方案可以用 readability 库或专门的提取服务。
from bs4 import BeautifulSoup
import re
@tool
def scrape_website(url: str) -> str:
"""
抓取给定URL的网页,并智能提取其主要文本内容。适用于新闻、博客、文档等页面。
Args:
url: 要抓取的网页完整URL。
Returns:
清理后的网页主要文本内容,去除广告、导航等噪音。
"""
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
try:
resp = requests.get(url, headers=headers, timeout=15)
resp.raise_for_status()
# 检查内容类型,非HTML不处理
if 'text/html' not in resp.headers.get('Content-Type', ''):
return "错误:该URL返回的内容不是HTML文本。"
soup = BeautifulSoup(resp.content, 'html.parser')
# 移除脚本、样式等无关标签
for script in soup(["script", "style", "nav", "footer", "aside", "header"]):
script.decompose()
# 尝试找到文章主体标签,如<article>, <main>,或包含最多<p>标签的容器
article = soup.find('article') or soup.find('main')
if article:
text = article.get_text(separator='\n', strip=True)
else:
# 备选方案:获取所有段落文本
text = soup.get_text(separator='\n', strip=True)
# 清理过多的空白字符和换行
text = re.sub(r'\n\s*\n+', '\n\n', text)
# 截取前8000字符,避免上下文过长(可根据模型上下文窗口调整)
if len(text) > 8000:
text = text[:8000] + "...【内容过长已截断】"
return text if text else "警告:从该页面未能提取出有效文本内容。"
except requests.exceptions.RequestException as e:
return f"网络请求失败:{str(e)}"
except Exception as e:
return f"抓取或解析过程出错:{str(e)}"
避坑指南 :
- 网络请求 :务必设置超时(
timeout)和用户代理(User-Agent),否则容易被目标网站拒绝或长时间挂起。- 内容清理 :网页抓取最大的挑战是噪音去除。上面的方法(找
<article>标签)对现代博客和新闻网站有效,但对老旧网站或复杂页面可能失效。生产环境应考虑使用更鲁棒的库如trafilatura或商业服务。- 长度限制 :LLM的上下文窗口是有限的(如GPT-4 Turbo是128k,但实际使用时也要考虑成本)。一定要对抓取的文本进行截断,只保留最相关的部分。更高级的做法是先用LLM对抓取内容进行摘要,再将摘要喂给Agent。
- 工具描述 :
@tool装饰器下的文档字符串(Docstring)至关重要!LLM就是靠这个描述来决定是否以及如何调用这个工具。描述要清晰说明功能、输入参数和输出格式。
现在,我们有了 search_web 和 scrape_website 这两个工具,它们将成为Agent探索网络世界的眼睛和手。
4. 使用LangGraph构建智能体工作流
有了工具,我们需要一个“大脑”来指挥它们,并用一个“工作流引擎”来管理整个循环。这就是LangGraph发挥作用的地方。
4.1 定义Agent状态与模型绑定
首先,我们要定义Agent在运行过程中需要记住哪些信息。这被称为“状态”(State)。我们用一个TypedDict来定义。
from typing import TypedDict, List, Annotated
import operator
from langgraph.graph import StateGraph, END
# 定义Agent的状态结构
class AgentState(TypedDict):
# 用户原始问题
question: str
# Agent生成的当前计划步骤列表
plan: List[str]
# 当前正在执行的步骤
current_step: str
# 从网络收集到的所有信息片段
gathered_info: List[str]
# 最终生成的报告草稿
report_draft: str
# 迭代次数,防止死循环
iteration: int
接下来,我们初始化LLM(大脑)。这里以OpenAI为例,如果你用DeepSeek,只需更换 ChatOpenAI 为对应的ChatModel类并配置正确的API Base URL即可。
from langchain_openai import ChatOpenAI
# 假设你已设置环境变量 OPENAI_API_KEY
llm = ChatOpenAI(model="gpt-4o", temperature=0) # temperature=0使输出更确定
# 如果是DeepSeek,可能类似这样(请参考对应SDK文档):
# from langchain_community.chat_models import ChatOpenAI as BaseChatOpenAI
# class DeepSeekChat(BaseChatOpenAI):
# openai_api_base = "https://api.deepseek.com/v1"
# llm = DeepSeekChat(model="deepseek-chat", api_key="your_deepseek_key")
然后,我们将LLM和工具绑定在一起,创建一个“具备工具调用能力的AI节点”。在LangChain中,这通常通过 bind_tools 方法实现。
# 将工具打包成列表
tools = [search_web, scrape_website]
# 让LLM绑定这些工具,这样它就能在需要时生成调用工具的请求了
llm_with_tools = llm.bind_tools(tools)
4.2 设计工作流节点:规划、执行、评估
LangGraph将工作流分解为多个“节点”(Node),每个节点是一个函数,负责处理状态的一部分。我们的智能研究助手至少需要三个核心节点:
节点A:规划节点(PlanNode) 这个节点负责分析用户问题,并制定一个初步的研究计划。它只在工作流开始时运行一次。
def plan_node(state: AgentState) -> AgentState:
"""分析问题,生成研究计划步骤。"""
print(f"\n=== 进入规划节点 ===")
# 构造一个提示词,让LLM生成计划
planner_prompt = f"""
你是一个资深研究助理。用户提出了以下问题:
“{state['question']}”
你的任务是进行网络调研并生成一份报告。请制定一个分步研究计划。
计划应具体,包含需要搜索的关键词和可能需要查看的网页类型。
输出格式为:
1. 第一步:...
2. 第二步:...
...
"""
# 调用LLM生成计划
plan_message = llm.invoke(planner_prompt)
plan_text = plan_message.content
# 将计划文本按行分割,过滤空行,存入状态
steps = [step.strip() for step in plan_text.split('\n') if step.strip() and (step.startswith(tuple(str(i) for i in range(1, 10))) or '步骤' in step)]
# 如果解析失败,使用一个默认计划
if not steps:
steps = [
f"1. 搜索关于 '{state['question']}' 的概述性文章。",
f"2. 查找关于此主题的最新进展或权威报告。",
f"3. 收集不同的观点或争议点。",
f"4. 总结核心发现并组织成报告。"
]
state['plan'] = steps
state['current_step'] = steps[0] if steps else "开始研究"
state['gathered_info'] = []
state['report_draft'] = ""
state['iteration'] = 0
print(f"生成的计划:{steps}")
return state
节点B:执行节点(ExecuteNode) 这是核心。它根据当前步骤( current_step ),决定是调用搜索工具还是抓取工具,或者直接进行信息处理。
def execute_node(state: AgentState) -> AgentState:
"""执行当前计划步骤,可能调用工具。"""
print(f"\n=== 进入执行节点 (迭代 {state['iteration']}) ===")
print(f"当前步骤:{state['current_step']}")
# 准备给LLM的上下文:问题、计划、已收集信息
context = f"""
用户问题:{state['question']}
整体计划:{state['plan']}
已收集信息:{state['gathered_info'][-3:]} (显示最近3条)
当前需要执行的步骤:{state['current_step']}
"""
# 我们让LLM根据当前步骤和上下文,决定下一步做什么。
# 更精细的设计是让LLM直接输出一个“行动”(Action),比如“调用搜索工具,关键词为XXX”。
# 这里为了简化,我们做一个简单的规则判断:
action_prompt = f"""
{context}
请根据“当前需要执行的步骤”判断下一步应该做什么。
选项:
A. 需要搜索新信息。如果是,请直接输出一个最相关的搜索查询词。
B. 需要详细阅读某个已搜到的链接。如果是,请直接输出该链接的URL。
C. 不需要新信息,可以基于已有信息进行总结或推进计划。
只输出A、B、C中的一个选项,如果选A,后面跟上搜索词;如果选B,后面跟上URL。
示例:
A 量子计算 密码学 安全威胁
B https://example.com/article
C
"""
action_msg = llm.invoke(action_prompt)
action = action_msg.content.strip()
print(f"LLM决策:{action}")
result = ""
if action.startswith('A'):
# 执行搜索
query = action[1:].strip()
print(f"执行搜索:{query}")
result = search_web.invoke(query)
state['gathered_info'].append(f"[搜索] 关键词:{query}\n结果:{result[:500]}...") # 只存摘要
elif action.startswith('B'):
# 执行抓取
url = action[1:].strip()
print(f"执行抓取:{url}")
result = scrape_website.invoke(url)
state['gathered_info'].append(f"[抓取] URL:{url}\n内容摘要:{result[:800]}...")
else: # 选项C或其他
# 不调用工具,可能进入总结或下一步
result = "无需调用外部工具,准备进入信息整合阶段。"
print(result)
# 将本次执行的结果也暂存,可供后续节点使用
state['last_execution_result'] = result
return state
节点C:评估与整合节点(EvaluateNode) 这个节点检查执行结果,判断当前步骤是否完成,并更新报告草稿。它决定工作流是继续下一个步骤,还是结束。
def evaluate_node(state: AgentState) -> AgentState:
"""评估执行结果,更新报告,决定下一步。"""
print(f"\n=== 进入评估节点 ===")
state['iteration'] += 1
# 防止无限循环
if state['iteration'] > 8:
print("达到最大迭代次数,强制结束。")
state['report_draft'] += "\n\n[系统提示:因迭代次数限制,研究提前终止。]"
return state
# 模拟一个简单的评估:如果已收集到一定量的信息,就开始总结
total_info_length = sum(len(info) for info in state['gathered_info'])
if total_info_length > 2000 or state['iteration'] > 4: # 阈值可调
# 信息足够,生成报告草稿
print("信息收集已较充分,开始生成报告草稿。")
summary_prompt = f"""
基于以下关于“{state['question']}”的研究信息,撰写一份简洁的研究报告草稿。
要求:结构清晰,包含引言、核心发现(分点论述)、总结。
已收集信息:
{chr(10).join(state['gathered_info'][-5:])} # 取最近5条信息
"""
report_msg = llm.invoke(summary_prompt)
state['report_draft'] = report_msg.content
print("报告草稿已生成。")
# 生成报告后,我们认为主要工作完成,可以结束
return state
else:
# 信息还不够,移动到计划中的下一个步骤
print("信息尚不充分,继续下一个研究步骤。")
current_index = state['plan'].index(state['current_step']) if state['current_step'] in state['plan'] else -1
next_index = current_index + 1
if next_index < len(state['plan']):
state['current_step'] = state['plan'][next_index]
print(f"下一步:{state['current_step']}")
else:
# 计划已执行完,但仍未达到信息量阈值,则重新评估或结束
state['current_step'] = "综合评估已有信息"
return state
4.3 组装工作流图并运行
现在,我们用LangGraph把这些节点像拼图一样组装起来,并定义它们之间的流转关系。
# 创建图构建器
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("plan", plan_node)
workflow.add_node("execute", execute_node)
workflow.add_node("evaluate", evaluate_node)
# 设置边的流转逻辑
workflow.set_entry_point("plan") # 从规划开始
workflow.add_edge("plan", "execute") # 规划完就去执行
# 关键:执行后,总是进入评估
workflow.add_edge("execute", "evaluate")
# 评估节点后,需要条件判断:是继续还是结束?
def decide_next_step(state: AgentState) -> str:
"""根据状态决定下一步是继续执行还是结束。"""
# 如果报告草稿已生成,或者迭代次数太多,则结束
if state['report_draft'] or state['iteration'] > 8:
return "end"
else:
# 否则,继续执行下一个循环
return "continue"
workflow.add_conditional_edges(
"evaluate",
decide_next_step,
{
"continue": "execute", # 回到执行节点,继续下一个步骤
"end": END # 结束工作流
}
)
# 编译图
app = workflow.compile()
大功告成!现在我们可以运行这个Agent了。
# 初始化状态
initial_state: AgentState = {
"question": "量子计算对现代密码学有哪些潜在威胁和机遇?",
"plan": [],
"current_step": "",
"gathered_info": [],
"report_draft": "",
"iteration": 0
}
# 运行工作流
print("开始运行智能研究助手...")
final_state = None
# 我们以流式方式运行,可以看到每一步的输出
for event in app.stream(initial_state, stream_mode="values"):
node_name = list(event.keys())[0]
print(f"\n[工作流事件] 节点更新: {node_name}")
# 可以在这里打印状态的某些部分,如 current_step
state = event[node_name]
if node_name == "evaluate":
print(f" 当前迭代: {state['iteration']}, 收集信息条数: {len(state['gathered_info'])}")
if state.get('report_draft'):
print(f"\n=== 报告草稿生成 ===\n{state['report_draft'][:300]}...") # 预览前300字
final_state = state
print("\n=== 工作流执行完毕 ===")
if final_state['report_draft']:
print("\n最终报告草稿:")
print("-" * 50)
print(final_state['report_draft'])
print("-" * 50)
else:
print("未能生成完整报告。")
运行这段代码,你会看到控制台打印出Agent“思考”和“行动”的过程:它先制定计划,然后搜索、抓取、评估,再搜索、再评估……直到生成一份初步的报告。这就是一个具备自主性的AI Agent在工作。
5. 高级技巧、优化与问题排查
上面的例子是一个最小可行产品(MVP)。要让Agent真正可靠、强大,还需要考虑很多问题。
5.1 提升Agent的可靠性与效率
1. 更好的规划与反思 : 我们之前的规划节点只运行一次。更高级的Agent应该具备 动态重规划 能力。即在执行过程中,如果发现新信息与预期不符,或任务无法推进,可以触发重新规划。这可以在 evaluate_node 中实现,当检测到“死胡同”时,不是简单地进入下一步,而是跳回 plan_node 或一个专门的 replan_node 。
2. 工具描述的优化 : LLM调用工具的准确性严重依赖工具的描述。描述要尽可能精确,包括参数的类型、格式、约束,以及工具可能返回的错误。例如, scrape_website 工具的描述可以加上:“输入必须是一个完整的、以http或https开头的URL。该工具可能因网站反爬机制而失败。”
3. 引入记忆机制 : 我们的 gathered_info 是一个简单的列表。对于长对话或多轮复杂任务,需要更结构化的记忆。可以考虑:
- 向量数据库记忆 :将每次交互的关键信息嵌入成向量存储,在需要时进行语义检索。这有助于Agent在长流程中记住很早之前的关键事实。
- 摘要记忆 :随着对话或任务进行,定期用LLM对之前的记忆进行摘要,压缩信息,避免上下文爆炸。
4. 并行执行与子智能体 : 有些任务步骤是独立的,可以并行执行。例如,研究一个话题时,可以同时搜索“威胁”和“机遇”。LangGraph支持创建分支和合并的流程。对于极其复杂的任务,可以创建多个专门的子Agent(如“搜索专家”、“分析专家”、“写作专家”),让它们通过消息队列协作,这就是AutoGen的思路。
5.2 常见问题与调试实录
在开发Agent过程中,你肯定会遇到各种“诡异”的情况。下面是我踩过的一些坑和解决方法:
问题1:Agent陷入死循环,不断重复相同操作。
- 现象 :Agent反复搜索同一个词,或反复抓取同一个页面。
- 原因 :状态管理有误,或者评估逻辑没有正确识别任务完成条件。也可能是LLM在相同输入下做出了相同的决策。
- 排查 :
- 打印每一步的
state,检查current_step、gathered_info是否在更新。 - 在
decide_next_step函数中加入更严格的终止条件,比如“如果最近三次收集的信息内容高度重复,则终止”。 - 在
execute_node中,为LLM提供更丰富的上下文,包括最近几次的行动历史,避免它做出重复决策。
- 打印每一步的
问题2:LLM不调用工具,或调用参数错误。
- 现象 :LLM总是回复“根据我的知识...”,而不去调用
search_web;或者调用scrape_website时给的URL格式不对。 - 原因 :
- 提示词不佳 :没有明确指令LLM必须使用工具。需要在系统提示词中强调“你 必须 使用提供的工具来获取最新信息”。
- 工具绑定问题 :确认
llm.bind_tools(tools)是否正确执行。有些模型对工具调用的支持有差异。 - 参数格式 :LLM生成工具调用参数时是文本格式,需要框架正确解析。确保工具函数的参数有明确的类型提示(如
str)。
- 解决 :
- 使用LangChain的
AgentExecutor或更高级的create_react_agent,它们内置了更强的工具调用引导逻辑。 - 在提示词中提供工具调用的具体示例(Few-shot Prompting)。
- 对于参数错误,可以在工具函数内部增加严格的输入验证和格式化逻辑。
- 使用LangChain的
问题3:网页抓取工具返回大量无关文本或失败。
- 现象 :
scrape_website返回了导航菜单、广告、JavaScript代码,或者直接超时。 - 解决 :
- 更换解析策略 :使用更专业的库如
trafilatura或readability-lxml,它们专门为提取文章主体内容设计。 - 设置请求头 :模拟真实浏览器(如Chrome)的User-Agent,有些网站会对简单爬虫屏蔽。
- 使用渲染引擎 :对于大量JavaScript动态加载的内容,简单的
requests+BeautifulSoup无效。需要考虑使用Selenium或Playwright无头浏览器,但这会大幅增加复杂性和运行时间。 - 超时与重试 :增加网络请求的超时时间,并实现简单的重试机制(如最多重试2次)。
- 降级方案 :如果抓取失败,可以尝试调用
search_web工具,看搜索结果中是否有直接摘要可用。
- 更换解析策略 :使用更专业的库如
问题4:上下文长度超限,导致后续步骤效果变差。
- 现象 :随着
gathered_info列表越来越长,在提示词中全部传入会导致超出模型的上下文窗口,LLM开始遗忘前面的内容或输出无意义内容。 - 解决 :
- 摘要 :这是最重要的策略。不要存储原始的长文本,而是每收集到几条信息,就用LLM将其总结成一段精炼的要点,只存储摘要。
- 选择性回忆 :在需要信息时,不是把所有历史都喂给LLM,而是根据当前问题,从向量数据库中检索最相关的几条历史信息。
- 分段处理 :对于超长文档,先分段,然后分别总结每段,再对总结进行总结。
5.3 部署与生产化考量
当你有一个能跑通的Agent原型后,下一步就是考虑如何让它成为一个可用的服务。
1. 封装为API服务 : 使用FastAPI或Flask,将你的Agent工作流包装成一个HTTP端点。接收用户问题,返回研究报告。注意处理异步、超时和错误。
2. 加入持久化存储 : 将每次运行的任务状态、结果存储到数据库(如SQLite、PostgreSQL),以便查询历史、恢复任务或进行分析。
3. 实现异步与队列 : Agent任务可能耗时较长(几十秒到几分钟)。不能同步阻塞HTTP请求。应该采用“任务提交 -> 立即返回任务ID -> 后台异步执行 -> 通过WebSocket或轮询查询结果”的模式。可以使用Celery + Redis/RabbitMQ来实现任务队列。
4. 监控与日志 : 详细记录Agent的每一步决策、工具调用参数和结果、LLM的输入输出。这对于调试和优化至关重要。可以集成像LangSmith这样的LLM应用监控平台。
5. 成本与性能优化 :
- 缓存 :对相同的搜索查询或网页内容进行缓存,避免重复调用昂贵的LLM或网络请求。
- 模型分级 :对于简单的分类、提取任务,使用便宜的小模型(如GPT-3.5-Turbo);对于复杂的规划、总结任务,再用大模型(如GPT-4)。
- 限制迭代次数和工具调用 :设置明确的超时和最大步骤限制,防止失控运行产生高额费用。
开发AI Agent是一个不断迭代和调优的过程。从今天这个简单的“智能研究助手”出发,你可以为其添加更多工具(数据库查询、图表生成、邮件发送),设计更复杂的规划逻辑,甚至引入多智能体协作,从而解决真实业务场景中那些繁琐、多步骤的认知型任务。记住,核心始终是理解那个“感知-规划-行动-反思”的循环,并围绕它来构建你的系统。
更多推荐
所有评论(0)