CrewAI Community Version(二)——Agent
目录
- 1. Agent总览
- 2. Agent属性
- 3. 创建Agent
- 3.1 YAML配置
- 3.2 直接用代码定义
- 3.3 运行结果
- 参考
1. Agent总览
在CrewAI框架中，Agent是一个能具备下列能力的自主单元：
1. 执行特定的任务
2. 基于它的角色和目标进行决策
3. 使用工具完成任务
4. 与其他Agent交流协作
5. 维持交互时的记忆
6. 条件允许的情况下委派任务
2. Agent属性
| 属性 | 参数 | 类型 | 含义 |
|---|---|---|---|
| 角色 | role | str | 定义Agent在Crew中的能力 |
| 目标 | goal | str | 指导Agent进行决策 |
| 背景信息 | backstory | str | 为Agent提供上下文和个性特征，丰富交互 |
| 大语言模型 | llm | Union[str, LLM, any] | 增强Agent的大语言模型，默认为OPENAI_MODEL_NAME环境变量指定的模型或GPT-4 |
| 工具 | tools | List[BaseTool] | Agent可用的工具，默认为空列表 |
| 函数调用的大语言模型 | function_calling_llm | Optional[Any] | 用于调用工具的大语言模型，如果指定，将会覆盖crew的llm |
| 最大迭代 | max_iter | int | Agent提供最佳答案之前的最大迭代数，默认为20 |
| 每分钟最大请求数 | max_rpm | Optional[int] | 每分钟最大请求限制以避免速率限制 |
| 最大执行时间 | max_execution_time | Optional[int] | 任务执行的最大时间 |
| 记忆(疑似参数不存在) | memory | bool | Agent是否应该保持交互时的记忆 |
| 详细日志 | verbose | bool | 打开详细的执行日志以便调试，默认为false |
| 允许委派任务 | allow_delegation | bool | 允许Agent将任务委派给其他Agent，默认为False |
| 步骤回调 | step_callback | Optional[Any] | 每个Agent步骤调用的函数，覆盖crew回调 |
| Cache | cache | bool | 为工具启用缓存，默认为True |
| 系统提示词模板 | system_template | Optional[str] | 自定义Agent系统提示词模板 |
| 提示词模板 | prompt_template | Optional[str] | 自定义Agent提示词模板 |
| 响应模板 | response_template | Optional[str] | 自定义Agent响应模板 |
| 允许代码执行 | allow_code_execution | Optional[bool] | 为Agent启用代码执行，默认为False |
| 最大重试限制 | max_retry_limit | int | 发生错误时重试的最大次数，默认值为2 |
| 尊重上下文窗口 | respect_context_window | bool | 通过总结将消息保持在上下文窗口大小之内，默认为True |
| 代码执行模式 | code_execution_mode | Literal[“safe”, “unsafe”] | 代码执行模式：safe(Docker)或unsafe(直接)，默认为safe |
| 嵌入器 | embedder | Optional[Dict[str, Any]] | Agent使用的嵌入器的配置 |
| 知识来源 | knowledge_sources | Optional[List[BaseKnowledgeSource]] | Agent使用的知识来源 |
| 使用系统提示词模板 | use_system_prompt | Optional[bool] | 是否使用系统提示词模板(支持o1模型)，默认为True |
3. 创建Agent
3.1 YAML配置
通过crewai create crew <your-project-name>后，在<your-project-name>/src/<your-project-name>/config/agents.yaml中以如下的形式定义Agents：
code_assistant:role: >高级Python开发工程师goal: >编写和调试Python代码backstory: >有10年开发经验的Python开发专家
在<your-project-name>/src/<your-project-name>/config/tasks.yaml中以如下的形式定义Tasks：
coding_task:description: >1. 根据用户的要求编写Python代码。2. 解释用户给出的Python代码。3. 调试并修改用户给出的Python代码expected_output: >根据用户的要求或代码{code}，给出一份正确且高效的Python代码agent: code_assistantlibrary_used: ['scikit-learn', 'numpy', 'matplotlib']
在<your-project-name>/src/<your-project-name>/crew.py中创建一个继承CrewBase的类：
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task# If you want to run a snippet of code before or after the crew starts,
# you can use the @before_kickoff and @after_kickoff decorators
# https://docs.crewai.com/concepts/crews#example-crew-class-with-decorators@CrewBase
class CodeAssistant():"""CodeAssistant crew"""# Learn more about YAML configuration files here:# Agents: https://docs.crewai.com/concepts/agents#yaml-configuration-recommended# Tasks: https://docs.crewai.com/concepts/tasks#yaml-configuration-recommendedagents_config = 'config/agents.yaml'tasks_config = 'config/tasks.yaml'# If you would like to add tools to your agents, you can learn more about it here:# https://docs.crewai.com/concepts/agents#agent-tools@agentdef code_assistant(self) -> Agent:return Agent(config=self.agents_config['code_assistant'],allow_code_execution=True,code_execution_mode='safe',max_execution_time=300,max_retry_limit=3,verbose=True)@taskdef coding_task(self) -> Task:return Task(config=self.tasks_config['coding_task'],)@crewdef crew(self) -> Crew:"""Creates the CodeAssistant crew"""# To learn how to add knowledge sources to your crew, check out the documentation:# https://docs.crewai.com/concepts/knowledge#what-is-knowledgereturn Crew(agents=self.agents, # Automatically created by the @agent decoratortasks=self.tasks, # Automatically created by the @task decoratorprocess=Process.sequential,verbose=True,# process=Process.hierarchical, # In case you wanna use that instead https://docs.crewai.com/how-to/Hierarchical/)
在<your-project-name>/src/<your-project-name>/main.py中修改inputs：
mycode = """
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.datasets import load_iris
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as pltplt.style.use("deeplearning.mplstyle")
plt.rcParams["font.family"] = "sans-serif"
plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei"] # 设置中文字体
plt.rcParams["axes.unicode_minus"] = False # 正确显示负号class MySoftMaxRegression:def __init__(self, alpha, la=1, penalty=None, epoch=1e3):self.m = Noneself.c = None # 类别个数self.w = Noneself.b = Noneself.dict = Noneself.reverse_dict = Noneself.alpha = alphaself.la = laself.e = epochself.ln = penalty is Noneself.l1 = penalty == "l1"self.l2 = penalty == "l2"self.jwb = []def _onehotencoder(self, y):y_unique = np.unique(y)t = tuple(zip(y_unique, np.arange(len(y_unique))))self.dict = {key: value for key, value in t}self.reverse_dict = {key: value for value, key in t}y_temp = np.zeros((len(y), len(np.unique(y))))for i in range(len(y)):y_temp[i][self.dict.get(y[i])] = 1return y_tempdef _softmax(self, x):x_temp = np.exp(np.dot(x, self.w.T) + self.b.T)x_sum = np.sum(x_temp, axis=1).reshape(len(x), 1)return x_temp / x_sumdef _loss(self, x, y):loss = -np.sum(np.log(self._softmax(x)) * y)if self.ln:return lossif self.l1:return loss + np.sum(np.abs(self.w))if self.l2:return loss + np.sum(self.w ** 2)def plotlearingcurve(self, st=0, end=None):end = int(self.e) if end is None else endplt.figure(1)plt.plot(np.arange(st, end + 1), self.jwb[st:end + 1])plt.xlabel("迭代次数")plt.ylabel("代价函数值")plt.title(f"学习曲线图从迭代次数{st}开始")def fit(self, x, y):self.m = len(x)self.c = len(np.unique(y))self.w = np.zeros((self.c, len(x[0])))self.b = np.zeros((self.c, 1))y_onehot = self._onehotencoder(y)self.jwb.append(self._loss(x, y_onehot))count = 0while count < self.e:w_temp = self.w - self.alpha * np.dot(((self._softmax(x) - 1) * y_onehot).T, x) / self.mif self.l1:w_temp = w_temp - self.alpha * self.la * np.sign(w_temp)if self.l2:w_temp = w_temp - 2 * self.alpha * self.la * w_tempself.b = self.b - self.alpha * np.sum((self._softmax(x) - 1) * y_onehot, axis=0).reshape(self.c, 1) / self.mself.w = w_tempself.jwb.append(self._loss(x, y_onehot))count += 1print(self.jwb)def predict(self, x):return self._softmax(x)def parameters(self):return self.w, self.b# 加载数据集，这里以鸢尾花数据集为例
iris = load_iris()
X = iris.data
y = iris.target# 分割数据集，60%作为训练集，40%作为测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.4, random_state=42)
scaler = StandardScaler()
Xn_train = scaler.fit_transform(X_train)
Xn_test = scaler.transform(X_test)
SoftReg = MySoftMaxRegression(1)
SoftReg.fit(Xn_train, y_train)
SoftReg.plotlearingcurve()
y_pred = np.argmax(SoftReg.predict(Xn_test), axis=1)
print(np.sum(y_pred == y_test) / len(y_test))
print(SoftReg.parameters())
"""def run():"""Run the crew."""inputs = {"code": mycode}try:CodeAssistant().crew().kickoff(inputs=inputs)except Exception as e:raise Exception(f"An error occurred while running the crew: {e}")def train():"""Train the crew for a given number of iterations."""inputs = {"code": mycode}try:CodeAssistant().crew().train(n_iterations=int(sys.argv[1]), filename=sys.argv[2], inputs=inputs)except Exception as e:raise Exception(f"An error occurred while training the crew: {e}")def replay():"""Replay the crew execution from a specific task."""try:CodeAssistant().crew().replay(task_id=sys.argv[1])except Exception as e:raise Exception(f"An error occurred while replaying the crew: {e}")def test():"""Test the crew execution and returns the results."""inputs = {"code": mycode}try:CodeAssistant().crew().test(n_iterations=int(sys.argv[1]), openai_model_name=sys.argv[2], inputs=inputs)except Exception as e:raise Exception(f"An error occurred while testing the crew: {e}")
3.2 直接用代码定义
只需将上面crew.py中的Agent定义以如下方式更改：
@agentdef code_assistant(self) -> Agent:return Agent(role='高级Python开发工程师',goal='编写和调试Python代码',backstory='有10年开发经验的Python开发专家',allow_code_execution=True,code_execution_mode='safe',max_execution_time=300,max_retry_limit=3,verbose=True)
3.3 运行结果
在项目路径下先安装依赖，然后运行：
crewai install
crewai run
由于设置了allow_code_execution=True和code_execution_mode=‘safe’，因此需要在电脑上安装docker，否则将无法运行。
总的来说，CrewAI的代码执行相当差劲，从接下来的运行结果图中就可以看出。
下图是运行过程创建的docker容器，容器名为code-interpreter。
下面是运行结果：
这里对inputs中的code进行了分析，得到了这段代码的目的是实现softmax回归的分析。
这里执行代码，发现缺少sklearn即scikit-learn库。
下面三张图都在尝试解决依赖问题，但是发现安装scikit-learn的时候缺少C++编译。
修改后的代码：
import numpy as np
import matplotlib.pyplot as pltplt.rcParams['font.family'] = 'sans-serif'
plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei'] # 设置中文字体
plt.rcParams['axes.unicode_minus'] = False # 正确显示负号class MySoftMaxRegression:def __init__(self, alpha=0.1, la=1, penalty=None, epoch=1000):self.m = None # 样本数量self.c = None # 类别个数self.w = None # 权重矩阵self.b = None # 偏置向量self.dict = None # 类别到索引的映射self.reverse_dict = None # 索引到类别的映射self.alpha = alpha # 学习率self.la = la # 正则化系数self.epoch = epoch # 迭代次数self.ln = penalty is None # 无正则化self.l1 = penalty == "l1" # L1正则化self.l2 = penalty == "l2" # L2正则化self.loss_history = [] # 损失历史记录def _onehotencoder(self, y):"""将类别标签转换为one-hot编码"""y_unique = np.unique(y)self.dict = {label: idx for idx, label in enumerate(y_unique)}self.reverse_dict = {idx: label for idx, label in enumerate(y_unique)}y_onehot = np.zeros((len(y), len(y_unique)))for i, label in enumerate(y):y_onehot[i, self.dict[label]] = 1return y_onehotdef _softmax(self, x):"""计算softmax概率"""scores = np.dot(x, self.w.T) + self.b.T# 数值稳定性处理scores -= np.max(scores, axis=1, keepdims=True)exp_scores = np.exp(scores)return exp_scores / np.sum(exp_scores, axis=1, keepdims=True)def _loss(self, x, y_onehot):"""计算损失函数"""probs = self._softmax(x)# 交叉熵损失loss = -np.sum(np.log(probs) * y_onehot) / len(x)# 正则化项if self.l1:loss += self.la * np.sum(np.abs(self.w))elif self.l2:loss += self.la * np.sum(self.w ** 2)return lossdef plot_learning_curve(self):"""绘制学习曲线"""plt.figure(figsize=(10, 6))plt.plot(range(len(self.loss_history)), self.loss_history)plt.xlabel("迭代次数")plt.ylabel("损失值")plt.title("学习曲线")plt.grid(True)plt.show()def fit(self, x, y):"""训练模型"""self.m = len(x)self.c = len(np.unique(y))# 初始化参数self.w = np.random.randn(self.c, x.shape[1]) * 0.01self.b = np.zeros((self.c, 1))y_onehot = self._onehotencoder(y)for i in range(self.epoch):# 前向传播probs = self._softmax(x)# 计算梯度dw = np.dot((probs - y_onehot).T, x) / self.mdb = np.sum(probs - y_onehot, axis=0).reshape(self.c, 1) / self.m# 应用正则化if self.l1:dw += self.la * np.sign(self.w)elif self.l2:dw += 2 * self.la * self.w# 参数更新self.w -= self.alpha * dwself.b -= self.alpha * db# 记录损失current_loss = self._loss(x, y_onehot)self.loss_history.append(current_loss)# 每100次打印损失if i % 100 == 0:print(f"Epoch {i}, Loss: {current_loss:.4f}")def predict(self, x):"""预测类别"""probs = self._softmax(x)return np.array([self.reverse_dict[idx] for idx in np.argmax(probs, axis=1)])def score(self, x, y):"""计算准确率"""y_pred = self.predict(x)return np.mean(y_pred == y)def get_parameters(self):"""获取模型参数"""return self.w, self.b# 示例数据集 - 手动创建简单的多分类数据集
def create_dataset():np.random.seed(42)# 类别0的数据x0 = np.random.randn(50, 2) + np.array([2, 2])y0 = np.zeros(50)# 类别1的数据x1 = np.random.randn(50, 2) + np.array([-2, 2])y1 = np.ones(50)# 类别2的数据x2 = np.random.randn(50, 2) + np.array([0, -2])y2 = np.ones(50) * 2x = np.vstack([x0, x1, x2])y = np.hstack([y0, y1, y2])# 打乱数据indices = np.arange(len(x))np.random.shuffle(indices)return x[indices], y[indices]# 数据标准化
def standardize(x):mean = np.mean(x, axis=0)std = np.std(x, axis=0)return (x - mean) / std# 创建并准备数据
x, y = create_dataset()
x = standardize(x)# 分割训练集和测试集
split = int(0.8 * len(x))
x_train, y_train = x[:split], y[:split]
x_test, y_test = x[split:], y[split:]# 创建并训练模型
model = MySoftMaxRegression(alpha=0.1, epoch=1000)
model.fit(x_train, y_train)# 评估模型
print(f"训练集准确率: {model.score(x_train, y_train):.4f}")
print(f"测试集准确率: {model.score(x_test, y_test):.4f}")# 绘制学习曲线
model.plot_learning_curve()虽然修改后的代码仍然能执行并没有错误修改，但是没有在docker里执行并得到结果，因此根据需求编写代码任务得到的代码可信度存疑。
参考
https://docs.crewai.com/concepts/agents