Repository navigation
Expand file tree
/
Copy pathsimple_content_filter.py
More file actions
62 lines (50 loc) · 1.9 KB
/
Copy pathsimple_content_filter.py
File metadata and controls
62 lines (50 loc) · 1.9 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
import logging
from pathlib import Path
from typing import AsyncIterable, Optional
from dotenv import load_dotenv
from livekit import rtc
from livekit.agents import JobContext, WorkerOptions, cli
from livekit.agents.voice import Agent, AgentSession
from livekit.plugins import openai, deepgram, silero
import asyncio
load_dotenv(dotenv_path=Path(__file__).parent.parent / '.env')
logger = logging.getLogger("simple-content-filter")
logger.setLevel(logging.INFO)
class SimpleAgent(Agent):
def __init__(self) -> None:
super().__init__(
instructions="""
You are a helpful agent.
""",
stt=deepgram.STT(),
llm=openai.LLM(),
tts=openai.TTS(),
vad=silero.VAD.load()
)
async def on_enter(self):
self.session.generate_reply()
async def llm_node(
self, chat_ctx, tools, model_settings=None
):
async def process_stream():
async with self.llm.chat(chat_ctx=chat_ctx, tools=tools, tool_choice=None) as stream:
async for chunk in stream:
if chunk is None:
continue
content = getattr(chunk.delta, 'content', None) if hasattr(chunk, 'delta') else str(chunk)
if content is None:
yield chunk
continue
offensive_terms = ['fail']
print(content)
yield "CONTENT FILTERED" if any(term in content.lower() for term in offensive_terms) else chunk
return process_stream()
async def entrypoint(ctx: JobContext):
await ctx.connect()
session = AgentSession()
await session.start(
agent=SimpleAgent(),
room=ctx.room
)
if __name__ == "__main__":
cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))