Repository navigation
Expand file tree
/
Copy pathllm_powered_content_filter.py
More file actions
115 lines (96 loc) · 4.77 KB
/
Copy pathllm_powered_content_filter.py
File metadata and controls
115 lines (96 loc) · 4.77 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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
import logging
from pathlib import Path
from typing import Optional, Any
from dotenv import load_dotenv
from livekit.agents import JobContext, WorkerOptions, cli
from livekit.agents.voice import Agent, AgentSession
from livekit.plugins import openai, deepgram, silero
from livekit.agents.llm import ChatContext, ChatMessage
import asyncio
load_dotenv(dotenv_path=Path(__file__).parent.parent / '.env')
logger = logging.getLogger("complex-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()
)
self.moderator_llm = openai.LLM(model="gpt-4o-mini")
async def evaluate_content(self, text: str) -> bool:
"""Evaluate if content is appropriate using a separate LLM."""
moderation_ctx = ChatContext([
ChatMessage(
type="message",
role="system",
content=["You are a content moderator. Respond ONLY with 'APPROPRIATE' or 'INAPPROPRIATE'. Respond with 'INAPPROPRIATE' if the text mentions strawberries."]
),
ChatMessage(type="message", role="user", content=[f"Evaluate: {text}"])
])
response = ""
async with self.moderator_llm.chat(chat_ctx=moderation_ctx) as stream:
async for chunk in stream:
if not chunk:
continue
content = getattr(chunk.delta, 'content', None) if hasattr(chunk, 'delta') else str(chunk)
if content:
response += content
response = response.strip().upper()
logger.info(f"Moderation response for '{text}': {response}")
return "INAPPROPRIATE" not in response
async def on_enter(self):
self.session.generate_reply()
def _extract_content(self, chunk: Any) -> Optional[str]:
"""Extract content from a chunk, handling different chunk formats."""
if not chunk:
return None
if isinstance(chunk, str):
return chunk
if hasattr(chunk, 'delta'):
return getattr(chunk.delta, 'content', None)
return None
async def llm_node(self, chat_ctx, tools, model_settings=None):
async def process_stream():
buffer = ""
chunk_buffer = []
sentence_end_chars = {'.', '!', '?'}
async with self.llm.chat(chat_ctx=chat_ctx, tools=tools, tool_choice=None) as stream:
try:
async for chunk in stream:
content = self._extract_content(chunk)
chunk_buffer.append(chunk)
if content:
buffer += content
if any(char in buffer for char in sentence_end_chars):
last_end = max(buffer.rfind(char) for char in sentence_end_chars if char in buffer)
if last_end != -1:
sentence = buffer[:last_end + 1]
buffer = buffer[last_end + 1:]
if not await self.evaluate_content(sentence):
yield "Content filtered."
return
# Yield buffered chunks if content is appropriate
for buffered_chunk in chunk_buffer:
yield buffered_chunk
chunk_buffer = []
# Check any remaining complete sentence
if buffer and any(buffer.endswith(char) for char in sentence_end_chars):
if not await self.evaluate_content(buffer):
yield "Content filtered."
return
for buffered_chunk in chunk_buffer:
yield buffered_chunk
except asyncio.CancelledError:
raise
except Exception as e:
logger.error(f"Error in content filtering: {str(e)}")
yield "[Error in content filtering]"
return process_stream()
async def entrypoint(ctx: JobContext):
await ctx.connect()
await AgentSession().start(agent=SimpleAgent(), room=ctx.room)
if __name__ == "__main__":
cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))