Skip to content

Commit dd8dd33

Browse files
authored
Merge pull request #42 from AI-agent-assembly/v0.0.1/AAASM-990/feat/edge_emission_hooks
[AAASM-990] ✨ (adapters): Add edge emission hooks for LangGraph and OpenAI Agents
2 parents 059d916 + 4e6f27b commit dd8dd33

7 files changed

Lines changed: 395 additions & 13 deletions

File tree

agent_assembly/adapters/langgraph/patch.py

Lines changed: 30 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,8 @@
55
import contextlib
66
import importlib
77
import inspect
8-
from dataclasses import dataclass
8+
import threading
9+
from dataclasses import dataclass, field
910
from typing import Any
1011

1112
from agent_assembly.core.spawn import _SPAWN_CTX, SpawnContext, spawn_context_scope
@@ -15,16 +16,32 @@
1516
_NODE_WRAPPED_FLAG = "_agent_assembly_node_wrapped"
1617
_INVOKE_WRAPPED_FLAG = "_agent_assembly_invoke_wrapped"
1718

19+
# Thread-local storage for the name of the most-recently-completed node so that
20+
# the next _record_node_enter can emit a directed Messages edge between them.
21+
_NODE_TRANSITION: threading.local = threading.local()
22+
23+
# Module-level edge emitter instance; set via set_edge_emitter().
24+
_EDGE_EMITTER: Any = None
25+
26+
27+
def set_edge_emitter(emitter: Any) -> None:
28+
"""Register the EdgeEmitter used for fire-and-forget topology edge reporting."""
29+
global _EDGE_EMITTER
30+
_EDGE_EMITTER = emitter
31+
1832

1933
@dataclass(slots=True)
2034
class LangGraphPatch:
2135
"""Applies LangGraph runtime monkey-patching for node-level governance hooks."""
2236

2337
callback_handler: Any
2438
process_agent_id: str | None = None
39+
edge_emitter: Any = field(default=None)
2540

2641
def apply(self) -> bool:
2742
"""Apply patching once and return whether patch wiring is active."""
43+
if self.edge_emitter is not None:
44+
set_edge_emitter(self.edge_emitter)
2845
state_graph_cls = _load_stategraph_class()
2946
if state_graph_cls is None:
3047
return False
@@ -450,6 +467,15 @@ def wrapped_sync_invoke(*invoke_args: Any, **invoke_kwargs: Any) -> Any:
450467

451468

452469
def _record_node_enter(callback_handler: Any, *, node_name: str, state: object, config: object) -> None:
470+
# Emit a Messages edge when a previous node has been recorded on this thread.
471+
prev_name: str | None = getattr(_NODE_TRANSITION, "name", None)
472+
if prev_name is not None and _EDGE_EMITTER is not None:
473+
emit = getattr(_EDGE_EMITTER, "emit", None)
474+
if callable(emit):
475+
transition_input_keys = list(state.keys()) if isinstance(state, dict) else []
476+
emit(prev_name, node_name, "messages", {"transition_input_keys": transition_input_keys})
477+
_NODE_TRANSITION.name = None
478+
453479
method = getattr(callback_handler, "on_graph_node_start", None)
454480
if not callable(method):
455481
return None
@@ -478,6 +504,9 @@ def _record_node_exit(
478504
next_state: object,
479505
config: object,
480506
) -> None:
507+
# Record which node just finished so the next enter can emit a directed edge.
508+
_NODE_TRANSITION.name = node_name
509+
481510
method = getattr(callback_handler, "on_graph_node_end", None)
482511
if not callable(method):
483512
return None

agent_assembly/adapters/openai_agents/patch.py

Lines changed: 22 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
import importlib
66
import importlib.util
77
import inspect
8-
from dataclasses import dataclass
8+
from dataclasses import dataclass, field
99
from functools import wraps
1010
from typing import Any, Literal
1111

@@ -24,19 +24,29 @@
2424
_ORIGINAL_HANDOFF_CALL = "_agent_assembly_original_openai_agents_handoff_call"
2525
_HANDOFF_PATCHED_FLAG = "_agent_assembly_openai_agents_handoff_patched"
2626
_PROCESS_AGENT_ID: str | None = None
27+
_EDGE_EMITTER: Any = None
2728
_MAX_AUDIT_RESULT_CHARS = 2000
2829
_MAX_DELEGATION_REASON_CHARS = 256
2930

3031

32+
def set_edge_emitter(emitter: Any) -> None:
33+
"""Register the EdgeEmitter used for fire-and-forget topology edge reporting."""
34+
global _EDGE_EMITTER
35+
_EDGE_EMITTER = emitter
36+
37+
3138
@dataclass(slots=True)
3239
class OpenAIAgentsPatch:
3340
"""Patch placeholder for OpenAI Agents SDK interception."""
3441

3542
callback_handler: Any
3643
process_agent_id: str | None = None
44+
edge_emitter: Any = field(default=None)
3745

3846
def apply(self) -> bool:
3947
set_process_agent_id(self.process_agent_id)
48+
if self.edge_emitter is not None:
49+
set_edge_emitter(self.edge_emitter)
4050
function_tool_cls = _load_openai_agents_function_tool_class()
4151
if function_tool_cls is None:
4252
return False
@@ -139,8 +149,17 @@ async def patched_call(self: Any, *args: Any, **kwargs: Any) -> Any:
139149
with spawn_context_scope(spawn_ctx):
140150
result = original_call(self, *args, **kwargs)
141151
if inspect.isawaitable(result):
142-
return await result
143-
return result
152+
result = await result
153+
154+
# Emit a DelegatesTo edge from the delegating agent to the handoff target.
155+
if _EDGE_EMITTER is not None and process_agent_id:
156+
target_id = getattr(self, "agent_name", None) or getattr(self, "name", None) or "unknown"
157+
emit = getattr(_EDGE_EMITTER, "emit", None)
158+
if callable(emit):
159+
reason = _extract_handoff_delegation_reason(self)
160+
emit(process_agent_id, str(target_id), "delegates_to", {"reason": reason})
161+
162+
return result
144163

145164
setattr(handoff_cls, _ORIGINAL_HANDOFF_CALL, original_call)
146165
handoff_cls.__call__ = patched_call

agent_assembly/client/emitter.py

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
"""Fire-and-forget edge emission helper."""
2+
3+
from __future__ import annotations
4+
5+
import logging
6+
import threading
7+
8+
from agent_assembly.client.gateway import GatewayClient
9+
10+
logger = logging.getLogger(__name__)
11+
12+
13+
class EdgeEmitter:
14+
"""Emits topology edges asynchronously without blocking the caller."""
15+
16+
def __init__(self, client: GatewayClient) -> None:
17+
self._client = client
18+
19+
def emit(
20+
self,
21+
source_agent_id: str,
22+
target_agent_id: str,
23+
edge_type: str,
24+
metadata: dict | None = None,
25+
) -> None:
26+
"""Schedule a fire-and-forget edge report on a daemon thread."""
27+
t = threading.Thread(
28+
target=self._send,
29+
args=(source_agent_id, target_agent_id, edge_type, metadata),
30+
daemon=True,
31+
)
32+
t.start()
33+
34+
def _send(
35+
self,
36+
source_agent_id: str,
37+
target_agent_id: str,
38+
edge_type: str,
39+
metadata: dict | None,
40+
) -> None:
41+
try:
42+
self._client.report_edge(source_agent_id, target_agent_id, edge_type, metadata)
43+
except Exception:
44+
logger.warning(
45+
"EdgeEmitter: failed to report edge %s -> %s (%s)",
46+
source_agent_id,
47+
target_agent_id,
48+
edge_type,
49+
exc_info=True,
50+
)

agent_assembly/client/gateway.py

Lines changed: 45 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@
22

33
from __future__ import annotations
44

5-
from typing import Optional
6-
75
import httpx
86

97
from agent_assembly.exceptions import GatewayError
@@ -16,14 +14,14 @@ def __init__(
1614
self,
1715
gateway_url: str,
1816
agent_id: str,
19-
api_key: Optional[str] = None,
17+
api_key: str | None = None,
2018
timeout: int = 30,
2119
*,
22-
parent_agent_id: Optional[str] = None,
23-
team_id: Optional[str] = None,
24-
delegation_reason: Optional[str] = None,
25-
spawned_by_tool: Optional[str] = None,
26-
depth: Optional[int] = None,
20+
parent_agent_id: str | None = None,
21+
team_id: str | None = None,
22+
delegation_reason: str | None = None,
23+
spawned_by_tool: str | None = None,
24+
depth: int | None = None,
2725
) -> None:
2826
"""
2927
Initialize the GatewayClient.
@@ -48,7 +46,7 @@ def __init__(
4846
self.delegation_reason = delegation_reason
4947
self.spawned_by_tool = spawned_by_tool
5048
self.depth = depth
51-
self._client: Optional[httpx.Client] = None
49+
self._client: httpx.Client | None = None
5250

5351
@property
5452
def client(self) -> httpx.Client:
@@ -131,3 +129,41 @@ async def check_policy_compliance(self, action: str) -> dict:
131129
return response.json()
132130
except httpx.HTTPError as e:
133131
raise GatewayError(f"Failed to check policy compliance: {e}") from e
132+
133+
def report_edge(
134+
self,
135+
source_agent_id: str,
136+
target_agent_id: str,
137+
edge_type: str,
138+
metadata: dict | None = None,
139+
) -> dict:
140+
"""
141+
Report a directed edge between two agents to the topology store.
142+
143+
Args:
144+
source_agent_id: ID of the source agent
145+
target_agent_id: ID of the target agent
146+
edge_type: Semantic type of the edge (e.g. "messages", "delegates_to")
147+
metadata: Optional freeform metadata to attach to the edge
148+
149+
Returns:
150+
Response data containing the assigned edge id
151+
152+
Raises:
153+
GatewayError: If the request fails
154+
"""
155+
import json as _json
156+
157+
body: dict = {
158+
"source_agent_id": source_agent_id,
159+
"target_agent_id": target_agent_id,
160+
"edge_type": edge_type,
161+
}
162+
if metadata is not None:
163+
body["metadata_json"] = _json.dumps(metadata)
164+
try:
165+
response = self.client.post("/topology/edges", json=body)
166+
response.raise_for_status()
167+
return response.json()
168+
except httpx.HTTPError as e:
169+
raise GatewayError(f"Failed to report edge: {e}") from e

test/unit/adapters/langgraph/__init__.py

Whitespace-only changes.
Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,137 @@
1+
"""Tests for LangGraph node-to-node Messages edge emission."""
2+
3+
from __future__ import annotations
4+
5+
from typing import Any
6+
7+
from agent_assembly.adapters.langgraph import patch as lg_patch
8+
9+
10+
class RecordingEdgeEmitter:
11+
"""Synchronous test double that records emitted edges."""
12+
13+
def __init__(self) -> None:
14+
self.edges: list[tuple[str, str, str, dict | None]] = []
15+
16+
def emit(self, source: str, target: str, edge_type: str, metadata: dict | None = None) -> None:
17+
self.edges.append((source, target, edge_type, metadata))
18+
19+
20+
class _NullCallbackHandler:
21+
"""Minimal callback handler that satisfies the patch interface."""
22+
23+
def on_graph_node_start(self, **_kwargs: Any) -> None:
24+
pass
25+
26+
def on_graph_node_end(self, **_kwargs: Any) -> None:
27+
pass
28+
29+
30+
def _make_node_map(*node_names: str, handler: Any) -> dict[str, Any]:
31+
"""Build a fake compiled-graph node map with simple callables."""
32+
node_map: dict[str, Any] = {}
33+
for name in node_names:
34+
def make_func(n: str) -> Any:
35+
def node_fn(state: Any) -> dict[str, Any]:
36+
return {"node": n, **state}
37+
38+
return node_fn
39+
40+
node_map[name] = make_func(name)
41+
return node_map
42+
43+
44+
def _run_two_node_graph(
45+
node_a: str,
46+
node_b: str,
47+
emitter: RecordingEdgeEmitter,
48+
initial_state: dict[str, Any] | None = None,
49+
) -> None:
50+
"""Simulate a 2-node sequential graph execution with patched wrappers."""
51+
handler = _NullCallbackHandler()
52+
node_map = _make_node_map(node_a, node_b, handler=handler)
53+
54+
lg_patch.set_edge_emitter(emitter)
55+
try:
56+
lg_patch._wrap_node_map(node_map, handler)
57+
58+
state: dict[str, Any] = initial_state or {}
59+
state = node_map[node_a](state)
60+
state = node_map[node_b](state)
61+
finally:
62+
lg_patch.set_edge_emitter(None)
63+
lg_patch._NODE_TRANSITION.name = None
64+
65+
66+
def test_messages_edge_emitted_between_two_sequential_nodes() -> None:
67+
emitter = RecordingEdgeEmitter()
68+
_run_two_node_graph("node_a", "node_b", emitter)
69+
70+
assert len(emitter.edges) == 1
71+
src, tgt, etype, meta = emitter.edges[0]
72+
assert src == "node_a"
73+
assert tgt == "node_b"
74+
assert etype == "messages"
75+
assert isinstance(meta, dict)
76+
assert "transition_input_keys" in meta
77+
78+
79+
def test_transition_input_keys_contains_state_keys() -> None:
80+
emitter = RecordingEdgeEmitter()
81+
_run_two_node_graph("node_a", "node_b", emitter, initial_state={"msg": "hi", "count": 0})
82+
83+
assert len(emitter.edges) == 1
84+
meta = emitter.edges[0][3]
85+
assert meta is not None
86+
# transition_input_keys reflects the state keys seen when node_a completed
87+
assert set(meta["transition_input_keys"]) >= {"msg", "count"}
88+
89+
90+
def test_no_edge_emitted_for_first_node_in_graph() -> None:
91+
"""The very first node has no predecessor so no edge should be emitted."""
92+
emitter = RecordingEdgeEmitter()
93+
lg_patch.set_edge_emitter(emitter)
94+
try:
95+
handler = _NullCallbackHandler()
96+
node_map = _make_node_map("only_node", handler=handler)
97+
lg_patch._wrap_node_map(node_map, handler)
98+
node_map["only_node"]({})
99+
finally:
100+
lg_patch.set_edge_emitter(None)
101+
lg_patch._NODE_TRANSITION.name = None
102+
103+
assert emitter.edges == []
104+
105+
106+
def test_three_node_graph_emits_two_edges() -> None:
107+
emitter = RecordingEdgeEmitter()
108+
lg_patch.set_edge_emitter(emitter)
109+
try:
110+
handler = _NullCallbackHandler()
111+
node_map = _make_node_map("a", "b", "c", handler=handler)
112+
lg_patch._wrap_node_map(node_map, handler)
113+
state: dict[str, Any] = {}
114+
state = node_map["a"](state)
115+
state = node_map["b"](state)
116+
node_map["c"](state)
117+
finally:
118+
lg_patch.set_edge_emitter(None)
119+
lg_patch._NODE_TRANSITION.name = None
120+
121+
assert len(emitter.edges) == 2
122+
assert emitter.edges[0][:3] == ("a", "b", "messages")
123+
assert emitter.edges[1][:3] == ("b", "c", "messages")
124+
125+
126+
def test_no_edge_emitted_when_emitter_is_none() -> None:
127+
"""When no emitter is registered, transitions must not raise."""
128+
lg_patch.set_edge_emitter(None)
129+
try:
130+
handler = _NullCallbackHandler()
131+
node_map = _make_node_map("x", "y", handler=handler)
132+
lg_patch._wrap_node_map(node_map, handler)
133+
state: dict[str, Any] = {}
134+
state = node_map["x"](state)
135+
node_map["y"](state) # Should not raise
136+
finally:
137+
lg_patch._NODE_TRANSITION.name = None

0 commit comments

Comments
 (0)