Why We Built a Typeform Knowledge Graph Connector for Cognee
If you’ve ever tried piping survey responses into an LLM application or RAG pipeline, you know the pain.
Forms don't output clean prose. They spit out disjointed JSON arrays: arbitrary field IDs, cryptic choice indexes, nested metadata, and missing answers. Dump that raw tabular data into standard vector search, and your similarity lookups fall apart the moment someone asks:
"What did enterprise leads in Europe say about our self-hosted pricing tier?"
To solve this, we built and contributed the official Typeform Connector for Cognee (cognee-community-connector-typeform). Here is how we designed it, how Cognee's document model works under the hood, and how you can use it in your own stacks.
The Problem with Naive Survey Ingestion
When ingesting form telemetry into a Knowledge Graph, two approaches usually fail:
-
Flat Relational Dumps: Flattening every submission into rows strips away semantic context. The LLM has to guess what
choice_id_892actually meant in relation to the respondent's company size. -
Raw JSON String Dumps: Embedding raw JSON blobs wastes token budget on structural boilerplate (
{"type": "multiple_choice", "field": {"ref": "..."}}) rather than the respondent's actual voice.
The Solution: Form-Aware Prose Serialization
Our connector hooks into Typeform's Forms and Responses APIs via a resilient dlt (data load tool) pipeline. It resolves form definition schemas, maps each question title directly to its submitted answer, and constructs a structured markdown document for every response:
flowchart LR
TF[Typeform API] -->|Form Schema + Responses| Client[TypeformClient]
Client -->|Resolve Question Titles & Answers| DLT[DLT Resource]
DLT -->|DOCUMENT_SOURCE_ATTR='typeform_responses'| Cognee[Cognee Cognify]
Cognee -->|Extract Entities & Relationships| Graph[(Knowledge Graph + Vector Store)]
How It Works Under the Hood
1. The Schema-to-Answer Resolver
Typeform splits form definitions (/forms/{id}) from respondent answers (/forms/{id}/responses). The connector caches question titles and maps varied response types (choices, text, number, boolean, payment):
def _extract_answer_value(ans: dict[str, Any]) -> str:
ans_type = ans.get("type", "")
if ans_type == "text":
return ans.get("text", "")
if ans_type == "choice":
return ans.get("choice", {}).get("label", "")
if ans_type == "choices":
labels = ans.get("choices", {}).get("labels", [])
return ", ".join(labels)
if ans_type == "boolean":
return "Yes" if ans.get("boolean") else "No"
if ans_type == "number":
return str(ans.get("number", ""))
return str(ans.get(ans_type, ""))
2. The Cognee Document Contract
Cognee's cognify() engine needs to know when incoming records should undergo entity and relation extraction rather than flat tabular storage. We register this via DOCUMENT_SOURCE_ATTR:
DOCUMENT_SOURCE_ATTR = "typeform_responses"
setattr(typeform_responses_resource, DOCUMENT_SOURCE_ATTR, True)
Each submission is transformed into clean prose:
# Typeform Response: Customer Feedback Survey (ID: resp_8821)
- **Submitted At:** 2026-10-04T16:00:00Z
- **Response ID:** resp_8821
### Survey Responses
- **What is your company size?** 50-200 employees
- **Which feature is most critical for your team?** On-premise vector index synchronization
- **Any additional feedback?** The current REST API latency is solid, but we need better RBAC controls.
When Cognee processes this, its GLiNER and LLM extractors recognize companies, features, feedback sentiments, and create connected nodes in your graph database (Kuzu, Neo4j, or Postgres).
Quickstart: Adding Typeform to Your Cognee Pipeline
1. Installation
pip install cognee-community-connector-typeform
2. Ingestion in 4 Lines of Python
import asyncio
import cognee
from cognee_community_connector_typeform import typeform_source
async def main():
# Configure connector
source = typeform_source(
token="YOUR_TYPEFORM_PERSONAL_ACCESS_TOKEN",
form_ids=["abc123xyz"], # Or omit to sync all forms
since="2026-09-01T00:00:00Z",
)
# Ingest and Cognify
await cognee.add(source, dataset_name="survey_insights")
await cognee.cognify(dataset_name="survey_insights")
# Ask multi-hop questions over survey feedback
insights = await cognee.search(
"What are mid-sized companies saying about RBAC requirements?",
dataset_name="survey_insights"
)
print(insights)
if __name__ == "__main__":
asyncio.run(main())
Resilient Offline Testing
Because CI runners shouldn't rely on live API keys, we wrote a complete mock-tested suite covering:
- Handling paginated responses and token auth headers.
- Graceful fallbacks when questions have no answers.
-
sinceISO timestamp filtering.
Run it locally anytime:
uv run pytest packages/connector/typeform/tests/test_typeform.py -v
Check It Out
Have ideas for more data connectors or graph models? Jump into Cognee's Discord and let's build!
Top comments (0)