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

从零实现四大智能体模式：基于Groq API的Python实战指南

news 2026/4/27 22:12:21

1. 项目概述：从零构建智能体模式

如果你最近在关注AI应用开发，尤其是基于大语言模型（LLM）的智能体（Agent），那你一定听过“智能体模式”（Agentic Patterns）这个概念。这个概念由吴恩达（Andrew Ng）在他的DeepLearning.AI系列文章中系统阐述，被认为是解锁LLM更复杂、更可靠能力的关键。简单来说，智能体模式就是一套设计范式，它让LLM从一个被动的、一次性的问答机器，转变为一个能够主动思考、使用工具、制定计划并协同工作的“智能体”。

市面上已经有很多成熟的框架来实现这些模式，比如LangChain、LangGraph、LlamaIndex、CrewAI等。它们功能强大，封装完善，但有时候，过于复杂的抽象层反而会让我们看不清底层的运作逻辑。这就好比学开车，一开始就用自动挡，虽然方便，但你可能永远不知道离合器是如何配合换挡的。这个名为neural-maze/agentic-patterns-course的项目，就选择了一条“手动挡”的学习路径：不使用任何现成框架，仅通过纯Python代码和直接的Groq API调用，从零开始实现吴恩达提出的四大智能体模式。

这个项目的核心价值在于“教育”和“理解”。它剥离了所有华丽的包装，直指智能体设计的核心骨架。无论你是想深入理解智能体工作原理的开发者，还是厌倦了框架黑盒、希望拥有更多控制权的实践者，这个项目都提供了一个绝佳的“解剖实验室”。通过亲手实现反射（Reflection）、工具使用（Tool Use）、规划（Planning）和多智能体协作（Multi-Agent）这四种模式，你将能透彻地掌握智能体是如何“思考”和“行动”的，从而在未来的项目中做出更明智的技术选型和架构设计。

2. 四大智能体模式深度解析

在深入代码之前，我们必须先厘清这四种模式分别解决了什么问题，以及它们之间的演进关系。这不仅仅是功能列表，更是一套逐步增强LLM自主性和解决问题能力的“能力阶梯”。

2.1 反射模式：让模型自我审视与修正

反射模式是最基础，但效果往往最立竿见影的一种模式。它的核心思想非常简单：让LLM对自己生成的初稿进行批判性审视，并提出改进意见，然后基于意见生成更好的版本。这个过程可以迭代多次。

为什么需要反射？因为LLM在单次生成中，容易陷入局部最优，或者遗漏一些细节。比如，你让模型写一段代码，它可能实现了功能，但忽略了错误处理、代码风格或者更优的算法选择。反射模式相当于引入了一个“代码审查员”的角色。这个审查员（由另一个LLM调用扮演）会从特定视角（如“你是一位资深软件架构师”）来评估初稿，提出诸如“这里需要添加输入验证”、“那个循环可以向量化以提升性能”等具体建议。生成模型再根据这些反馈进行修订。

实操心得：反射模式的成功关键有两个。一是反射提示词（Reflection Prompt）的设计。你不能简单地说“请改进它”，而需要给反射模型一个明确的角色和审查维度。例如：“你是一位专注安全性的DevOps工程师，请检查这段部署脚本是否存在权限过大、敏感信息硬编码等安全隐患。” 二是迭代次数（n_steps）的控制。通常2-3轮迭代就能带来显著提升，过多轮次可能导致改进边际效应递减，甚至引入新的无关问题，徒增成本。

2.2 工具模式：为模型装上感知和行动的“四肢”

LLM的本质是一个基于训练数据的概率模型，它的知识存在截止日期，也无法直接与外部世界交互。工具模式就是为了突破这个限制。其核心是让LLM学会在需要时调用预定义好的外部函数（工具），获取实时、准确的信息或执行具体操作。

一个典型的工具调用流程是：用户提问 -> LLM分析问题 -> LLM判断是否需要调用工具以及调用哪个工具 -> 以结构化格式（如JSON）输出工具调用请求 -> 系统执行工具函数 -> 将工具执行结果返回给LLM -> LLM整合信息生成最终回答。例如，用户问“今天纽约的天气如何？”，LLM会识别出需要调用“获取天气”的工具，然后生成类似{"action": "get_weather", "action_input": {"location": "New York"}}的请求。

注意事项：工具描述（Docstring）至关重要。LLM完全依赖你为工具函数编写的文档字符串来决定是否以及如何调用它。描述必须清晰、准确，包含函数功能、参数类型和含义、返回值格式。模糊的描述会导致模型错误调用或拒绝调用。此外，工具的设计应保持单一职责，一个工具只做一件事，这样模型的决策会更准确。

2.3 规划模式：让模型学会“先想后做”

当任务变得复杂，无法通过一次工具调用解决时，我们就需要规划模式。最具代表性的就是ReAct（Reasoning + Acting）范式。它让LLM进行链式思考：先推理（Reason）下一步该做什么，然后行动（Act）——通常是调用一个工具，观察（Observe）结果，再基于结果进行下一轮推理，如此循环，直至任务完成。

与简单的工具模式相比，ReAct模式中的LLM拥有了“工作记忆”。它会把之前的思考步骤、工具调用和观察结果都保留在上下文里，从而能够处理多步骤的复杂任务。例如，面对问题“爱因斯坦哪年获得诺贝尔奖？他获奖时所在机构的校长是谁？”，一个ReAct智能体可能会：1. 推理：需要先查找爱因斯坦的获奖年份。2. 行动：调用维基百科工具查询。3. 观察：得到“1921年”等信息。4. 推理：现在需要查找1921年爱因斯坦所在机构（例如普林斯顿高等研究院）及其校长信息。5. 行动：再次调用工具查询。6. 观察并整合最终答案。

核心挑战：规划模式最大的挑战是错误累积和循环失控。一旦某一步推理或工具调用出错，后续步骤可能会全盘皆错。因此，实现时需要设置最大步数限制，并考虑加入对中间结果的简单验证逻辑。同时，ReAct对模型的逻辑推理能力要求较高，选择像Groq提供的Llama 3 70B这类更强大的模型，效果会好于小参数模型。

2.4 多智能体模式：分工协作的“专家团队”

这是最复杂、也最接近人类协作的模式。其核心思想是创建多个具备不同角色、专长和目标的智能体，让它们通过通信和协作来完成一个共同的大任务。这模拟了一个项目团队，有项目经理、开发、测试、文案等不同角色。

在这个项目的实现中，借鉴了CrewAI的“Crew”（团队）和“Agent”（成员）抽象，并用类似Apache Airflow的>>运算符来定义执行依赖关系。例如，你可以创建一个“研究员”智能体负责搜集资料，一个“写手”智能体负责撰写报告，一个“审阅者”智能体负责润色。通过研究员 >> 写手 >> 审阅者这样的依赖定义，就构成了一个工作流管道（Pipeline）。

设计要点：多智能体系统的设计关键在于角色定义清晰和通信协议明确。每个智能体的“背景故事”（backstory）和“任务描述”（task_description）就是它的角色说明书，必须细致到足以让它理解自己的职责边界。同时，智能体之间如何传递信息（比如通过共享的上下文字典）也需要在架构层面设计好，避免信息丢失或冲突。

3. 环境准备与项目初始化

理解了理论，我们开始动手。这个项目基于Python和Groq云服务，因此准备工作主要围绕这两方面展开。

3.1 依赖安装与虚拟环境管理

项目代码库提供了两种安装方式：使用Poetry或直接pip安装PyPI包。对于学习目的，我强烈推荐克隆源码并使用Poetry安装，这样你能随时查看和修改底层实现。

首先，确保你的系统已安装Python（建议3.9以上版本）和Git。然后，打开终端执行以下命令：

# 1. 克隆项目仓库到本地 git clone https://github.com/neural-maze/agentic-patterns-course.git cd agentic-patterns-course # 2. 使用Poetry安装依赖（如果你已安装Poetry） poetry install # 如果你没有Poetry，可以先安装它，或者使用pip安装 # 安装Poetry： curl -sSL https://install.python-poetry.org | python3 - # 然后再次运行 `poetry install` # 3. 激活Poetry创建的虚拟环境 poetry shell

使用Poetry的好处是它能完美锁定所有依赖的版本，避免因为库版本冲突导致代码运行异常。如果你习惯使用pip和venv，也可以手动创建虚拟环境并安装：

python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate pip install -U agentic-patterns

避坑指南：如果你在安装过程中遇到与pydantic版本相关的错误，这很可能是由于项目中某些依赖对pydantic的版本有特定要求。一个稳妥的解决方法是，在Poetry安装前，先尝试在pyproject.toml中明确指定pydantic的版本范围（例如pydantic = "^1.10.0"），或者查看项目仓库的issue区是否有已知的解决方案。虚拟环境是Python项目的标配，务必使用，它能将项目依赖与系统Python环境隔离。

3.2 获取并配置Groq API密钥

本项目使用Groq作为LLM提供商。Groq以其极快的推理速度（特别是使用LPU硬件）而闻名，非常适合需要多次LLM调用的智能体应用。你需要：

访问 Groq官网并注册账号。
登录后，在控制台找到“API Keys” section，点击“Create API Key”来生成一个新的密钥。
复制生成的密钥字符串（形如gsk_xxxxxx）。

接下来，在项目的根目录下，创建一个名为.env的文件。你可以直接复制项目自带的模板：

# 在项目根目录下 cp .env.example .env

然后用文本编辑器打开.env文件，将你的Groq API密钥填入：

GROQ_API_KEY=你的_GROQ_API_密钥_粘贴在这里

安全提示：.env文件包含了你的敏感密钥，务必将它添加到.gitignore文件中，确保不会意外提交到公开的Git仓库。许多项目初始就已配置好。永远不要在代码中硬编码API密钥。

3.3 验证环境与快速测试

环境配置好后，我们可以写一个最简单的脚本来测试一切是否正常。创建一个test_groq.py文件：

import os from groq import Groq from dotenv import load_dotenv # 加载 .env 文件中的环境变量 load_dotenv() # 初始化Groq客户端 client = Groq(api_key=os.environ.get("GROQ_API_KEY")) # 发起一次简单的聊天补全请求 completion = client.chat.completions.create( model="llama3-8b-8192", # 可以换成 llama3-70b-8192 等Groq支持的模型 messages=[ {"role": "system", "content": "你是一个乐于助人的助手。"}, {"role": "user", "content": "用一句话介绍你自己。"} ], temperature=0.5, max_tokens=100, ) print(completion.choices[0].message.content)

运行这个脚本python test_groq.py。如果看到Groq模型的回复输出，恭喜你，环境配置成功！如果遇到认证错误，请检查.env文件路径是否正确、密钥是否有效，以及网络连接是否正常。

4. 核心模式实现与代码实战

现在，我们进入最核心的部分——逐一拆解并运行四大模式的实现代码。我将结合项目源码，解释关键类的设计思路，并给出扩展实践的思路。

4.1 反射模式实战：以代码生成为例

反射模式的实现集中在ReflectionAgent类。其工作流程是一个清晰的循环：生成 -> 反射 -> 修订。让我们看一个改进代码生成的例子。

from agentic_patterns import ReflectionAgent import time # 初始化反射智能体 agent = ReflectionAgent() # 定义角色和任务 generation_system_prompt = """你是一位资深的Python开发工程师，擅长编写高效、整洁、符合PEP 8规范的代码。你的代码必须包含适当的错误处理和详细的文档字符串。""" reflection_system_prompt = """你是Python开源项目（如Requests、Flask）的核心维护者，以对代码质量、可读性和性能的严苛要求而闻名。请严格审查以下代码，指出其在**算法效率**、**错误处理鲁棒性**、**代码风格**和**潜在边界条件**方面的具体问题，并提供非常具体的修改建议。""" user_msg = """请编写一个Python函数，用于从一个可能包含嵌套列表和字典的复杂数据结构（仅包含字符串、数字、列表、字典）中，提取出所有的整数值，并返回它们的和。例如，输入 {'a': 1, 'b': ['x', 2, {'c': 3}], 'd': 4} 应返回 10。""" print("开始反射优化过程...") start_time = time.time() final_response = agent.run( user_msg=user_msg, generation_system_prompt=generation_system_prompt, reflection_system_prompt=reflection_system_prompt, n_steps=3, # 进行3轮生成-反射循环 verbose=1, # 打印中间步骤，方便观察 ) end_time = time.time() print(f"\n最终优化后的代码：") print("-" * 50) print(final_response) print("-" * 50) print(f"总耗时：{end_time - start_time:.2f}秒")

运行这段代码，你会看到控制台输出每一步的“初稿”和“反射意见”。ReflectionAgent内部的关键在于它维护了两套提示词和可能两个独立的LLM调用（虽然这里共用一个Groq客户端）。在每一轮：

生成器（Generator）根据generation_system_prompt和当前上下文（上一轮的修订版或初始问题）生成内容。
反射器（Reflector）根据reflection_system_prompt和生成器产出的内容，生成批评和改进建议。
生成器将反射建议和之前的内容结合起来，生成下一轮修订版。

深度解析：ReflectionAgent.run()方法内部，n_steps参数控制循环次数。verbose参数在设置为1或2时，会打印出中间状态，这对于调试和理解反射过程非常有帮助。你可以尝试调整反射提示词，比如让反射器专注于安全性、可测试性或特定领域的规范，观察最终输出如何变化。

4.2 工具模式实战：构建一个信息查询助手

工具模式的核心是ToolAgent和@tool装饰器。我们来实现一个稍微复杂点的工具，它不仅能获取Hacker News头条，还能根据关键词搜索。

首先，我们定义两个工具：

import json import requests from agentic_patterns.tool_pattern.tool import tool from agentic_patterns.tool_pattern.tool_agent import ToolAgent from typing import List, Dict @tool def fetch_top_hacker_news_stories(top_n: int = 10) -> str: """ 从Hacker News API获取当前排名前N的故事。 此函数会查询Hacker News的topstories接口，然后获取每个故事ID的详细信息， 包括标题、URL、得分、作者和提交时间。 参数: top_n (int): 需要获取的故事数量，默认为10。 返回: str: 一个JSON格式的字符串，包含故事列表。每个故事是一个字典，包含 'title', 'url', 'score', 'by', 'time' 等字段。 """ try: # 获取顶级故事ID列表 top_stories_url = 'https://hacker-news.firebaseio.com/v0/topstories.json' id_response = requests.get(top_stories_url, timeout=10) id_response.raise_for_status() top_story_ids = id_response.json()[:top_n] stories = [] for story_id in top_story_ids: story_url = f'https://hacker-news.firebaseio.com/v0/item/{story_id}.json' story_response = requests.get(story_url, timeout=10) story_response.raise_for_status() story_data = story_response.json() stories.append({ 'title': story_data.get('title', 'No Title'), 'url': story_data.get('url', f'https://news.ycombinator.com/item?id={story_id}'), 'score': story_data.get('score', 0), 'author': story_data.get('by', 'Unknown'), 'time': story_data.get('time', 0) }) return json.dumps(stories, ensure_ascii=False, indent=2) except requests.exceptions.RequestException as e: return json.dumps({"error": f"网络请求失败: {str(e)}"}) except json.JSONDecodeError as e: return json.dumps({"error": f"解析JSON响应失败: {str(e)}"}) @tool def search_hn_by_keyword(keyword: str, max_results: int = 5) -> str: """ 通过Algolia的Hacker News搜索API，根据关键词搜索故事。 参数: keyword (str): 搜索关键词。 max_results (int): 返回的最大结果数，默认为5。 返回: str: 一个JSON格式的字符串，包含搜索到的故事列表。每个故事包含 'title', 'url', 'points', 'author' 等字段。 """ try: search_url = "http://hn.algolia.com/api/v1/search" params = { "query": keyword, "tags": "story", "hitsPerPage": max_results } response = requests.get(search_url, params=params, timeout=10) response.raise_for_status() search_data = response.json() results = [] for hit in search_data.get('hits', [])[:max_results]: results.append({ 'title': hit.get('title', 'No Title'), 'url': hit.get('url') or f"https://news.ycombinator.com/item?id={hit.get('objectID')}", 'points': hit.get('points', 0), 'author': hit.get('author', 'Unknown'), 'created_at': hit.get('created_at', '') }) return json.dumps(results, ensure_ascii=False, indent=2) except requests.exceptions.RequestException as e: return json.dumps({"error": f"搜索请求失败: {str(e)}"})

接下来，我们初始化ToolAgent并让它使用这些工具：

# 创建工具智能体，并传入我们定义的工具列表 hn_agent = ToolAgent(tools=[fetch_top_hacker_news_stories, search_hn_by_keyword]) # 场景1：直接询问头条新闻 query_1 = "现在Hacker News上最热门的3条新闻是什么？" print(f"用户提问: {query_1}") output_1 = hn_agent.run(user_msg=query_1) print(f"智能体回答:\n{output_1}\n{'-'*60}") # 场景2：询问一个需要模型判断用哪个工具的问题 query_2 = "我想了解最近关于‘人工智能伦理’方面的讨论，能找到相关的热门文章吗？" print(f"用户提问: {query_2}") output_2 = hn_agent.run(user_msg=query_2) print(f"智能体回答:\n{output_2}")

当你运行这段代码时，ToolAgent内部会发生以下事情：

它将所有工具的文档字符串（就是函数定义下面"""括起来的部分）格式化后，作为系统提示词的一部分提供给LLM。
LLM分析用户问题，判断是否需要调用工具。如果需要，它会根据工具描述，生成一个结构化的调用请求。
ToolAgent解析这个请求，找到对应的Python函数并执行，传入指定的参数。
将工具执行返回的结果（JSON字符串）再次喂给LLM。
LLM整合工具返回的信息和原始问题，生成一段自然语言的回答。

工具设计精髓：注意看工具函数的返回类型是str，并且我们返回的是json.dumps()序列化后的字符串。这是因为LLM处理结构化的文本（如JSON）比处理复杂的Python对象更容易。工具函数的文档字符串是LLM理解工具能力的唯一来源，务必写得像给一个新手程序员看一样清晰、无歧义。

4.3 规划模式实战：实现一个数学问题求解器

规划模式以ReactAgent为代表，它是ToolAgent的升级版，具备了在多个工具间进行链式推理和规划的能力。我们构建一个能解决多步数学问题的智能体。

首先，定义一组数学工具：

import math from agentic_patterns.tool_pattern.tool import tool @tool def add(a: float, b: float) -> float: """计算两个数的和。参数: a (float): 第一个数。 b (float): 第二个数。返回: float: a 与 b 的和。""" return a + b @tool def subtract(a: float, b: float) -> float: """计算两个数的差 (a - b)。参数: a (float): 被减数。 b (float): 减数。返回: float: a 减去 b 的差。""" return a - b @tool def multiply(a: float, b: float) -> float: """计算两个数的乘积。参数: a (float): 第一个因数。 b (float): 第二个因数。返回: float: a 与 b 的乘积。""" return a * b @tool def divide(a: float, b: float) -> float: """计算两个数的商 (a / b)。参数: a (float): 被除数。 b (float): 除数。返回: float: a 除以 b 的商。如果除数为零，返回 'Error: Division by zero'。""" if b == 0: return "Error: Division by zero" return a / b @tool def power(base: float, exponent: float) -> float: """计算幂运算 (base ^ exponent)。参数: base (float): 底数。 exponent (float): 指数。返回: float: base 的 exponent 次幂。""" return math.pow(base, exponent) @tool def sqrt(x: float) -> float: """计算一个非负数的平方根。参数: x (float): 需要计算平方根的数，必须 >= 0。返回: float: x 的平方根。如果 x < 0，返回 'Error: Cannot calculate square root of a negative number'。""" if x < 0: return "Error: Cannot calculate square root of a negative number" return math.sqrt(x)

然后，创建ReactAgent并让它解决一个复杂问题：

from agentic_patterns.planning_pattern.react_agent import ReactAgent # 初始化ReAct智能体，传入所有数学工具 math_agent = ReactAgent(tools=[add, subtract, multiply, divide, power, sqrt]) # 定义一个多步骤的数学问题 complex_problem = """ 我有一个数学问题需要解决： 1. 首先，计算15和25的和。 2. 然后，将上一步得到的结果乘以4。 3. 接着，计算100减去上一步结果的差。 4. 最后，对上一步得到的差求平方根，并告诉我最终答案。 请一步步思考并给出最终数值。 """ print("问题：", complex_problem) print("开始ReAct推理过程...") final_answer = math_agent.run(user_msg=complex_problem, verbose=1) # 设置verbose=1查看思考链 print("\n最终答案：", final_answer)

将verbose参数设为1或2，你会在控制台看到完整的“思考-行动-观察”链。例如：

思考：用户想要一个多步计算。第一步是计算15和25的和。我应该使用加法工具。 行动：调用工具 `add`，参数: a=15, b=25 观察：40 思考：第一步结果是40。第二步是将40乘以4。我应该使用乘法工具。 行动：调用工具 `multiply`，参数: a=40, b=4 观察：160 思考：第二步结果是160。第三步是计算100减去160。我应该使用减法工具。 行动：调用工具 `subtract`，参数: a=100, b=160 观察：-60 思考：第三步结果是-60。最后一步是对-60求平方根。但平方根工具要求输入非负数。这里有问题，需要对负数求平方根，这在实数范围内无解。我需要检查工具描述或告知用户。 行动：调用工具 `sqrt`，参数: x=-60 观察：Error: Cannot calculate square root of a negative number 思考：工具返回了错误信息。最终答案是：在实数范围内，无法对负数求平方根。所以最终结果是一个错误信息。 最终答案：在计算过程中，第三步得到的结果是-60。由于无法对负数求实数平方根，因此您问题的最后一步无法完成。实数平方根仅对非负数有定义。

ReAct的内部机制：ReactAgent在内部维护了一个“思考链”字符串，其中记录了所有的Thought、Action和Observation。每一轮，它都将整个思考链和工具结果作为上下文，送给LLM去生成下一步的Thought。这种设计让模型具备了“工作记忆”，能够进行连贯的多步推理。但这也带来了上下文长度增长的挑战，对于非常长的任务链，可能需要策略性地总结或截断历史。

4.4 多智能体模式实战：构建一个内容创作流水线

多智能体模式将任务分解，由多个各司其职的智能体协作完成。这个项目的实现提供了一个轻量级的Crew和Agent抽象。我们模拟一个简单的“技术博客创作流水线”。

from agentic_patterns.multiagent_pattern.crew import Crew from agentic_patterns.multiagent_pattern.agent import Agent from agentic_patterns.tool_pattern.tool import tool import json # 首先，定义一个工具，用于将内容保存为Markdown文件 @tool def save_to_markdown(content: str, filename: str = "output.md") -> str: """ 将给定的文本内容保存到指定的Markdown文件中。 参数: content (str): 需要保存的文本内容。 filename (str): 目标Markdown文件名，默认为'output.md'。 返回: str: 确认信息，例如“文件已成功保存至 {filename}”。 """ try: with open(filename, 'w', encoding='utf-8') as f: f.write(content) return f"内容已成功保存至文件：{filename}" except IOError as e: return f"保存文件时出错：{str(e)}" # 使用上下文管理器创建并运行一个Crew with Crew() as blog_crew: # 智能体1：主题研究员 - 负责生成博客主题和大纲 researcher = Agent( name="技术趋势研究员", backstory="你是一位专注于AI和软件开发领域的技术分析师，擅长从海量信息中提炼出有洞察力的趋势和主题。", task_description="基于当前AI Agent技术的发展，提出3个有吸引力的技术博客主题，并为每个主题提供一个简要的段落描述和核心要点大纲。", task_expected_output="一个清晰的JSON列表，每个元素包含'title'（主题）、'description'（描述）和'outline'（大纲要点列表）三个字段。" ) # 智能体2：内容写手 - 负责根据大纲撰写完整文章 # 注意：它的任务描述中引用了前一个agent的output变量，这是定义依赖的关键 writer = Agent( name="资深技术写手", backstory="你是一位经验丰富的技术博客作者，文笔流畅，善于将复杂的技术概念转化为通俗易懂、引人入胜的文章。", task_description="从研究员提供的主题列表中，选择第一个主题（{researcher.output}），并依据其大纲，撰写一篇完整的、约500字的技术博客文章。文章需包含引言、主体和结论。", task_expected_output="一篇格式良好、可直接发布的Markdown格式博客文章正文。" ) # 智能体3：编辑校对 - 负责润色和保存 editor = Agent( name="严格的技术编辑", backstory="你是一位一丝不苟的编辑，对技术准确性、语法、拼写和文章流畅度有极高的要求。", task_description="对写手生成的博客文章进行校对、润色和优化。检查技术术语准确性，修正语法错误，提升段落衔接，确保文章专业且可读性强。", task_expected_output="优化后的最终版Markdown文章内容。", tools=[save_to_markdown] # 编辑拥有保存文件的工具 ) # 定义执行依赖关系：研究员 -> 写手 -> 编辑 researcher >> writer >> editor # 运行整个团队 print("启动博客创作流水线...") final_output = blog_crew.run() print("\n流水线执行完成！") # 最终输出是最后一个智能体（editor）的产出，即优化后的文章内容或保存确认信息。 print("最终输出:", final_output[:500] + "..." if len(final_output) > 500 else final_output) # 打印前500字符 # 可视化工作流（需要graphviz库） try: blog_crew.plot() except Exception as e: print(f"无法生成流程图，可能缺少graphviz: {e}")

在这个例子中：

researcher >> writer表示writer智能体的任务依赖于researcher智能体的输出。在writer的task_description中，我们通过{researcher.output}这个占位符来引用前驱智能体的结果。Crew在执行时会自动进行变量替换。
Crew类负责管理智能体之间的依赖关系，并按照拓扑顺序（本例中是简单的线性顺序）依次执行。
每个Agent本质上是一个包装好的LLM调用，拥有自己的系统提示词（由backstory和role构成）和用户提示词（task_description）。

多智能体协作的关键：依赖管理和上下文传递是核心。这个实现采用了简单的字符串格式化（{agent_name.output}）来传递信息。在更复杂的场景中，你可能需要设计一个共享的“工作区”或“黑板”系统，让智能体可以读写共享状态。此外，为不同的智能体分配不同的LLM模型（比如研究员用更擅长分析的模型，写手用更擅长创作的模型）可以进一步提升整体效果。

5. 项目源码结构与扩展实践

要真正吃透这些模式，阅读甚至修改项目源码是最好的方式。让我们浏览一下核心的目录结构。

agentic-patterns-course/ ├── src/ │ └── agentic_patterns/ │ ├── __init__.py │ ├── reflection_pattern/ │ │ ├── __init__.py │ │ └── reflection_agent.py # ReflectionAgent 实现 │ ├── tool_pattern/ │ │ ├── __init__.py │ │ ├── tool.py # @tool 装饰器和 Tool 基类 │ │ └── tool_agent.py # ToolAgent 实现 │ ├── planning_pattern/ │ │ ├── __init__.py │ │ └── react_agent.py # ReactAgent 实现 │ └── multiagent_pattern/ │ ├── __init__.py │ ├── agent.py # Agent 类 │ └── crew.py # Crew 类 ├── notebooks/ # Jupyter Notebook 教程 │ ├── reflection_pattern.ipynb │ ├── tool_pattern.ipynb │ ├── planning_pattern.ipynb │ └── multiagent_pattern.ipynb ├── pyproject.toml # 项目依赖和配置 ├── .env.example └── README.md

5.1 如何自定义和扩展工具

工具是智能体能力的基石。项目中的@tool装饰器是一个优雅的设计，它自动将普通Python函数转化为智能体可识别的工具。查看src/agentic_patterns/tool_pattern/tool.py，你会发现其核心是创建一个Tool类对象，其中包含了函数名、描述、参数模式等信息，这些信息最终会被格式化成LLM能理解的提示词。

你可以轻松创建任何功能的工具。例如，创建一个获取天气的工具：

import requests from agentic_patterns.tool_pattern.tool import tool @tool def get_current_weather(city: str, country_code: str = "CN") -> str: """ 获取指定城市的当前天气情况。 此函数调用一个开放的天气API（例如Open-Meteo）来获取实时数据。 参数: city (str): 城市名称，例如“Beijing”。 country_code (str): 国家代码，默认为“CN”。遵循ISO 3166-1 alpha-2标准。 返回: str: 包含温度、天气状况、湿度等信息的格式化字符串。如果请求失败，返回错误信息。 """ # 注意：这里使用一个无需API密钥的示例端点，实际使用时请替换为可靠的API try: # 示例URL，实际可能需要更复杂的参数和认证 url = f"https://api.open-meteo.com/v1/forecast?latitude=39.9&longitude=116.4&current_weather=true" # 实际应用中，你需要根据city和country_code查询经纬度 response = requests.get(url, timeout=10) data = response.json() current = data.get('current_weather', {}) return (f"{city}的当前天气：温度 {current.get('temperature', 'N/A')}°C, " f"风速 {current.get('windspeed', 'N/A')} km/h, " f"天气代码 {current.get('weathercode', 'N/A')}。") except Exception as e: return f"获取{city}天气失败：{str(e)}"

将这个工具添加到任何一个ToolAgent或ReactAgent中，智能体就具备了查询天气的能力。关键在于工具描述的清晰度和错误处理的健壮性。

5.2 构建一个综合智能体应用

掌握了单个模式后，我们可以尝试组合它们，构建更强大的应用。例如，一个“技术调研助手”可以这样设计：

使用多智能体模式，创建“搜索专家”、“总结员”和“报告生成器”三个智能体。
在“搜索专家”内部，使用工具模式，调用搜索引擎API和学术数据库API。
在“报告生成器”内部，使用反射模式，对生成的报告草稿进行多次润色和优化。
整个流程由“规划模式”驱动，一个主控ReAct智能体来协调何时启动哪个多智能体小组，或者处理异常分支（比如搜索无结果时怎么办）。

这种“模式嵌套”是构建复杂智能体系统的常见思路。这个项目提供的底层实现，给了你最大的灵活性去尝试这种组合。

6. 常见问题、调试技巧与性能优化

在实际操作中，你肯定会遇到各种问题。这里我总结了一些常见坑点和解决思路。

6.1 工具调用失败或不准

症状：智能体应该调用工具却没调用，或者调用了错误的工具，或参数传递错误。
排查：
1. 检查工具描述：这是最常见的原因。确保你的工具函数文档字符串清晰、完整地描述了功能、每个参数的类型和意义、返回值格式。用简单的英语，避免歧义。
2. 启用详细日志：在初始化Agent时，设置verbose=2。这会打印出LLM在决定调用工具前的完整思考过程，以及它准备发送给工具的参数字典。对比这个字典和你函数定义的参数，就能发现问题。
3. 简化任务：先用一个极其简单的任务测试工具调用（如“计算1+1”），确保基础通路正常，再逐步增加复杂度。
4. 调整提示词：系统提示词（在Agent内部）可能影响了工具选择。你可以尝试在系统提示中更加强调“当你需要获取实时信息或进行计算时，必须使用提供的工具”。

6.2 ReAct智能体陷入循环或逻辑混乱

症状：智能体在几步之后开始重复动作，或者推理偏离正轨，无法达到终止条件。
排查与解决：
1. 设置最大步数：ReactAgent的run方法通常有max_iterations参数（在这个项目里可能是max_steps或类似参数），务必设置一个合理的上限（如10-20步），防止无限循环消耗API费用。
2. 强化终止条件提示：在给ReAct智能体的系统提示中，明确写出终止条件。例如：“当你认为已经得到了问题的最终答案，或者无法继续时，请输出Final Answer:开头的句子来结束任务。”
3. 检查工具观察结果：如果某一步工具返回了错误信息（如“未找到结果”），LLM可能会困惑。确保你的工具在失败时返回清晰、结构化的错误信息，帮助LLM理解现状并调整策略。
4. 使用更强的模型：复杂的多步推理对模型能力要求高。如果使用llama3-8b效果不佳，可以尝试切换到Groq提供的llama3-70b或mixtral-8x7b等更大、推理能力更强的模型。在初始化Agent时，可以通过model_name参数指定（如果该Agent类支持）。

6.3 多智能体协作上下文丢失

症状：下游智能体收不到上游智能体的输出，或者输出格式混乱无法解析。
解决：
1. 确认依赖语法：在这个项目的实现中，依赖是通过>>运算符和{agent_name.output}占位符定义的。检查拼写是否正确，智能体变量名是否匹配。
2. 检查输出格式：上游智能体的task_expected_output描述了下游智能体将收到的内容。确保上游的输出符合这个描述。例如，如果下游期望JSON，上游就必须输出严格、有效的JSON字符串，而不是自然语言描述。
3. 手动调试：可以先单独运行上游智能体，打印其输出，看是否符合预期。然后再集成到Crew中。

6.4 API费用与性能优化

智能体应用涉及多次LLM调用，成本是需要考虑的因素。

缓存：对于重复性查询（例如，同样的工具调用参数），可以考虑在工具层或Agent层添加简单的缓存机制（例如使用functools.lru_cache），避免重复调用昂贵的外部API或LLM。
精简上下文：ReAct和多智能体模式会累积很长的对话历史。定期总结或剔除过时的中间步骤，可以有效减少token消耗，也能避免模型因上下文过长而性能下降。
模型选型：在开发和测试阶段，可以使用更快、更便宜的模型（如llama3-8b）。在生产环境或对输出质量要求高的环节，再切换到更强大的模型。
异步调用：如果多个工具调用或智能体任务之间没有依赖关系，可以考虑使用异步IO来并行执行，大幅缩短总响应时间。不过，这需要更复杂的任务调度逻辑。

6.5 错误处理与稳定性

工具层容错：所有工具函数都应该有完善的try...except块，返回明确的错误信息，而不是抛出异常导致整个智能体崩溃。
Agent层容错：在ReactAgent或Crew的run方法外层添加异常捕获，记录错误日志，并可能返回一个友好的用户提示，而不是一个Python traceback。
验证与重试：对于关键步骤，可以设计一个验证机制。例如，让一个“验证”智能体检查上一步的输出是否合理，如果不合理，则触发重试或报警。

7. 从学习到生产：下一步的方向

通过这个项目，你已经掌握了智能体四大核心模式的底层原理和实现。但这只是起点。要将这些知识应用于实际生产项目，还需要考虑更多工程化问题：

状态持久化：目前的实现都是内存中的。真实的智能体可能需要处理长时间运行的任务，状态需要保存到数据库或文件中。
更复杂的编排：当前的Crew是线性或简单DAG。现实工作流可能需要条件分支、循环、并行执行等，可以考虑集成像Prefect或Airflow这样的工作流引擎。
记忆与知识库：为智能体配备长期记忆（向量数据库）和领域知识库，使其能进行更深度的对话和决策。
评估与监控：建立一套评估体系，监控智能体任务的成功率、工具调用的准确性、成本消耗等指标。
前端交互：为你的智能体构建一个Web界面、聊天机器人接口或API服务。

这个agentic-patterns-course项目是你工具箱里的一把精良的“螺丝刀”，它让你理解了智能体引擎的每个零件。当你需要构建一辆“汽车”（生产级应用）时，你可能会选择更成熟的“整车厂”（如LangChain），但因为你懂发动机（底层模式）的原理，你就能更好地驾驶、调试甚至改装它。

我个人的建议是，在下一个项目中，先尝试用这个“纯净版”的实现快速验证核心智能体逻辑的可行性。当概念验证通过后，再评估是否需要引入更重量级的框架来获得其生态、工具链和社区支持。这种从底层到上层的学习路径，能让你在快速变化的AI工程领域，始终保有深刻的理解力和解决问题的底气。

查看全文

http://www.jsqmd.com/news/710665/