三亩地 三亩地SAN MU DI · CODE DIARY
ARTICLE DETAIL

日记详情

真实记录编程学习的某一天,欢迎挑你感兴趣的翻一翻。

从零构建AI智能体:基于LangChain的Hello-Agents实战指南

从零构建AI智能体:基于LangChain的Hello-Agents实战指南

1. 项目概述:为什么“Hello-Agents”是理解AI Native的绝佳起点

最近和不少同行交流,发现一个挺有意思的现象:大家一提到“AI智能体”或者“Agent”,要么觉得是遥不可及的前沿黑科技,要么就停留在调用大模型API做个简单问答的层面。直到我上手折腾了“Hello-Agents”这个项目,才真正把那些抽象的概念,比如“AI Native”、“智能体工作流”、“工具调用”,给落到了实实在在的代码和运行日志里。这个项目,本质上是一个精心设计的、从零开始的实战教程,它不依赖于任何庞大复杂的商业平台,而是教你用最基础的Python和主流开源框架,亲手搭建一个能感知、思考、行动并完成特定任务的智能体。

对于开发者而言,它的价值在于“祛魅”。它清晰地拆解了一个智能体从无到有的构建过程:如何设计智能体的“大脑”(即核心决策逻辑),如何为它配备“手和脚”(即工具调用能力),以及如何让它在复杂环境中持续学习与演进。通过复现这个项目,你不仅能理解智能体是如何运作的,更能掌握一套可迁移的方法论,未来无论是想做一个自动分析数据的Agent,还是一个能协调多个步骤的流程自动化助手,你都知道该从哪里入手,需要关注哪些核心模块。这远比单纯学习某个平台的使用方法要深刻得多。

2. 核心架构解析:拆解一个智能体的四大支柱

一个真正意义上的智能体,绝非一个简单的“聊天机器人”。在“Hello-Agents”的实践中,我们可以将其核心架构归纳为四个相互协作的支柱,这构成了我们理解和构建任何智能体的基础框架。

2.1 感知与理解层:从用户指令到结构化意图

这是智能体与外界交互的起点。它的任务是将用户模糊、非结构化的自然语言指令,转化为机器可以明确理解和处理的“意图”。在“Hello-Agents”中,这一步通常由大型语言模型(LLM)承担。

但这里有个关键细节:直接让LLM自由发挥是危险的,容易导致输出不稳定或偏离目标。因此,我们需要通过“提示词工程”(Prompt Engineering)来约束和引导LLM。例如,我们会设计一个系统提示词(System Prompt),明确告诉LLM:“你是一个任务规划助手。请将用户的请求解析为如下JSON格式:{“action”: “任务类型”, “target”: “操作对象”, “parameters”: {}}”。这样,用户说“帮我查一下北京明天的天气”,LLM就会输出结构化的{“action”: “query_weather”, “target”: “北京”, “parameters”: {“date”: “tomorrow”}}

实操心得:提示词的设计需要反复调试。初期可以准备一批测试用例,观察LLM的解析结果,针对常见的歧义(比如“明天”指日期还是泛指未来)在提示词中增加更明确的示例(Few-shot Learning),能显著提升意图识别的准确率。

2.2 规划与决策层:智能体的“大脑”

获得结构化意图后,智能体需要决定“怎么做”。这一层是智能体体现“智能”的核心。对于简单任务,可能是直接映射到一个工具调用。但对于复杂任务,则需要“任务分解”和“规划”。

例如,用户指令是“总结上周销售报告的核心发现并邮件发给团队”。这个任务可以分解为:1. 读取销售报告文件;2. 分析并总结核心数据;3. 撰写邮件正文;4. 获取团队成员邮箱列表;5. 调用邮件发送接口。规划层需要决定这些子任务的执行顺序和依赖关系。

在“Hello-Agents”的实现中,规划可以有两种方式:

  1. 基于规则的规划:预先定义好任务模板和流程。适合流程固定、边界清晰的场景。
  2. 基于LLM的动态规划:将当前状态(已完成步骤、可用工具)和最终目标再次提交给LLM,让LLM生成下一步行动计划。这种方式更灵活,能处理未预见的状况,但对LLM的能力和提示词设计要求更高。

2.3 工具与执行层:智能体的“手和脚”

决策之后是行动。智能体必须能调用外部工具来影响现实世界或获取信息。这是“AI Native”应用区别于传统软件的关键——它不再仅仅处理内部数据,而是成为一个连接各种API和服务的“操作中枢”。

在项目中,我们需要为智能体定义一个“工具包”(Toolkit)。每个工具都是一个函数,有明确的名称、描述、输入参数和输出格式。例如:

  • 工具名get_weather
  • 描述:根据城市名和日期查询天气信息。
  • 参数city(字符串),date(字符串,格式YYYY-MM-DD)
  • 返回:JSON格式的天气数据。

智能体的执行层负责根据决策层的指令,找到对应的工具函数,传入正确的参数并执行,最后将执行结果格式化后返回给上层。

注意事项:工具的描述至关重要。LLM主要依靠工具的描述来决定在什么情况下调用哪个工具。描述应清晰、无歧义,并包含关键参数的示例。同时,工具函数内部必须有严格的错误处理和异常捕获,避免因为某个工具失败导致整个智能体崩溃。

2.4 记忆与学习层:让智能体拥有“经验”

一个只会机械执行单次任务的不是好的智能体。记忆层使智能体能拥有“上下文”和“历史经验”,从而实现多轮对话、从错误中学习和个性化适应。

记忆通常分为几种类型:

  • 短期记忆/对话历史:保存当前会话中用户与智能体的交互记录。这是实现连贯对话的基础。
  • 长期记忆:将重要的交互结果、学到的知识或用户偏好存储到向量数据库等外部存储中,供未来检索。
  • 反思记忆:这是更高级的能力。让智能体在任务失败或完成后,回顾其思考过程(Chain of Thought)和行动轨迹,分析哪里出了问题,并将总结出的经验教训存入长期记忆。下次遇到类似情况时,它可以先检索相关经验,避免重蹈覆辙。

在“Hello-Agents”的初级阶段,我们可以先实现短期记忆。随着项目深入,引入向量数据库(如ChromaDB, Weaviate)来实现长期记忆和基于语义的检索,是让智能体能力产生质变的关键一步。

3. 从零开始:手把手构建你的第一个智能体

理论说得再多,不如动手一行代码。下面,我将以创建一个“能够查询天气和管理待办事项的桌面助手”为例,展示如何运用上述架构,从零构建一个智能体。我们将使用LangChain框架,因为它对智能体相关的抽象做得非常好,能让我们更专注于逻辑而非底层通信。

3.1 环境准备与基础框架搭建

首先,确保你的Python环境在3.8以上。我们创建一个新的项目目录并安装核心依赖。

mkdir hello-agents-demo && cd hello-agents-demo python -m venv venv # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate pip install langchain langchain-openai # 安装一个轻量级LLM,例如使用Ollama本地运行Mistral,或直接使用OpenAI API # 如果使用OpenAI: pip install openai # 如果使用Ollama本地模型: pip install ollama

接下来,初始化智能体的核心——LLM。这里以使用OpenAI API为例(你需要准备一个API Key)。

# main.py import os from langchain_openai import ChatOpenAI # 设置你的OpenAI API Key os.environ["OPENAI_API_KEY"] = "your-api-key-here" # 初始化LLM。选择gpt-3.5-turbo性价比高,适合实验。 # temperature参数控制创造性,对于任务执行类智能体,建议设低一些(如0.1)以保证稳定性。 llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.1) print("LLM初始化成功。")

3.2 定义智能体的“工具包”

我们将为智能体打造两把“利器”:天气查询和待办事项管理。

# tools.py import requests import json from datetime import datetime from typing import List, Dict # 模拟一个简单的内存存储,用于存放待办事项 todo_list = [] def get_weather(city: str, date: str = None) -> str: """ 查询指定城市未来三天的天气。 参数: city: 城市名称,例如“北京”。 date (可选): 查询日期,格式YYYY-MM-DD。默认为今天。 返回: 格式化后的天气信息字符串。 """ # 注意:这里使用了一个免费的模拟天气API。实际应用中应替换为可靠的API(如和风天气、OpenWeatherMap)。 # 并且务必处理API密钥、请求频率限制和错误。 if date is None: date = datetime.now().strftime("%Y-%m-%d") try: # 示例URL,实际不可用,仅作演示 # url = f"https://api.weather.com/v3/forecast?city={city}&date={date}" # response = requests.get(url) # data = response.json() # 模拟返回数据 mock_data = { "city": city, "date": date, "forecast": [ {"day": "今天", "condition": "晴", "high": "25", "low": "15"}, {"day": "明天", "condition": "多云", "high": "23", "low": "16"}, {"day": "后天", "condition": "小雨", "high": "20", "low": "14"} ] } result = f"{city}未来三天天气:\n" for day in mock_data["forecast"]: result += f"{day['day']}: {day['condition']}, 气温{day['low']}~{day['high']}℃\n" return result except Exception as e: return f"查询天气时出错:{str(e)}" def add_todo(item: str, priority: str = "中") -> str: """ 添加一个待办事项。 参数: item: 待办事项内容。 priority: 优先级,可选“高”、“中”、“低”。默认为“中”。 返回: 操作确认信息。 """ todo_id = len(todo_list) + 1 new_todo = { "id": todo_id, "item": item, "priority": priority, "created_at": datetime.now().isoformat(), "completed": False } todo_list.append(new_todo) return f"已添加待办事项(ID:{todo_id}): {item},优先级:{priority}。" def list_todos(show_completed: bool = False) -> str: """ 列出待办事项。 参数: show_completed: 是否显示已完成事项。默认为False。 返回: 格式化后的待办事项列表。 """ if not todo_list: return "当前没有待办事项。" filtered_todos = todo_list if show_completed else [t for t in todo_list if not t["completed"]] if not filtered_todos: return "没有找到符合条件的待办事项。" result = "待办事项列表:\n" for todo in filtered_todos: status = "✓" if todo["completed"] else "□" result += f"{status} ID:{todo['id']} [{todo['priority']}] {todo['item']}\n" return result def complete_todo(todo_id: int) -> str: """ 标记一个待办事项为已完成。 参数: todo_id: 待办事项的ID。 返回: 操作确认信息。 """ for todo in todo_list: if todo["id"] == todo_id: if todo["completed"]: return f"待办事项(ID:{todo_id})已经是完成状态。" todo["completed"] = True return f"已完成待办事项(ID:{todo_id}): {todo['item']}" return f"未找到ID为{todo_id}的待办事项。"

3.3 创建智能体并绑定工具

现在,我们将工具和LLM组合起来,形成智能体。LangChain提供了create_react_agent等高级函数,但为了理解本质,我们从相对底层的initialize_agent开始。

# agent_builder.py from langchain.agents import initialize_agent, AgentType from langchain.agents import Tool from tools import get_weather, add_todo, list_todos, complete_todo from main import llm # 导入之前初始化的llm # 1. 将函数包装成LangChain可识别的Tool对象 tools = [ Tool( name="GetWeather", func=get_weather, description="根据城市名查询未来三天的天气。输入应包含'city'参数。" ), Tool( name="AddTodo", func=add_todo, description="添加一个待办事项。需要输入'item'(事项内容)和可选的'priority'(优先级,高/中/低)。" ), Tool( name="ListTodos", func=list_todos, description="列出所有未完成的待办事项。可选参数'show_completed'(True/False)来显示已完成事项。" ), Tool( name="CompleteTodo", func=complete_todo, description="根据ID标记一个待办事项为已完成。输入应包含'todo_id'参数。" ), ] # 2. 初始化智能体 # AgentType.ZERO_SHOT_REACT_DESCRIPTION 是一种经典的智能体类型,它会让LLM根据工具描述进行推理(Reason)和行动(Act)。 agent = initialize_agent( tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True, # 开启详细日志,方便观察智能体的思考过程 handle_parsing_errors=True # 优雅地处理解析错误 ) print("智能体初始化完成!")

3.4 运行与交互测试

让我们写一个简单的循环来与智能体对话。

# run_agent.py from agent_builder import agent def run_agent_loop(): print("你好!我是你的桌面助手。我可以帮你查天气和管理待办事项。输入‘退出’来结束对话。") while True: try: user_input = input("\n你:") if user_input.lower() in ['退出', 'exit', 'quit']: print("助手:再见!") break # 将用户输入交给智能体处理 response = agent.run(user_input) print(f"助手:{response}") except Exception as e: # 处理智能体运行中可能出现的错误 print(f"助手:抱歉,处理你的请求时出现了点问题。({str(e)})") # 在实际应用中,这里可以加入更细致的错误分类和提示 if __name__ == "__main__": run_agent_loop()

现在,运行python run_agent.py,你就可以体验你的第一个智能体了!尝试输入:

  • “北京天气怎么样?”
  • “添加一个待办事项:下午三点开会,优先级高。”
  • “列出我的待办事项。”
  • “把ID为1的待办事项标记为完成。”

观察控制台verbose=True模式下打印的日志,你会看到类似以下的思考链,这正是智能体“推理-行动”过程的体现:

Thought: 用户想查询北京的天气。我需要使用GetWeather工具。 Action: GetWeather Action Input: {"city": "北京"} Observation: 北京未来三天天气:... Thought: 我已经获得了天气信息,可以回答用户了。 Final Answer: 北京未来三天天气:...

4. 进阶实战:打造更强大的智能体系统

完成了基础版本,我们可以从以下几个方向深化,打造一个更健壮、更智能的系统。

4.1 引入记忆机制实现多轮对话

上面的智能体是“健忘的”,每次对话都是独立的。我们需要引入ConversationBufferMemory来让它记住上下文。

# agent_with_memory.py from langchain.memory import ConversationBufferMemory from langchain.agents import initialize_agent, AgentType from agent_builder import tools, llm # 创建记忆体 memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True) # 创建带记忆的智能体 agent_with_memory = initialize_agent( tools, llm, agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION, # 注意更换了Agent类型 verbose=True, memory=memory, handle_parsing_errors=True ) # 测试多轮对话 print(agent_with_memory.run(“添加一个待办:买牛奶”)) print(agent_with_memory.run(“我刚刚让你添加了什么?”)) # 智能体现在能记得之前的对话了!

4.2 集成真实API与错误处理

将模拟的天气查询替换为真实API,并完善错误处理。

# real_weather_tool.py import requests from typing import Optional def get_real_weather(city: str, date: Optional[str] = None) -> str: """ 使用真实天气API(示例用和风天气)查询。 需要注册并获取API Key。 """ api_key = "YOUR_HEFENG_API_KEY" location_url = f"https://geoapi.qweather.com/v2/city/lookup?key={api_key}&location={city}" try: # 1. 获取城市Location ID loc_resp = requests.get(location_url, timeout=10) loc_resp.raise_for_status() loc_data = loc_resp.json() if loc_data['code'] != '200' or not loc_data['location']: return f"未找到城市‘{city}’,请检查名称。" location_id = loc_data['location'][0]['id'] # 2. 获取3天预报 forecast_url = f"https://devapi.qweather.com/v7/weather/3d?key={api_key}&location={location_id}" forecast_resp = requests.get(forecast_url, timeout=10) forecast_resp.raise_for_status() forecast_data = forecast_resp.json() if forecast_data['code'] != '200': return "获取天气数据失败。" result = f"{city}未来三天天气:\n" for day in forecast_data['daily']: date = day['fxDate'] cond_day = day['textDay'] temp_max = day['tempMax'] temp_min = day['tempMin'] result += f"{date}: 白天{cond_day},气温{temp_min}~{temp_max}℃\n" return result except requests.exceptions.Timeout: return "天气查询请求超时,请稍后重试。" except requests.exceptions.RequestException as e: return f"网络请求出错:{str(e)}" except (KeyError, IndexError, json.JSONDecodeError) as e: return f"解析天气数据时出错:{str(e)}"

核心技巧:在生产环境中,工具函数必须进行防御性编程。包括参数验证、网络超时设置、API响应状态码检查、异常捕获和友好的用户错误提示。一个崩溃的工具会导致整个智能体失效。

4.3 实现自主规划与复杂任务分解

对于“总结报告并发送邮件”这类复杂任务,我们需要让智能体学会自己规划。这可以通过LangChain的PlanAndExecute执行器,或者使用更高级的LLMCompilerGPT Engineer等模式来实现。其核心思想是:让一个“规划者”LLM先将大任务分解成子任务序列,再由一个“执行者”智能体(或自身)按顺序调用工具去完成。

# planner_agent.py from langchain_experimental.plan_and_execute import PlanAndExecute, load_agent_executor, load_chat_planner from agent_builder import tools, llm # 创建规划器和执行器 planner = load_chat_planner(llm) executor = load_agent_executor(llm, tools, verbose=True) # 组合成计划执行智能体 planning_agent = PlanAndExecute(planner=planner, executor=executor, verbose=True) # 测试复杂指令 complex_task = “我需要你帮我做以下几件事:1. 查询上海明天的天气。2. 添加一个待办事项‘根据天气准备衣物’。3. 列出所有待办事项提醒我。” result = planning_agent.run(complex_task) print(result)

5. 避坑指南与性能优化实战录

在开发和调试智能体的过程中,我踩过不少坑,也总结出一些提升稳定性和效率的经验。

5.1 提示词设计中的常见陷阱与优化

  1. 描述模糊导致工具误调用:早期我给GetWeather工具的描述是“查询天气”,结果用户说“我心情不好”,智能体居然去调用了天气查询工具。优化:在工具描述中明确输入格式和边界。改为“根据给定的城市名称(例如:北京、上海),查询该城市未来三天的天气预报。输入必须是一个明确的城市名。”
  2. LLM不按格式输出:期望LLM输出{“city”: “北京”},它却输出“城市是北京”。优化:在系统提示词中强化输出格式要求,并使用PydanticStructuredOutputParser(LangChain组件)来强制解析结构,解析失败时让LLM重试。
  3. 上下文过长导致性能下降:记忆体无限制增长会使每次提示词巨大,增加成本和延迟,还可能触及LLM的上下文长度限制。优化:使用ConversationSummaryMemoryConversationBufferWindowMemory(只保留最近N轮对话),定期对历史对话进行摘要,而非全文存储。

5.2 工具调用失败的处理策略

工具调用可能因网络、权限、参数错误等原因失败。智能体不能就此“死机”。

  • 重试机制:对于暂时的网络错误,可以实现简单的重试逻辑(如最多3次,每次间隔递增)。
  • 降级方案:当主要天气API失败时,可以尝试备用API或返回缓存的最近数据。
  • 明确反馈:工具函数应返回结构化的错误信息,如{“success”: false, “error”: “API服务暂时不可用”},让智能体能理解错误原因,并可能选择其他工具或如实告知用户。
  • 在Agent层面处理:使用handle_parsing_errors=True参数,并自定义agent_executor的错误处理回调函数。

5.3 成本与延迟优化

智能体频繁调用LLM和API,成本可控性很重要。

  1. 缓存:对频繁且结果变化不快的查询(如城市信息、某些静态知识)引入缓存(如langchain.cache配合SQLiteCacheRedisCache)。
  2. 小模型协同:并非所有步骤都需要最强模型。可以用小模型(如gpt-3.5-turbo)处理意图识别、简单分类,用大模型(如gpt-4)只处理核心的复杂规划和创意生成。
  3. 异步执行:如果多个子任务间没有依赖关系,可以使用异步并发来同时执行,大幅减少总等待时间。LangChain支持异步调用。
  4. 本地模型:对于隐私要求高或需要极致成本控制的场景,考虑使用Ollama部署本地模型(如Llama 3Qwen系列)。虽然能力可能稍弱,但完全私有化且无调用成本。

5.4 评估与持续改进

如何知道你的智能体变强了?需要建立评估体系。

  • 单元测试:为每个工具函数编写测试用例。
  • 集成测试:构建一个涵盖常见用户指令(正面案例)和刁钻、模糊指令(负面案例)的测试集。定期运行,监控智能体的成功率、工具调用准确率。
  • A/B测试:当优化提示词或增加新工具后,用小流量对比新旧版本的效果。
  • 用户反馈闭环:设计简单的“是否满意”反馈机制,将不满意的对话记录下来,用于分析问题所在,是工具不足、提示词不佳还是规划逻辑有缺陷。

构建AI Native智能体是一个迭代过程,从“Hello-Agents”这样的简单原型开始,逐步融入记忆、规划、复杂工具链和优化策略,你就能打造出真正理解意图、高效解决问题的数字助手。这个过程中积累的对LLM行为模式、工具编排、错误处理的理解,是任何现成平台都无法给予的核心能力。

← 返回列表