Pythonworkflow.py
Copy
1
from dataclasses import dataclass, field
2
from openai import OpenAI
3
 
4
client = OpenAI()
5
 
6
@dataclass
7
class WorkflowState:
8
    user_input: str
9
    intent: str = ""
10
    fetched_data: str = ""
11
    final_response: str = ""
12
    error: str = ""
13
 
14
def extract_intent(state: WorkflowState) -> WorkflowState:
15
    resp = client.chat.completions.create(
16
        model="gpt-4o-mini",
17
        messages=[
18
            {"role": "system", "content": "Extract the user intent as one short phrase."},
19
            {"role": "user", "content": state.user_input},
20
        ],
21
    )
22
    state.intent = resp.choices[0].message.content or ""
23
    return state
24
 
25
def fetch_data(state: WorkflowState) -> WorkflowState:
26
    # Stub: in real use, call an API or database
27
    state.fetched_data = f"Data relevant to: {state.intent}"
28
    return state
29
 
30
def generate_response(state: WorkflowState) -> WorkflowState:
31
    resp = client.chat.completions.create(
32
        model="gpt-4o",
33
        messages=[
34
            {"role": "system", "content": "Use the provided data to answer the user."},
35
            {"role": "user", "content": f"Query: {state.user_input}\nData: {state.fetched_data}"},
36
        ],
37
    )
38
    state.final_response = resp.choices[0].message.content or ""
39
    return state
40
 
41
def run_workflow(user_input: str) -> WorkflowState:
42
    state = WorkflowState(user_input=user_input)
43
    for node in [extract_intent, fetch_data, generate_response]:
44
        try:
45
            state = node(state)
46
        except Exception as e:
47
            state.error = str(e)
48
            break
49
    return state
50
 
51
if __name__ == "__main__":
52
    result = run_workflow("What are the best practices for async Python?")
53
    print(result.final_response)
54