-
Notifications
You must be signed in to change notification settings - Fork 18
Expand file tree
/
Copy pathfsm_definition.py
More file actions
76 lines (59 loc) · 2.68 KB
/
fsm_definition.py
File metadata and controls
76 lines (59 loc) · 2.68 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
from chatfaq_sdk import ChatFAQSDK
from chatfaq_sdk.clients import llm_request, query_kis
from chatfaq_sdk.fsm import FSMDefinition, State, Transition
from chatfaq_sdk.layers import Message, StreamingMessage
from chatfaq_sdk.utils import convert_mml_to_llm_format
from chatfaq_sdk.types import CacheConfig
"""
This FSM is a simple example of how to do RAG with the whole knowledge base, given that context windows are
growing for some uses cases we may not need to do retrieval and just pass the whole knowledge base to the LLM.
"""
async def send_greeting(sdk: ChatFAQSDK, ctx: dict):
yield Message("How can I help you today?")
async def send_rag_answer(sdk: ChatFAQSDK, ctx: dict):
messages = convert_mml_to_llm_format(ctx["conv_mml"][1:])
# Get all the knowledge items from the knowledge base
# TODO: Replace with actual knowledge base name.
knowledge_base = "deepseek_v3_paper"
knowledge_items = await query_kis(sdk, knowledge_base)
# Create system prompt with context
system_prompt = """You are a helpful assistant that answers questions based on the provided information.
Your task is to answer the user's question using ONLY the information provided below.
If you cannot answer the question based on the provided information, say so."""
# Add context from knowledge items
if knowledge_items:
context_content = "\n".join([f"- {item.content}" for item in knowledge_items])
system_prompt += f"\n\nInformation:\n{context_content}"
else:
system_prompt += "\n\nNo relevant information found."
# Prepare messages for LLM and until which message we want to cache
messages.insert(
0,
{
"role": "system",
"content": system_prompt,
"cache_control": {"type": "ephemeral"},
},
)
# Generate response
generator = llm_request(
sdk,
"gemini-1.5-flash", # TODO: Replace with your actual LLM config name
use_conversation_context=False,
conversation_id=ctx["conversation_id"],
bot_channel_name=ctx["bot_channel_name"],
messages=messages,
cache_config=CacheConfig(name="deepseek_v3_paper", ttl=3600), # 1 hour
stream=True,
)
references = {"knowledge_items": [ki.dict() for ki in knowledge_items]}
yield StreamingMessage(generator, references=references)
# Define states
greeting_state = State(name="Greeting", events=[send_greeting], initial=True)
answering_state = State(name="Answering", events=[send_rag_answer])
# Define transitions
to_answer = Transition(dest=answering_state)
# Create FSM definition
fsm_definition = FSMDefinition(
states=[greeting_state, answering_state], transitions=[to_answer]
)