Hugging Face's logo

Spaces:
hasanalrobasi
/
anycoder-a79b8d80
Runtime error

App Files Community
anycoder-a79b8d80
File size: 3,995 Bytes
3d0eeb3
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
# app.py
from daggr import GradioNode, InferenceNode, FnNode, Graph
import gradio as gr
from typing import Dict, Any
import requests
import os

# Environment variables for API keys
API_KEYS = {
    "OPENAI": os.getenv("OPENAI_API_KEY"),
    "HUGGINGFACE": os.getenv("HF_API_KEY")
}

# ========== Input Processing Node ==========
def preprocess_inputs(user_input: str, metadata: Dict[str, Any]) -> Dict[str, Any]:
    """Clean and validate inputs with metadata enrichment"""
    return {
        "cleaned_input": user_input.strip(),
        "timestamp": metadata.get("timestamp"),
        "source": metadata.get("source", "web")
    }

input_processor = FnNode(
    fn=preprocess_inputs,
    inputs={
        "user_input": gr.Textbox(label="User Input"),
        "metadata": gr.JSON(label="Metadata")
    },
    outputs={
        "processed_data": gr.JSON(label="Processed Input")
    }
)

# ========== LLM Processing Node ==========
llm_processor = InferenceNode(
    model="meta-llama/Llama-3-70B-Instruct",
    inputs={
        "prompt": gr.Textbox(label="LLM Prompt"),
        "temperature": gr.Slider(0, 1, value=0.7)
    },
    outputs={
        "response": gr.Textbox(label="LLM Response")
    },
    api_key=API_KEYS["HUGGINGFACE"]
)

# ========== Image Generation Node ==========
image_generator = GradioNode(
    space_or_url="stabilityai/stable-diffusion-xl-base-1.0",
    api_name="/generate",
    inputs={
        "prompt": gr.Textbox(label="Image Prompt"),
        "negative_prompt": gr.Textbox(label="Negative Prompt"),
        "steps": gr.Slider(10, 50, value=30)
    },
    outputs={
        "image": gr.Image(label="Generated Image")
    }
)

# ========== API Integration Node ==========
def call_external_api(data: Dict[str, Any]) -> Dict[str, Any]:
    """Generic API caller with error handling"""
    try:
        response = requests.post(
            "https://api.example.com/v1/process",
            json=data,
            headers={"Authorization": f"Bearer {API_KEYS.get('OPENAI')}"},
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    except Exception as e:
        return {"error": str(e)}

api_integrator = FnNode(
    fn=call_external_api,
    inputs={
        "api_data": gr.JSON(label="API Payload")
    },
    outputs={
        "api_response": gr.JSON(label="API Results")
    }
)

# ========== Output Formatter Node ==========
def format_output(llm_response: str, image: Any, api_data: Dict) -> Dict[str, Any]:
    """Create unified output format"""
    return {
        "text_response": llm_response,
        "visual_response": image,
        "api_data": api_data,
        "status": "success"
    }

output_formatter = FnNode(
    fn=format_output,
    inputs={
        "llm_response": gr.Textbox(),
        "image": gr.Image(),
        "api_data": gr.JSON()
    },
    outputs={
        "final_output": gr.JSON(label="Final Output")
    }
)

# ========== Create and Connect Workflow ==========
workflow = Graph(
    name="Global Integration Platform",
    nodes=[
        input_processor,
        llm_processor,
        image_generator,
        api_integrator,
        output_formatter
    ],
    connections=[
        (input_processor.outputs["processed_data"], llm_processor.inputs["prompt"]),
        (input_processor.outputs["processed_data"], image_generator.inputs["prompt"]),
        (input_processor.outputs["processed_data"], api_integrator.inputs["api_data"]),
        (llm_processor.outputs["response"], output_formatter.inputs["llm_response"]),
        (image_generator.outputs["image"], output_formatter.inputs["image"]),
        (api_integrator.outputs["api_response"], output_formatter.inputs["api_data"])
    ]
)

# ========== Launch Application ==========
if __name__ == "__main__":
    workflow.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=True,
        auth=("admin", os.getenv("APP_PASSWORD")),
        favicon_path="https://example.com/favicon.ico"
    )