code generation

pyproject.toml

@@ -30,6 +30,7 @@ pydantic-ai = {extras = ["logfire"], version = "^0.0.14"}
 dspy-ai = "^2.5.43"
 cloudpickle = "^3.1.0"
 dslmodel = "^2024.12.22.2"
+pyttsx3 = "^2.98"
 
 [tool.poetry.group.test.dependencies]  # https://python-poetry.org/docs/master/managing-dependencies/
 coverage = { extras = ["toml"], version = ">=7.4.4" }

src/pyn8n/api.py

@@ -29,8 +29,8 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     coloredlogs.install()
 
     # print(n8n_router.routes)
-    # for route in app.routes:
-    #     print(f"Route path: {route.path}, methods: {route.methods}")
+    for route in app.routes:
+        print(f"Route path: {route.path}, methods: {route.methods}")
     yield
     # Shutdown events.
 
@@ -47,7 +47,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 )
 
 # Include the n8n router
-app.include_router(n8n_router, prefix="/actions")
+app.include_router(n8n_router, prefix="/nodes")
 
 
 @app.get("/compute")
@@ -114,3 +114,10 @@ async def voice_endpoint(payload: VoicePayload):
     resp = await VoiceResponse.from_prompt(prompt, model="groq:llama3-groq-8b-8192-tool-use-preview")
     print(f"Response: {resp}")
     return resp
+
+
+# Add the main entry point
+if __name__ == "__main__":
+    import uvicorn  # Import Uvicorn to run the app
+
+    uvicorn.run("pyn8n.api:app", host="0.0.0.0", port=8000, reload=True)

src/pyn8n/jinja_models.py

@@ -0,0 +1,92 @@
+from jinja2 import FileSystemLoader
+from pydantic import BaseModel, Field
+from typing import List, Optional, Dict, Union
+
+
+from pydantic import BaseModel, Field
+from typing import Optional
+
+from jinja2 import Environment
+
+from dslmodel import DSLModel, init_instant
+
+env = Environment(loader=FileSystemLoader("templates"), trim_blocks=True, lstrip_blocks=True)
+
+
+class Argument(BaseModel):
+    """Represents an argument in a command with Annotated style."""
+    name: str = Field(..., description="The name of the argument.")
+    type: str = Field(..., description="The type of the argument (e.g., str, int).")
+    help: Optional[str] = Field(None, description="The help text for the argument.")
+    rich_help_panel: Optional[str] = Field(None, description="The help panel category for the argument.")
+    default: Optional[str] = Field(None, description="The default value for the argument.")
+
+
+class Option(BaseModel):
+    name: str
+    type: str
+    prompt: Optional[Union[bool, str]] = None
+    help: Optional[str] = None
+    rich_help_panel: Optional[str] = None
+    show_default: Optional[bool] = None
+    default: Optional[str] = None
+
+
+class Signature(DSLModel):
+    function_name: str
+    arguments: List[Argument]
+    options: List[Option]
+    docstring: str
+
+
+class Docstring(BaseModel):
+    """Represents the docstring for a command."""
+    summary: str = Field(..., description="A short summary of the command.")
+    arguments: List[Argument] = Field(..., description="Arguments documented in the docstring.")
+    options: List[Option] = Field(..., description="Options documented in the docstring.")
+    returns: str = Field(..., description="Description of the return value.")
+
+
+class Command(BaseModel):
+    """Represents a CLI command."""
+    name: str = Field(..., description="The name of the command.")
+    help_text: str = Field(..., description="Help text for the command.")
+    docstring: Docstring = Field(..., description="The docstring for the command.")
+    signature: Signature = Field(..., description="The signature of the command.")
+
+
+class App(BaseModel):
+    """Represents the entire Typer application."""
+    app_help: str = Field(..., description="Help text for the Typer application.")
+    commands: List[Command] = Field(..., description="List of commands in the application.")
+
+
+def render_option(option: Option) -> str:
+    """Renders the option template with the provided data."""
+    template = env.get_template("option.j2")
+    return template.render(ctx=option)
+
+
+def render_signature(signature: Signature) -> str:
+    """Renders the signature template using the provided data."""
+    template = env.get_template("signature.j2")
+    return template.render(ctx=signature.mod)
+
+
+def render(template_name: str, model: BaseModel) -> str:
+    """Renders the template using the provided model."""
+    template = env.get_template(template_name)
+    print(model.model_dump())
+    return template.render(**model.model_dump())
+
+
+if __name__ == "__main__":
+    # Render the template
+
+    init_instant()
+
+    sig = Signature.from_prompt("Create a function called 'review_code' with 2 arguments and 2 options. The function should perform a code review on the specified repository and branch.")
+    print(sig.to_yaml())
+
+    rendered_signature = render("signature.j2", sig)
+    print(rendered_signature)

src/pyn8n/n8n_client.py

@@ -501,6 +501,7 @@ async def shutdown(self) -> None:
 # ---------------------------
 
 async def main():
+    api_key = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiI2ZjdkNmM1Zi1iMTBhLTQwOGYtOTdjOS05ZjEzNWRmY2QxNzkiLCJpc3MiOiJuOG4iLCJhdWQiOiJwdWJsaWMtYXBpIiwiaWF0IjoxNzM2ODM5OTI5fQ.jDr4jghwLHLJAGiVVxKtsQD07R3BHpXukSLNgcaldeE"
     client = N8nClient(
         base_url="http://localhost:5678/api/v1",
         api_key=api_key,

src/pyn8n/n8n_decorator.py

@@ -34,7 +34,7 @@ def decorator(func: Callable[..., BaseModel]):
             "output_model": output_model,
         }
 
-        # print(f"Registered n8n node: {action_name}")
+        print(f"Registered n8n node: {action_name}")
 
         # Define a FastAPI endpoint for the node
         @router.post(f"/{action_name}", response_model=output_model)

src/pyn8n/n8n_nodes.py

@@ -35,3 +35,18 @@ def factorial(n: int) -> int:
     print(f"{factorial.__name__}: Factorial of {body.number} is {result}")
     return FactorialOutput(result=result)
 
+
+class TTSInput(BaseModel):
+    text: str = Field(..., description="The text to be spoken")
+
+class TTSOutput(BaseModel):
+    ...
+
+@n8n_node(input_model=TTSInput, output_model=TTSOutput)
+def text_to_speech(body: TTSInput) -> TTSOutput:
+    """Convert text to speech."""
+    import pyttsx3
+    engine = pyttsx3.init()
+    engine.say(body.text)
+    engine.runAndWait()
+    return TTSOutput()

src/pyn8n/node_renderer.py

@@ -0,0 +1,96 @@
+
+from jinja2 import Template
+
+template = Template("""
+from pydantic import BaseModel, Field
+from pyn8n.n8n_decorator import n8n_node
+
+{% for import in import_list %}
+{{ import }}
+{% endfor %}
+
+# Input model
+class {{ input_model.name }}Input(BaseModel):
+    {% for field in input_model.fields %}
+    {{ field.name }}: {{ field.type }} = Field(..., description="{{ field.description }}")
+    {% endfor %}
+
+# Output model
+class {{ output_model.name }}(BaseModel):
+    {% for field in output_model.fields %}
+    {{ field.name }}: {{ field.type }} = Field(..., description="{{ field.description }}")
+    {% endfor %}
+
+# Node definition
+@n8n_node(input_model={{ input_model.name }}Input, output_model={{ output_model.name }})
+def {{ function_name }}(body: {{ input_model.name }}Input) -> {{ output_model.name }}:
+    \"\"\"{{ function_description }}\"\"\"
+    {{ implementation }}
+    return {{ output_model.name }}({{ return_fields }})
+""")
+
+input_model = {
+    "name": "ExampleNode",
+    "fields": [
+        {"name": "field1", "type": "int", "description": "An example field"}
+    ]
+}
+
+output_model = {
+    "name": "ExampleNodeOutput",
+    "fields": [
+        {"name": "result", "type": "int", "description": "The result of the computation"}
+    ]
+}
+
+function_data = {
+    "function_name": "example_node",
+    "function_description": "An example node that performs a computation.",
+    "implementation": "result = body.field1 * 2",
+    "return_fields": "result=result"
+}
+
+class FieldTemplate(BaseModel):
+    name: str
+    type: str
+    description: str
+
+class InputModelTemplate(BaseModel):
+    name: str
+    fields: list[FieldTemplate]
+
+class OutputModelTemplate(BaseModel):
+    name: str
+    fields: list[FieldTemplate]
+
+class FunctionDataTemplate(BaseModel):
+    function_name: str
+    function_description: str
+    implementation: str
+    return_fields: str
+
+
+if __name__ == "__main__":
+    input_model = InputModelTemplate(
+        name="ExampleNode",
+        fields=[
+            FieldTemplate(name="field1", type="int", description="An example field")
+        ]
+    )
+
+    output_model = OutputModelTemplate(
+        name="ExampleNodeOutput",
+        fields=[
+            FieldTemplate(name="result", type="int", description="The result of the computation")
+        ]
+    )
+
+    function_data = FunctionDataTemplate(
+        function_name="example_node",
+        function_description="An example node that performs a computation.",
+        implementation="result = body.field1 * 2",
+        return_fields="result=result"
+    )
+
+    rendered_code = template.render(input_model=input_model, output_model=output_model, **function_data.model_dump())
+    print(rendered_code)

src/pyn8n/templates/argument.j2

@@ -0,0 +1,13 @@
+{% macro render_argument(ctx) -%}
+{{ ctx.name }}: Annotated[
+    {{ ctx.type }},
+    typer.Argument(
+        {% if ctx.help %}
+        help="{{ ctx.help }}",
+        {% endif %}
+        {% if ctx.rich_help_panel %}
+        rich_help_panel="{{ ctx.rich_help_panel }}",
+        {% endif %}
+    )
+]{% if ctx.default is not none %} = {{ ctx.default }}{% endif %}
+{%- endmacro %}

src/pyn8n/templates/option.j2

@@ -0,0 +1,19 @@
+{% macro render_option(ctx) -%}
+{{ ctx.name }}: Annotated[
+    {{ ctx.type }},
+    typer.Option(
+        {% if ctx.prompt is not none %}
+        prompt={{ "true" if ctx.prompt is true else '"' ~ ctx.prompt ~ '"' }},
+        {% endif %}
+        {% if ctx.help %}
+        help="{{ ctx.help }}",
+        {% endif %}
+        {% if ctx.rich_help_panel %}
+        rich_help_panel="{{ ctx.rich_help_panel }}",
+        {% endif %}
+        {% if ctx.show_default is not none %}
+        show_default={{ "true" if ctx.show_default else "false" }},
+        {% endif %}
+    )
+]{% if ctx.default is not none %} = {{ ctx.default }}{% endif %}
+{%- endmacro %}

src/pyn8n/templates/signature.j2

@@ -0,0 +1,15 @@
+{% from "argument.j2" import render_argument %}
+{% from "option.j2" import render_option %}
+
+def {{ function_name }}(
+    {%- for argument in arguments %}
+    {{ render_argument(argument) }}{% if not loop.last %},{% endif %}
+    {% endfor %}
+    {%- for option in options %}
+    {{ render_option(option) }}{% if not loop.last %},{% endif %}
+    {% endfor %}
+):
+    """
+    {{ docstring }}
+    """
+    pass

src/pyn8n/wip/workflow_of_thoughts.py

@@ -0,0 +1,63 @@
+
+from pydantic import BaseModel, Field
+from typing import List
+
+from dslmodel.agent_model import AgentModel
+
+
+class WorkflowStep(BaseModel):
+    """
+    Represents a single step in the workflow pattern tutorial.
+    """
+    explanation: str = Field(
+        ...,
+        description="A detailed explanation of the purpose and logic of this step."
+    )
+    output: str = Field(
+        ...,
+        description="The specific workflow action, node, or configuration generated in this step."
+    )
+    references: List[str] = Field(
+        default_factory=list,
+        description="Optional references to workflow patterns or standards for this step."
+    )
+
+class WorkflowPattern(AgentModel):
+    """
+    Represents the overall workflow pattern including the steps and final workflow.
+    """
+    name: str = Field(
+        ...,
+        description="The name of the workflow pattern (e.g., Parallel Split, Multi-Instance Processing)."
+    )
+    category: str = Field(
+        ...,
+        description="The category of the workflow pattern (e.g., Control Flow, Resource Allocation)."
+    )
+    steps: List[WorkflowStep] = Field(
+        ...,
+        description="A list of steps with explanations and outputs for building this workflow pattern."
+    )
+    use_case: str = Field(
+        ...,
+        description="A detailed expert-level use case demonstrating the practical application of this pattern."
+    )
+    final_workflow: str = Field(
+        ...,
+        description="A complete textual or JSON representation of the final workflow generated by the pattern."
+    )
+
+
+async def main():
+    wf = await WorkflowPattern.from_prompt(
+        prompt="Create a workflow pattern for parallel processing.",
+        model="groq:llama3-groq-8b-8192-tool-use-preview"
+    )
+
+    print("Workflow pattern created successfully.", wf)
+
+
+if __name__ == '__main__':
+    import asyncio
+
+    asyncio.run(main())