Integrating into Applications

The code example below demonstrates how to integrate Group Sense as an adapter between a group chat system and an existing single-user AI assistant. The pattern shows key integration steps: setting up the concurrent reasoner with custom prompts, handling incoming group messages, processing triage decisions, and feeding assistant responses back into the shared conversation context. A complete running implementation using Group Terminal as a group chat system is available at examples/chat/application.py.

Setup

# Reasoner setup
template = self._load_reasoner_template(reasoner_template_name)
self._factory = DefaultGroupReasonerFactory(system_prompt_template=template)
self._reasoner = ConcurrentGroupReasoner(factory=self._factory)

Message Handler

async def _handle_message(self, content: str, sender: str):
    message = self._create_reasoner_message(content, sender)
    # Initiate reasoner processing in message arrival order
    # (guarantees equal internal and chat message ordering)
    future = self._reasoner.process(message)
    # Asynchronously await and process reasoner response
    create_task(self._handle_response(future, sender))

async def _handle_response(self, future: Future[Response], sender: str):
    try:
        reasoner_response = await future
    except Exception:
        logger.exception("Reasoner error")
        return

    logger.debug(f"Reasoner decision: {reasoner_response.decision.value}")
    if reasoner_response.decision == Decision.IGNORE:
        return

    if not reasoner_response.query:
        logger.warning("Reasoner delegated without query")
        return

    try:
        # Run downstream assistant with query generated by reasoner
        logger.debug(f"Assistant query: {reasoner_response.query}")
        assistant_response = await self._service.run(reasoner_response.query, sender=sender)
    except Exception:
        logger.exception("Assistant error")
    else:
        logger.debug(f"Assistant response: {assistant_response}")

        if reasoner_response.receiver:
            # If reasoner set a dynamic receiver, @mention them in the chat message
            assistant_response = f"@{reasoner_response.receiver} {assistant_response}"

        message = Message(
            content=assistant_response,
            sender="system",
            receiver=reasoner_response.receiver,
        )

        # Add response message to reasoner
        # (needed for concurrent reasoning)
        self._reasoner.append(message)

        # Send response to chat clients
        await self._server.send_message(message.content, sender=message.sender)