Build Fastapi Apps with Instructor
Accelerate Fastapi development with Instructor, enabling structured data handling and streamlined API creation for robust application building.
Why it matters
Streamline your application development by leveraging Instructor to build robust and efficient Fastapi applications. This asset provides a foundation for creating complex APIs with structured outputs.
Outcomes
What it gets done
Generate boilerplate code for Fastapi applications.
Integrate Instructor for structured data handling within Fastapi.
Facilitate debugging and testing of API endpoints.
Prepare applications for deployment via CI/CD pipelines.
Install
Add it to your toolbox
Run in your project directory:
curl -fsSL https://spark.entire.vc/get/inst-fastapiapp | bash Capabilities
What this skill does
Writes source code or scripts from a description.
Traces errors to their root cause and suggests fixes.
Runs build pipelines, tests, and deploys to environments.
Overview
Fastapi App
What it does
--- main.py ---
from fastapi import FastAPI
from instructor import ResponseSchema
import instructor.dsl as dsl
from pydantic import BaseModel, Field
app = FastAPI(title="Example Application using instructor")
class SearchRequest(BaseModel):
body: str
class SearchQuery(ResponseSchema):
title: str = Field(..., description="Question that the query answers")
query: str = Field(
...,
description="Detailed, comprehensive, and specific query to be used for semantic search",
)
SearchResponse = dsl.MultiTask(
subtask_class=SearchQuery,
description="Correctly segmented set of search queries",
)
@app.post("/search", response_model=SearchResponse)
async def search(request: SearchRequest):
task = (
dsl.ChatCompletion(name="Segmenting Search requests example")
| dsl.SystemTask(task="Segment search results")
| dsl.TaggedMessage(content=request.body, tag="query")
| dsl.TipsMessage(
tips=[
"Expand query to contain multiple forms of the same word (SSO -> Single Sign On)",
"Use the title to explain what the query should return, but use the query to complete the search",
"The query should be detailed, specific, and cast a wide net when possible",
]
)
| SearchRequest
)
return await task.acreate()
--- script.py ---
from instructor import ResponseSchema, dsl
from pydantic import Field
import json
class SearchQuery(ResponseSchema):
query: str = Field(
...,
description="Detailed, comprehensive, and specific query to be used for semantic search",
)
SearchResponse = dsl.MultiTask(
subtask_class=SearchQuery,
description="Correctly segmented set of search queries",
)
task = (
dsl.ChatCompletion(name="Segmenting Search requests example")
| dsl.SystemTask(task="Segment search results")
| dsl.TaggedMessage(
content="can you send me the data about the video investment and the one about spot the dog?",
tag="query",
)
| dsl.TipsMessage(
tips=[
"Expand query to contain multiple forms of the same word (SSO -> Single Sign On)",
"Use the title to explain what the query should return, but use the query to complete the search",
"The query should be detailed, specific, and cast a wide net when possible",
]
)
| SearchResponse
)
print(json.dumps(task.kwargs, indent=1))
"""
{
"tasks": [
{
"query": "data about video investment"
},
{
"query": "data about spot the dog"
}
]
}
"""
Source README
--- main.py ---
from fastapi import FastAPI
from instructor import ResponseSchema
import instructor.dsl as dsl
from pydantic import BaseModel, Field
app = FastAPI(title="Example Application using instructor")
class SearchRequest(BaseModel):
body: str
class SearchQuery(ResponseSchema):
title: str = Field(..., description="Question that the query answers")
query: str = Field(
...,
description="Detailed, comprehensive, and specific query to be used for semantic search",
)
SearchResponse = dsl.MultiTask(
subtask_class=SearchQuery,
description="Correctly segmented set of search queries",
)
@app.post("/search", response_model=SearchResponse)
async def search(request: SearchRequest):
task = (
dsl.ChatCompletion(name="Segmenting Search requests example")
| dsl.SystemTask(task="Segment search results")
| dsl.TaggedMessage(content=request.body, tag="query")
| dsl.TipsMessage(
tips=[
"Expand query to contain multiple forms of the same word (SSO -> Single Sign On)",
"Use the title to explain what the query should return, but use the query to complete the search",
"The query should be detailed, specific, and cast a wide net when possible",
]
)
| SearchRequest
)
return await task.acreate()
--- script.py ---
from instructor import ResponseSchema, dsl
from pydantic import Field
import json
class SearchQuery(ResponseSchema):
query: str = Field(
...,
description="Detailed, comprehensive, and specific query to be used for semantic search",
)
SearchResponse = dsl.MultiTask(
subtask_class=SearchQuery,
description="Correctly segmented set of search queries",
)
task = (
dsl.ChatCompletion(name="Segmenting Search requests example")
| dsl.SystemTask(task="Segment search results")
| dsl.TaggedMessage(
content="can you send me the data about the video investment and the one about spot the dog?",
tag="query",
)
| dsl.TipsMessage(
tips=[
"Expand query to contain multiple forms of the same word (SSO -> Single Sign On)",
"Use the title to explain what the query should return, but use the query to complete the search",
"The query should be detailed, specific, and cast a wide net when possible",
]
)
| SearchResponse
)
print(json.dumps(task.kwargs, indent=1))
"""
{
"tasks": [
{
"query": "data about video investment"
},
{
"query": "data about spot the dog"
}
]
}
"""
Discussion
Questions & comments · 0
Sign In Sign in to leave a comment.