Files
wehub-resource-sync 97e91a83f3
Ruff / Ruff (push) Waiting to run
Test / Core Tests (push) Waiting to run
Test / Offline Coverage Tests (Python 3.10) (push) Waiting to run
Test / Offline Coverage Tests (Python 3.11) (push) Waiting to run
Test / Offline Coverage Tests (Python 3.12) (push) Waiting to run
Test / Offline Coverage Tests (Python 3.13) (push) Waiting to run
Test / Offline Coverage Tests (Python 3.9) (push) Waiting to run
Test / Full Coverage (Python 3.11) (push) Waiting to run
Test / Core Provider Tests (OpenAI) (push) Blocked by required conditions
Test / Core Provider Tests (Anthropic) (push) Blocked by required conditions
Test / Core Provider Tests (Google) (push) Blocked by required conditions
Test / Core Provider Tests (Other) (push) Blocked by required conditions
Test / Anthropic Tests (push) Blocked by required conditions
Test / Gemini Tests (push) Blocked by required conditions
Test / Google GenAI Tests (push) Blocked by required conditions
Test / Vertex AI Tests (push) Blocked by required conditions
Test / OpenAI Tests (push) Blocked by required conditions
Test / Writer Tests (push) Blocked by required conditions
Test / Auto Client Tests (push) Blocked by required conditions
ty / type-check (push) Waiting to run
chore: import upstream snapshot with attribution
2026-07-13 13:36:38 +08:00

88 lines
2.3 KiB
Python

from enum import Enum
from pydantic import BaseModel, Field
import instructor
from openai import OpenAI
client = instructor.from_openai(OpenAI())
class CRMSource(Enum):
personal = "personal"
business = "business"
work_contacts = "work_contacts"
all = "all"
class CRMSearch(BaseModel):
"""A CRM search query
The search description is a natural language description of the search query
the backend will use semantic search so use a range of phrases to describe the search
"""
source: CRMSource
city_location: str = Field(
..., description="City location used to match the desired customer profile"
)
search_description: str = Field(
..., description="Search query used to match the desired customer profile"
)
class CRMSearchQuery(BaseModel):
"""
A set of CRM queries to be executed against a CRM system,
for large locations decompose into multiple queries of smaller locations
"""
queries: list[CRMSearch]
def query_crm(query: str) -> CRMSearchQuery:
queries = client.chat.completions.create(
model="gpt-3.5-turbo",
response_model=CRMSearchQuery,
messages=[
{
"role": "system",
"content": """
You are a world class CRM search career generator.
You will take the user query and decompose it into a set of CRM queries queries.
""",
},
{"role": "user", "content": query},
],
)
return queries
if __name__ == "__main__":
query = "find me all the pottery businesses in San Francisco and my friends in the east coast big cities"
print(query_crm(query).model_dump_json(indent=2))
"""
{
"queries": [
{
"source": "business",
"city_location": "San Francisco",
"search_description": "pottery businesses"
},
{
"source": "personal",
"city_location": "New York",
"search_description": "friends in New York"
},
{
"source": "personal",
"city_location": "Boston",
"search_description": "friends in Boston"
},
{
"source": "personal",
"city_location": "Philadelphia",
"search_description": "friends in Philadelphia"
}
]
}
"""