AutoGen扩展指南：如何为AutoGen框架贡献自定义Agent和能力
·
AutoGen扩展指南：如何为AutoGen框架贡献自定义Agent和能力
1. 引言
在人工智能快速发展的今天，我们正从单智能体时代迈向多智能体协作的新纪元。AutoGen，作为微软研究院推出的多智能体协作框架，为开发者提供了一个强大而灵活的平台，使我们能够构建能够相互交流、协作完成复杂任务的AI代理系统。
然而，随着应用场景的不断扩展，我们经常会发现内置的Agent类型和能力无法完全满足特定需求。这时候，掌握如何扩展AutoGen框架，自定义Agent和能力，就成为了发挥AutoGen真正威力的关键。
在本篇指南中，我们将深入探讨AutoGen的内部工作原理，手把手教你如何创建自定义Agent，添加新的能力，并最终为这个活跃的开源社区做出贡献。无论你是想为你的特定业务场景定制Agent，还是想参与到AutoGen的开源发展中，这篇文章都将为你提供全面的指导。
2. AutoGen核心概念
2.1 什么是AutoGen？
AutoGen是一个开源框架，它简化了基于LLM（大语言模型）的应用程序的编排、优化和自动化。该框架允许开发者创建多个可交互的Agent，这些Agent可以相互对话来解决任务。AutoGen的Agent是可定制和可扩展的，它们可以使用LLM、人类输入或两者的组合。
2.2 AutoGen的核心组件
2.2.1 Agent（代理）
Agent是AutoGen的核心概念。一个Agent是一个实体，它可以发送消息、接收消息、处理消息，并根据消息执行操作。AutoGen提供了几种内置的Agent类型：
- AssistantAgent: 这是一个基于LLM的助手Agent，可以使用工具执行各种任务。
- UserProxyAgent: 这是一个代表用户的代理，可以执行代码并与其他代理交互。
- ConversableAgent: 这是所有可对话Agent的基类，提供了发送和接收消息的基础功能。
2.2.2 对话模式（Conversation Patterns）
AutoGen支持多种对话模式，让Agent之间可以以不同的方式交互：
- 一对一对话: 两个Agent之间的直接对话。
- 群组对话: 多个Agent在一个群组中进行对话，可以是顺序的也可以是并行的。
- 嵌套对话: 一个对话会话内部可以触发另一个对话会话。
2.2.3 工具使用（Tool Usage）
Agent可以使用各种工具来扩展其能力，这些工具可以是内置的，也可以是自定义的。常见的工具包括：
- 代码执行
- 网络搜索
- 数据库查询
- 文件操作
2.3 AutoGen的工作原理
AutoGen的工作原理基于Agent之间的消息传递。当一个Agent发送消息时，接收方Agent会处理这个消息，并可能生成回复。这个过程可以持续下去，直到任务完成或达到某个终止条件。
核心的消息传递流程如下：
- 一个Agent（发送方）创建并发送一条消息。
- 消息被传递给目标Agent（接收方）。
- 接收方Agent处理消息，可能使用LLM、工具或其他机制。
- 接收方Agent生成回复消息。
- 回复被发送回原始发送方或另一个目标Agent。
这个循环继续，直到对话结束。
3. 问题背景与挑战
3.1 为什么需要扩展AutoGen？
虽然AutoGen提供了强大的基础功能，但在实际应用中，我们经常会遇到以下情况：
- 特定领域需求: 内置Agent可能不具备特定行业或领域的专业知识。
- 自定义工作流: 我们可能需要Agent遵循特定的对话模式或决策流程。
- 专有工具集成: 我们可能需要Agent使用公司内部的专有工具或API。
- 性能优化: 对于特定任务，我们可能需要优化Agent的行为以提高效率。
- 新交互模式: 我们可能想要实现创新的Agent交互方式。
3.2 扩展AutoGen面临的挑战
扩展AutoGen框架并非易事，开发者通常会面临以下挑战：
- 理解内部架构: AutoGen有其独特的设计理念和内部工作机制，需要深入理解才能有效扩展。
- 保持兼容性: 自定义扩展需要与现有框架保持兼容，否则可能导致系统不稳定。
- 测试复杂性: 多Agent系统的测试比单Agent系统更复杂，需要考虑各种交互场景。
- 文档完善: 好的扩展需要完善的文档，以便其他开发者能够理解和使用。
在接下来的章节中，我们将探讨如何解决这些挑战，成功地扩展AutoGen框架。
4. 自定义Agent开发详解
4.1 Agent的内部结构
在深入了解如何创建自定义Agent之前，让我们先了解一下AutoGen中Agent的内部结构。
AutoGen中的所有Agent都继承自ConversableAgent类，它提供了基本的对话功能。AssistantAgent和UserProxyAgent是ConversableAgent的子类，它们分别添加了特定的功能。
一个Agent的核心组件包括：
- 系统消息（System Message）: 定义Agent的角色和行为。
- LLM配置: 指定Agent使用的语言模型及其参数。
- 回复生成函数: 定义Agent如何生成对收到消息的回复。
- 工具/函数: Agent可以使用的工具或函数。
- 消息处理钩子（Hooks）: 在消息处理过程中触发的自定义函数。
4.2 创建自定义Agent的步骤
创建自定义Agent通常涉及以下步骤：
- 选择基础Agent类进行继承。
- 定义Agent的系统消息。
- 配置LLM（如果需要）。
- 实现自定义的回复生成逻辑。
- 添加特定的工具或函数。
- 注册消息处理钩子。
让我们通过一个具体的例子来演示这些步骤。
4.3 示例：创建一个代码审查Agent
在这个例子中，我们将创建一个专门用于代码审查的Agent，它可以分析代码，指出潜在问题，并提供改进建议。
首先，让我们安装必要的库：
# 安装AutoGen和相关依赖
!pip install pyautogen
现在，让我们创建我们的代码审查Agent：
import autogen
from autogen import AssistantAgent, UserProxyAgent, ConversableAgent
from typing import Dict, Optional, List, Any, Union
# 定义代码审查Agent
class CodeReviewAgent(ConversableAgent):
"""专门用于代码审查的自定义Agent"""
def __init__(
self,
name: str,
system_message: Optional[str] = None,
llm_config: Optional[Union[Dict, bool]] = None,
**kwargs
):
# 默认系统消息，定义代码审查Agent的角色
if system_message is None:
system_message = """你是一位专业的代码审查专家。你的职责是：
1. 分析提交的代码，识别潜在的问题，包括：
- 语法错误
- 逻辑错误
- 性能问题
- 安全漏洞
- 代码风格不一致
- 缺少文档或注释
2. 提供具体、可操作的改进建议
3. 以友好、建设性的方式提供反馈
4. 在适当的情况下，提供改进后的代码示例
请确保你的审查全面但重点突出，优先关注最严重的问题。"""
# 调用父类构造函数
super().__init__(
name=name,
system_message=system_message,
llm_config=llm_config,
**kwargs
)
# 注册自定义回复函数
self.register_reply(
[ConversableAgent, None],
CodeReviewAgent._generate_code_review_reply
)
def _generate_code_review_reply(
self,
messages: Optional[List[Dict]] = None,
sender: Optional[ConversableAgent] = None,
config: Optional[Any] = None,
) -> Union[str, Dict, None]:
"""生成代码审查回复的自定义逻辑"""
# 如果没有消息，返回None
if messages is None:
return None
# 获取最后一条消息
last_message = messages[-1]
content = last_message.get("content", "")
# 检查是否包含代码
if "```" in content:
# 如果包含代码，使用LLM进行审查
if self.llm_config is not False:
return self.generate_oai_reply(messages, sender)[1]
# 如果不包含代码，提示用户提供代码
return "请提供你想要审查的代码，我会尽力帮助你分析和改进它！"
现在，让我们创建一个用户代理Agent，用于与我们的代码审查Agent交互：
# 配置LLM（这里使用OpenAI的API作为示例）
llm_config = {
"model": "gpt-4",
"api_key": "your-api-key-here", # 请替换为你的API密钥
}
# 创建用户代理
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER", # 不要求人类输入
max_consecutive_auto_reply=10,
is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
code_execution_config={
"work_dir": "coding",
"use_docker": False, # 可选：使用Docker执行代码
},
llm_config=llm_config,
system_message="""你是一位用户，想要让代码审查专家审查你的代码。当代码审查完成后，请回复TERMINATE。""",
)
# 创建代码审查Agent
code_reviewer = CodeReviewAgent(
name="code_reviewer",
llm_config=llm_config,
)
现在，让我们测试我们的代码审查Agent：
# 定义一个需要审查的Python代码示例
code_to_review = """
```python
def calculate_factorial(n):
# 计算阶乘
if n == 0:
return 1
else:
return n * calculate_factorial(n - 1)
# 使用示例
result = calculate_factorial(5)
print(f"5的阶乘是: {result}")
# 另一个示例
for i in range(10):
print(f"{i}的阶乘是: {calculate_factorial(i)}")
“”"
开始对话
user_proxy.initiate_chat(
code_reviewer,
message=f"请帮我审查以下代码：\n{code_to_review}",
)
### 4.4 Agent继承层次与自定义选项
AutoGen的Agent设计采用了灵活的继承结构，允许我们在不同的抽象级别进行自定义：
```mermaid
classDiagram
ConversableAgent <|-- AssistantAgent
ConversableAgent <|-- UserProxyAgent
ConversableAgent <|-- CodeReviewAgent
class ConversableAgent{
+register_reply()
+generate_reply()
+send()
+receive()
}
class AssistantAgent{
+system_message
+llm_config
}
class UserProxyAgent{
+human_input_mode
+code_execution_config
}
class CodeReviewAgent{
+_generate_code_review_reply()
}
根据你的需求，你可以选择在不同的级别进行扩展：
- 继承ConversableAgent: 如果你需要完全控制Agent的行为。
- 继承AssistantAgent: 如果你需要一个基于LLM的助手，但要自定义其行为。
- 继承UserProxyAgent: 如果你需要一个可以执行代码或代表用户的Agent。
4.5 深入理解回复生成机制
AutoGen的回复生成机制是其核心功能之一。理解它的工作原理对于创建有效的自定义Agent至关重要。
当Agent收到消息时，它会按顺序尝试注册的回复生成函数，直到找到一个能够生成非None回复的函数。你可以通过register_reply()方法注册自定义的回复生成函数。
让我们更深入地了解如何使用这个机制：
# 注册多个回复函数，它们将按顺序尝试
agent.register_reply(
trigger=SomeAgentClass, # 触发此回复函数的发送者类型
reply_func=custom_reply_function, # 自定义回复函数
position=0, # 注册位置，数值越小优先级越高
config={"some_config": "value"}, # 传递给回复函数的配置
remove_other_reply_funcs=False, # 是否移除其他回复函数
)
这种灵活的机制允许我们创建复杂的Agent行为，根据不同的上下文和发送者生成不同类型的回复。
5. 自定义能力扩展
除了创建自定义Agent外，我们还可以为AutoGen添加新的能力。这些能力可以通过多种方式实现，包括自定义工具、函数调用、检索增强生成(RAG)等。
5.1 自定义工具和函数
工具和函数允许Agent与外部系统交互或执行特定任务。AutoGen提供了两种主要方式来集成工具：
- 函数调用（Function Calling）: 直接将Python函数暴露给LLM。
- 工具使用（Tool Usage）: 更结构化的方式来定义和使用工具。
让我们看一个例子，展示如何为Agent添加自定义工具：
import autogen
from autogen import AssistantAgent, UserProxyAgent
from typing import Optional, Type, List, Dict, Any
import requests
import json
# 定义一个天气查询工具
def get_weather(location: str, unit: str = "celsius") -> str:
"""
查询指定位置的天气信息
参数:
location: 城市名称或坐标
unit: 温度单位，可选值为"celsius"或"fahrenheit"
返回:
天气信息的字符串描述
"""
# 这里使用一个模拟的天气API，实际应用中你应该使用真实的天气API
# 例如 OpenWeatherMap, WeatherAPI.com 等
# 模拟API响应
weather_data = {
"location": location,
"temperature": 22 if unit == "celsius" else 72,
"condition": "晴朗",
"humidity": 65,
"wind_speed": 12
}
return json.dumps(weather_data, ensure_ascii=False)
# 定义另一个工具：数据分析工具
def analyze_data(data: str, analysis_type: str = "summary") -> str:
"""
分析提供的数据
参数:
data: 要分析的数据，JSON格式字符串
analysis_type: 分析类型，如"summary", "statistics", "trends"等
返回:
分析结果
"""
try:
data_obj = json.loads(data)
if analysis_type == "summary":
return f"数据摘要: 包含 {len(data_obj)} 个条目，键包括: {', '.join(data_obj.keys() if isinstance(data_obj, dict) else ['N/A'])}"
elif analysis_type == "statistics":
# 简单的统计分析示例
if isinstance(data_obj, list) and all(isinstance(x, (int, float)) for x in data_obj):
import statistics
return f"统计分析: 平均值={statistics.mean(data_obj)}, 中位数={statistics.median(data_obj)}, 标准差={statistics.stdev(data_obj) if len(data_obj) > 1 else 'N/A'}"
else:
return "无法进行统计分析: 数据格式不正确"
else:
return f"未知的分析类型: {analysis_type}"
except Exception as e:
return f"分析数据时出错: {str(e)}"
# 配置LLM
llm_config = {
"model": "gpt-4",
"api_key": "your-api-key-here",
"functions": [
{
"name": "get_weather",
"description": "查询指定位置的天气信息",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市名称或坐标",
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位",
},
},
"required": ["location"],
},
},
{
"name": "analyze_data",
"description": "分析提供的数据",
"parameters": {
"type": "object",
"properties": {
"data": {
"type": "string",
"description": "要分析的数据，JSON格式字符串",
},
"analysis_type": {
"type": "string",
"enum": ["summary", "statistics", "trends"],
"description": "分析类型",
},
},
"required": ["data"],
},
},
],
}
# 创建一个能够使用这些工具的Agent
tool_assistant = AssistantAgent(
name="tool_assistant",
llm_config=llm_config,
system_message="你是一个有用的助手，可以使用工具来帮助用户。当你收到工具的执行结果后，请用自然语言总结结果。",
)
# 创建用户代理
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=10,
function_map={
"get_weather": get_weather,
"analyze_data": analyze_data,
},
code_execution_config={"work_dir": "tools"},
)
# 示例1: 使用天气查询工具
user_proxy.initiate_chat(
tool_assistant,
message="请查询北京的天气如何？",
)
# 重置对话
tool_assistant.reset()
user_proxy.reset()
# 示例2: 使用数据分析工具
sample_data = json.dumps([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
user_proxy.initiate_chat(
tool_assistant,
message=f"请分析以下数据并提供统计信息: {sample_data}",
)
5.2 检索增强生成(RAG)集成
检索增强生成(RAG)是一种将信息检索与文本生成相结合的技术，可以帮助Agent访问最新或特定领域的信息，而无需重新训练模型。
让我们看一下如何为AutoGen Agent添加RAG能力：
import autogen
from autogen import AssistantAgent, UserProxyAgent, ConversableAgent
from typing import List, Dict, Optional, Any
import chromadb
from chromadb.utils import embedding_functions
import os
# 定义一个简单的RAG类
class SimpleRAG:
def __init__(self, collection_name: str = "autogen_docs"):
# 初始化ChromaDB客户端
self.client = chromadb.Client()
# 使用OpenAI的嵌入函数（这里使用模拟实现）
# 实际使用中，你需要配置OpenAI API密钥
self.embedding_function = embedding_functions.SentenceTransformerEmbeddingFunction(
model_name="all-MiniLM-L6-v2"
)
# 创建或获取集合
self.collection = self.client.get_or_create_collection(
name=collection_name,
embedding_function=self.embedding_function
)
def add_documents(self, documents: List[str], ids: List[str], metadatas: Optional[List[Dict]] = None):
"""向RAG系统添加文档"""
self.collection.add(
documents=documents,
ids=ids,
metadatas=metadatas
)
def query(self, query_text: str, n_results: int = 3) -> Dict[str, Any]:
"""查询相关文档"""
results = self.collection.query(
query_texts=[query_text],
n_results=n_results
)
return results
# 创建一个带有RAG能力的Agent
class RAGAgent(ConversableAgent):
def __init__(
self,
name: str,
rag_system: SimpleRAG,
system_message: Optional[str] = None,
llm_config: Optional[Dict] = None,
**kwargs
):
if system_message is None:
system_message = """你是一个知识助手，可以使用检索到的信息回答用户问题。
在回答问题时，请使用提供的相关文档内容。如果相关文档中没有足够的信息，请告知用户。
请确保你的回答基于检索到的信息，而不是你的先验知识。"""
super().__init__(
name=name,
system_message=system_message,
llm_config=llm_config,
**kwargs
)
self.rag_system = rag_system
# 注册自定义回复函数
self.register_reply(
[ConversableAgent, None],
RAGAgent._generate_rag_reply,
position=0
)
def _generate_rag_reply(
self,
messages: Optional[List[Dict]] = None,
sender: Optional[ConversableAgent] = None,
config: Optional[Any] = None,
):
if messages is None:
return None
# 获取最后一条消息
last_message = messages[-1]
user_query = last_message.get("content", "")
# 查询相关文档
rag_results = self.rag_system.query(user_query)
# 将检索到的文档添加到上下文中
context = "\n\n".join(rag_results["documents"][0])
# 创建增强的消息列表
enhanced_messages = messages.copy()
enhanced_messages[-1]["content"] = f"""用户问题: {user_query}
相关文档:
{context}
请基于以上相关文档回答用户的问题。"""
# 使用LLM生成回复
if self.llm_config is not False:
return self.generate_oai_reply(enhanced_messages, sender)
return None
# 初始化RAG系统并添加示例文档
rag = SimpleRAG()
rag.add_documents(
documents=[
"AutoGen是一个多智能体对话框架，由微软研究院开发，允许创建交互式AI应用程序。",
"AutoGen中的Agent可以相互对话，使用LLM、人类输入或工具来完成任务。",
"AutoGen支持多种对话模式，包括一对一对话、群组对话和嵌套对话。",
"开发者可以为AutoGen创建自定义Agent，继承自ConversableAgent类。"
],
ids=["doc1", "doc2", "doc3", "doc4"],
metadatas=[
{"topic": "AutoGen简介"},
{"topic": "AutoGen代理"},
{"topic": "对话模式"},
{"topic": "自定义代理"}
]
)
# 配置LLM
llm_config = {
"model": "gpt-4",
"api_key": "your-api-key-here",
}
# 创建RAG Agent和用户代理
rag_agent = RAGAgent(
name="rag_assistant",
rag_system=rag,
llm_config=llm_config,
)
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=3,
code_execution_config=False,
)
# 测试RAG Agent
user_proxy.initiate_chat(
rag_agent,
message="什么是AutoGen？它支持哪些对话模式？"
)
5.3 多模态能力扩展
随着多模态AI的发展，我们可能希望Agent能够处理图像、音频等非文本数据。让我们看看如何为AutoGen添加多模态能力：
import autogen
from autogen import AssistantAgent, UserProxyAgent
from typing import Optional, Dict, Any, List
import base64
from io import BytesIO
from PIL import Image
import requests
# 定义一个多模态Agent
class MultimodalAgent(AssistantAgent):
def __init__(
self,
name: str,
system_message: Optional[str] = None,
llm_config: Optional[Dict] = None,
**kwargs
):
if system_message is None:
system_message = """你是一个多模态AI助手，可以分析图像并回答相关问题。
当用户提供图像时，请仔细分析图像内容，并提供详细的描述和回答。"""
super().__init__(
name=name,
system_message=system_message,
llm_config=llm_config,
**kwargs
)
def process_image(self, image_path: str) -> str:
"""处理图像，将其转换为Base64编码"""
with Image.open(image_path) as img:
# 调整图像大小以降低API调用成本
img.thumbnail((512, 512))
# 保存到内存
buffer = BytesIO()
img.save(buffer, format="PNG")
buffer.seek(0)
# 转换为Base64
img_base64 = base64.b64encode(buffer.read()).decode("utf-8")
return img_base64
def create_multimodal_message(self, text: str, image_path: Optional[str] = None) -> Dict:
"""创建包含文本和图像的多模态消息"""
message_content = [{"type": "text", "text": text}]
if image_path:
image_base64 = self.process_image(image_path)
message_content.append({
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{image_base64}"
}
})
return {"role": "user", "content": message_content}
# 配置支持多模态的LLM（这里使用GPT-4 Vision作为示例）
llm_config = {
"model": "gpt-4-vision-preview",
"api_key": "your-api-key-here",
}
# 创建多模态Agent
multimodal_agent = MultimodalAgent(
name="multimodal_assistant",
llm_config=llm_config,
)
# 创建用户代理
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=3,
code_execution_config=False,
)
# 在实际使用中，你需要替换为实际的图像路径
# 这里我们通过编程方式创建一个简单的图像用于演示
def create_sample_image():
"""创建一个简单的示例图像"""
img = Image.new('RGB', (200, 200), color='red')
img.save('sample_image.png')
return 'sample_image.png'
# 创建示例图像
sample_image_path = create_sample_image()
# 创建多模态消息
multimodal_message = multimodal_agent.create_multimodal_message(
text="请描述这张图片的内容",
image_path=sample_image_path
)
# 开始对话
user_proxy.send(multimodal_message, multimodal_agent)
6. 项目实战：构建一个完整的多Agent协作系统
在本节中，我们将把前面学到的知识整合起来，构建一个完整的多Agent协作系统。这个系统将包括多个专门的Agent，它们协同工作以完成复杂的任务。
6.1 项目概述
我们将构建一个"AI内容创作工作室"，它包含以下Agent：
- 创意总监Agent: 负责整体创意方向和项目管理。
- 内容作者Agent: 负责撰写主要内容。
- 图片编辑Agent: 负责创建和编辑图像。
- 内容审核Agent: 负责审核内容质量和准确性。
- 发布Agent: 负责将最终内容格式化为适合不同平台的形式。
这些Agent将协同工作，接收用户的内容创作请求，并通过多轮对话最终生成高质量的内容。
6.2 环境准备与安装
首先，让我们设置项目环境并安装必要的依赖：
# 创建项目目录
mkdir autogen-content-studio
cd autogen-content-studio
# 创建虚拟环境
python -m venv venv
source venv/bin/activate # Windows上使用: venv\Scripts\activate
# 安装必要的库
pip install pyautogen
pip install python-dotenv
pip install pillow
pip install chromadb
pip install sentence-transformers
6.3 系统架构设计
我们的内容创作工作室将采用以下架构：
6.4 系统实现
现在，让我们开始实现这个系统。首先，创建一个.env文件来存储API密钥和其他配置：
OPENAI_API_KEY=your-openai-api-key-here
接下来，让我们创建主程序文件content_studio.py：
import os
import autogen
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from typing import Dict, Optional, List, Any, Union
from dotenv import load_dotenv
import base64
from io import BytesIO
from PIL import Image
# 加载环境变量
load_dotenv()
# 配置LLM
llm_config = {
"model": "gpt-4",
"api_key": os.getenv("OPENAI_API_KEY"),
"temperature": 0.7,
}
# 创建用户代理
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="TERMINATE",
max_consecutive_auto_reply=10,
is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
code_execution_config={
"work_dir": "content_studio",
"use_docker": False,
},
llm_config=False,
system_message="你是用户的代理，负责代表用户与内容创作工作室交互。当内容创作完成后，请回复TERMINATE。",
)
# 创建创意总监Agent
creative_director = AssistantAgent(
name="creative_director",
llm_config=llm_config,
system_message="""你是创意总监，负责管理内容创作项目的整体方向和质量。你的职责包括：
1. 理解用户需求并将其转化为明确的创作方向
2. 为内容作者和图片编辑提供指导和反馈
3. 协调团队成员之间的工作流程
4. 审核中间结果并提供改进建议
5. 最终确认内容质量
请以专业、建设性的方式与团队成员沟通，确保项目高效推进并达到高质量标准。""",
)
# 创建内容作者Agent
content_author = AssistantAgent(
name="content_author",
llm_config=llm_config,
system_message="""你是一位专业的内容作者，负责撰写高质量的文章、博客或其他形式的内容。你的职责包括：
1. 根据创意总监的指导和用户需求撰写内容
2. 确保内容结构清晰、逻辑连贯、引人入胜
3. 使用恰当的语调和风格
4. 根据反馈进行修改和优化
5. 确保内容准确性和原创性
请专注于创建优秀的内容，并在创作过程中与创意总监保持沟通。""",
)
# 创建图片编辑Agent
image_editor = AssistantAgent(
name="image_editor",
llm_config=llm_config,
system_message="""你是一位专业的图片编辑和视觉创意专家，负责为内容创建和编辑图像。你的职责包括：
1. 根据内容主题和创意总监的指导生成图像描述
2. 提供详细的图像创建建议
3. 描述图像的风格、色彩、构图等要素
4. 根据反馈调整图像描述
5. 确保视觉元素与内容主题一致
虽然你不能直接生成图像，但你可以提供详细的图像描述，以便用户或其他工具生成实际图像。""",
)
# 创建内容审核Agent
content_reviewer = AssistantAgent(
name="content_reviewer",
llm_config=llm_config,
system_message="""你是一位专业的内容审核员，负责审查内容的质量、准确性和适宜性。你的职责包括：
1. 检查内容的语法、拼写和格式错误
2. 验证事实准确性和引用来源
3. 评估内容的逻辑性和连贯性
4. 检查内容是否符合道德标准和法规要求
5. 提供具体的修改建议
请以客观、专业的态度进行审核，并提供详细的反馈和改进建议。""",
)
# 创建发布Agent
publisher = AssistantAgent(
name="publisher",
llm_config=llm_config,
system_message="""你是一位发布专家，负责将最终内容格式化为适合不同平台的形式。你的职责包括：
1. 根据目标平台优化内容格式
2. 确保内容在不同设备上的良好展示效果
3. 添加适当的标签、摘要和元数据
4. 为不同平台创建略有差异的内容版本
5. 最终检查内容的完整性和一致性
请专注于将内容准备好进行发布，并提供不同平台的内容版本。""",
)
# 创建一个GroupChat，定义Agent之间的交互规则
groupchat = GroupChat(
agents=[user_proxy, creative_director, content_author, image_editor, content_reviewer, publisher],
messages=[],
max_round=20,
speaker_selection_method="auto", # 自动选择下一个发言者
allow_repeat_speaker=False, # 不允许连续发言
)
# 创建GroupChatManager来管理群聊
manager = GroupChatManager(
groupchat=groupchat,
llm_config=llm_config,
system_message="""你是内容创作工作室的经理，负责协调各个Agent之间的工作流程。
请根据当前对话状态和各Agent的专业领域，选择最合适的下一个发言者。
确保内容创作流程顺利进行：从理解需求，到内容创作，图像设计，内容审核，最后到发布准备。""",
)
# 开始内容创作工作室
def start_content_studio():
print("欢迎使用AI内容创作工作室！")
print("请描述您想要创建的内容：")
# 获取用户输入
user_request = input("> ")
# 初始化对话
user_proxy.initiate_chat(
manager,
message=f"""我需要以下内容创作服务：
{user_request}
请创意总监先了解我的需求，然后协调团队完成这个内容创作项目。""",
)
if __name__ == "__main__":
start_content_studio()
现在，让我们再添加一些自定义函数和能力，使我们的内容创作工作室更加强大。我们将创建一个扩展文件content_studio_extensions.py：
import autogen
from autogen import AssistantAgent, UserProxyAgent
from typing import Dict, Optional, List, Any, Callable
import os
import json
import re
# 定义内容模板系统
class ContentTemplateSystem:
def __init__(self, templates_dir: str = "templates"):
self.templates_dir = templates_dir
self.templates = {}
# 确保模板目录存在
os.makedirs(templates_dir, exist_ok=True)
# 加载默认模板
self._load_default_templates()
def _load_default_templates(self):
"""加载默认内容模板"""
# 博客文章模板
self.templates["blog_post"] = """
# {title}
## 简介
{introduction}
## 主要内容
{main_content}
## 结论
{conclusion}
## 关于作者
{author_bio}
"""
# 社交媒体帖子模板
self.templates["social_media"] = """
{hook}
{main_message}
{call_to_action}
{hashtags}
"""
def add_template(self, name: str, template: str):
"""添加自定义模板"""
self.templates[name] = template
# 保存到文件
with open(os.path.join(self.templates_dir, f"{name}.txt"), "w") as f:
f.write(template)
def fill_template(self, template_name: str, **kwargs) -> str:
"""填充模板"""
if template_name not in self.templates:
# 尝试从文件加载
template_path = os.path.join(self.templates_dir, f"{template_name}.txt")
if os.path.exists(template_path):
with open(template_path, "r") as f:
self.templates[template_name] = f.read()
else:
raise ValueError(f"模板 '{template_name}' 不存在")
template = self.templates[template_name]
# 替换模板变量
for key, value in kwargs.items():
template = template.replace(f"{{{key}}}", value)
# 移除未替换的变量
template = re.sub(r"\{[^{}]+\}", "", template)
return template
# 为内容作者Agent添加模板功能
def enhance_content_author_with_templates(agent: AssistantAgent, template_system: ContentTemplateSystem):
"""为内容作者Agent添加模板功能"""
def use_template(template_name: str, **kwargs) -> str:
"""使用模板生成内容"""
try:
return template_system.fill_template(template_name, **kwargs)
except Exception as e:
return f"使用模板时出错: {str(e)}"
def list_templates() -> str:
"""列出可用的模板"""
templates = list(template_system.templates.keys())
# 检查模板目录中的其他模板
if os.path.exists(template_system.templates_dir):
for filename in os.listdir(template_system.templates_dir):
if filename.endswith(".txt"):
template_name = filename[:-4]
if template_name not in templates:
templates.append(template_name)
return "可用模板: " + ", ".join(templates)
# 更新Agent的函数映射
if hasattr(agent, "function_map"):
agent.function_map["use_template"] = use_template
agent.function_map["list_templates"] = list_templates
# 更新Agent的LLM配置以包含这些函数
if agent.llm_config and isinstance(agent.llm_config, dict) and "functions" in agent.llm_config:
agent.llm_config["functions"].extend([
{
"name": "use_template",
"description": "使用模板生成内容",
"parameters": {
"type": "object",
"properties": {
"template_name": {
"type": "string",
"description": "要使用的模板名称",
},
},
"required": ["template_name"],
# 注意：这是一个简化版本，实际中我们需要更灵活地处理额外参数
},
},
{
"name": "list_templates",
"description": "列出可用的模板",
"parameters": {
"type": "object",
"properties": {},
},
},
])
# 更新Agent的系统消息
original_system_message = agent.system_message
agent.update_system_message(f"""{original_system_message}
你现在可以使用内容模板系统来帮助你更高效地创建内容。你可以：
1. 使用list_templates函数查看可用的模板
2. 使用use_template函数选择并填充模板
请根据内容类型选择合适的模板，这将帮助你保持内容结构的一致性。""")
# 定义内容SEO优化工具
class SEOOptimizer:
def __init__(self):
# 常见SEO关键词（示例）
self.common_keywords = [
"最佳", "指南", "如何", "什么是", "为什么",
"教程", "示例", "技巧", "建议", "方法",
"2023", "2024", "最新", "完整", "简单",
"免费", "在线", "工具", "服务", "解决方案"
]
def analyze_content(self, content: str) -> Dict[str, Any]:
"""分析内容的SEO友好性"""
# 计算内容长度
content_length = len(content)
# 计算段落数
paragraphs = content.split("\n\n")
paragraph_count = len(paragraphs)
# 计算标题数
headings = re.findall(r"^#{1,6}\s+.+$", content, re.MULTILINE)
heading_count = len(headings)
# 检查是否包含常见关键词
found_keywords = [kw for kw in self.common_keywords if kw in content]
# 简单的可读性评分（基于句子长度和段落长度）
sentences = re.split(r"[.!?]+", content)
avg_sentence_length = sum(len(s.split()) for s in sentences) / max(len(sentences), 1)
# 生成SEO建议
suggestions = []
if content_length < 300:
suggestions.append("内容长度较短，建议增加更多有价值的内容")
if heading_count < 3:
suggestions.append("建议添加更多标题以提高内容结构清晰度")
if avg_sentence_length > 25:
suggestions.append("句子偏长，建议使用更简短清晰的句子")
if not found_keywords:
suggestions.append("建议在内容中自然地融入一些常见的搜索关键词")
return {
"content_length": content_length,
"paragraph_count": paragraph_count,
"heading_count": heading_count,
"found_keywords": found_keywords,
"avg_sentence_length": avg_sentence_length,
"suggestions": suggestions
}
def suggest_title(self, topic: str) -> List[str]:
"""根据主题建议SEO友好的标题"""
templates = [
"{topic}：完整指南与实用技巧",
"什么是{topic}？初学者必读指南",
"如何掌握{topic}：专家分享的秘诀",
"{topic}的10个最佳实践（2024年更新）",
"为什么{topic}很重要：你需要知道的一切"
]
return [template.format(topic=topic) for template in templates]
# 为内容审核Agent添加SEO优化能力
def enhance_reviewer_with_seo(agent: AssistantAgent, seo_optimizer: SEOOptimizer):
"""为内容审核Agent添加SEO优化能力"""
def analyze_seo(content: str) -> str:
"""分析内容的SEO友好性"""
try:
analysis = seo_optimizer.analyze_content(content)
return json.dumps(analysis, ensure_ascii=False, indent=2)
except Exception as e:
return f"SEO分析时出错: {str(e)}"
def suggest_titles(topic: str) -> str:
"""建议SEO友好的标题"""
try:
titles = seo_optimizer.suggest_title(topic)
return json.dumps(titles, ensure_ascii=False, indent=2)
except Exception as e:
return f"生成标题建议时出错: {str(e)}"
# 更新Agent的函数
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
1
0- 0
已为社区贡献209条内容
所有评论(0)