Flows
Flows define the logic for running tasks or sets of tasks. They orchestrate how different components work together to achieve complex objectives.
Flow Structure
A flow consists of:
- Tasks: The units of work to be executed
- Orchestration Logic: How tasks are sequenced and coordinated
- Context: Shared data and state
- Error Handling: How failures are managed
Chat Flow Example
Here's an example of a chat flow that handles user interactions:
async def chat_flow(
chat_request: ChatRequest,
db: Session,
):
"""
A flow to handle chat requests.
Compatible with vercel ai sdk.
"""
thread_id = chat_request.thread_id or str(uuid.uuid4())
history = SqlDbHistory()
# Setup history and context
history.add_events(
thread_id,
[UserMessage(content=chat_request.get_user_request().as_llm_message().get("content"))]
)
# Create and configure the flow
flow = cf.Flow(thread_id=thread_id, context=ctx, history=history)
with flow:
async for event, _, _ in await cf.run_tasks_async(
[chat_task],
flow=flow,
run_until=(AnyComplete(min_complete=2) | MaxLLMCalls(15)),
stream=True,
max_llm_calls=20,
handlers=[],
):
# Handle events and streaming
RAG Flow Example
Here's how flows handle Retrieval-Augmented Generation:
async def rag_flow(
rag_request: RAGRequest,
max_llm_calls: int = 5,
):
"""
A flow to handle RAG requests with:
- Query processing
- Document retrieval
- Response generation
"""
thread_id = rag_request.thread_id or str(uuid.uuid4())
history = SqlDbHistory()
if rag_request.query:
history.add_events(
thread_id,
[UserMessage(content=rag_request.query)],
)
Flow Components
Flows integrate several key components:
- History Management: Track conversation and action history
- Event Handling: Process and respond to various events
- Task Orchestration: Coordinate multiple tasks
- Error Recovery: Handle failures and exceptions
Flow Types
The system supports different types of flows:
- Simple Flow: Single task execution
- Complex Flow: Multiple tasks with dependencies
- Chat Flow: Interactive conversation handling
- RAG Flow: Document retrieval and generation
Best Practices
When working with flows:
- Define clear task dependencies
- Handle errors gracefully
- Manage state carefully
- Monitor flow progress
- Use appropriate flow types