Skip to content

ConversationAdapter

autogen.beta.network.adapters.conversation.ConversationAdapter #

ConversationAdapter()

Bidirectional multi-turn 1+1 session.

Knobs: none. Participants: exactly 2 (initiator + respondent). Default view: :class:WindowedSummary(recent_n=10) — bounded prompt size at any turn count.

Source code in autogen/beta/network/adapters/conversation.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def __init__(self) -> None:
    self.manifest = SessionManifest(
        type=CONVERSATION_TYPE,
        version=1,
        participants=ParticipantSchema(
            min=2,
            max=2,
            roles=[ParticipantRole.INITIATOR.value, ParticipantRole.RESPONDENT.value],
        ),
        knobs_schema={},
        default_view_policy=WindowedSummary.name,
        expectations=[
            Expectation(
                name="max_silence",
                on_violation="audit",
                params={"seconds": 3600},
            ),
        ],
    )

manifest instance-attribute #

manifest = SessionManifest(type=CONVERSATION_TYPE, version=1, participants=ParticipantSchema(min=2, max=2, roles=[INITIATOR.value, RESPONDENT.value]), knobs_schema={}, default_view_policy=name, expectations=[Expectation(name='max_silence', on_violation='audit', params={'seconds': 3600})])

initial_state #

initial_state(metadata)
Source code in autogen/beta/network/adapters/conversation.py 
119
120
def initial_state(self, metadata: SessionMetadata) -> ConversationState:
    return ConversationState()

fold #

fold(envelope, state)
Source code in autogen/beta/network/adapters/conversation.py 
122
123
124
125
126
127
128
129
130
131
def fold(self, envelope: Envelope, state: ConversationState) -> ConversationState:
    if _is_session_protocol_event(envelope) or _is_task_event(envelope):
        return state
    if envelope.event_type != EV_TEXT:
        return state
    return ConversationState(
        turn_count=state.turn_count + 1,
        last_speaker_id=envelope.sender_id,
        last_envelope_id=envelope.envelope_id,
    )

validate_create #

validate_create(metadata)
Source code in autogen/beta/network/adapters/conversation.py 
133
134
135
136
137
138
139
140
def validate_create(self, metadata: SessionMetadata) -> None:
    roles = {p.role for p in metadata.participants}
    if ParticipantRole.INITIATOR not in roles:
        raise ProtocolError("conversation requires exactly one initiator")
    if ParticipantRole.RESPONDENT not in roles:
        raise ProtocolError("conversation requires exactly one respondent")
    if len(metadata.participants) != 2:
        raise ProtocolError(f"conversation requires exactly 2 participants, got {len(metadata.participants)}")

validate_send #

validate_send(metadata, envelope, state)
Source code in autogen/beta/network/adapters/conversation.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def validate_send(
    self,
    metadata: SessionMetadata,
    envelope: Envelope,
    state: ConversationState,
) -> None:
    if _is_session_protocol_event(envelope) or _is_task_event(envelope):
        return
    if envelope.event_type != EV_TEXT:
        # Unknown event types accepted as informational data — same
        # convention as consulting.
        return
    participant_ids = {p.agent_id for p in metadata.participants}
    if envelope.sender_id not in participant_ids:
        raise ProtocolError(
            f"conversation session {metadata.session_id!r} only accepts "
            f"sends from participants, got {envelope.sender_id!r}"
        )

on_accepted #

on_accepted(metadata, envelope, state)
Source code in autogen/beta/network/adapters/conversation.py 
161
162
163
164
165
166
167
168
169
def on_accepted(
    self,
    metadata: SessionMetadata,
    envelope: Envelope,
    state: ConversationState,
) -> AdapterResult:
    # Conversations end via explicit ``Hub.close_session`` or TTL
    # — never via adapter-initiated transitions on accepted content.
    return AdapterResult()

default_view_policy #

default_view_policy(metadata, participant_id)
Source code in autogen/beta/network/adapters/conversation.py 
171
172
173
174
175
176
def default_view_policy(
    self,
    metadata: SessionMetadata,
    participant_id: str,
) -> ViewPolicy:
    return WindowedSummary(recent_n=_DEFAULT_RECENT_N)

extract_turn_input #

extract_turn_input(envelope)
Source code in autogen/beta/network/adapters/conversation.py 
178
179
def extract_turn_input(self, envelope):
    return default_extract_turn_input(envelope)

build_round_envelope #

build_round_envelope(metadata, sender_id, reply, events, state, hub)
Source code in autogen/beta/network/adapters/conversation.py 
181
182
def build_round_envelope(self, metadata, sender_id, reply, events, state, hub):
    return default_build_round_envelope(metadata, sender_id, reply, events, state, hub)

render_envelope #

render_envelope(envelope)
Source code in autogen/beta/network/adapters/conversation.py 
184
185
def render_envelope(self, envelope):
    return default_render_envelope(envelope)