mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-09-25 19:15:11 +02:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5d6e7f2161 |
@@ -173,8 +173,18 @@ class RunModule:
|
|||||||
config: dict[str, Any] | None = None,
|
config: dict[str, Any] | None = None,
|
||||||
metadata: dict[str, Any] | None = None,
|
metadata: dict[str, Any] | None = None,
|
||||||
langsmith_tracing: LangSmithTracing | None = None,
|
langsmith_tracing: LangSmithTracing | None = None,
|
||||||
|
context: dict[str, Any] | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""Send `run.start` to the server. Returns the result (`{"run_id": ...}`)."""
|
"""Send `run.start` to the server. Returns the result (`{"run_id": ...}`).
|
||||||
|
|
||||||
|
Args:
|
||||||
|
input: the run input; omitted from the wire payload when None.
|
||||||
|
config: the run config; omitted when None.
|
||||||
|
metadata: run metadata; omitted when None.
|
||||||
|
langsmith_tracing: tracing options; omitted when None.
|
||||||
|
context: per-run static context; omitted from the wire payload
|
||||||
|
when None (server applies its default context behavior).
|
||||||
|
"""
|
||||||
params: dict[str, Any] = {"assistant_id": self._owner.assistant_id}
|
params: dict[str, Any] = {"assistant_id": self._owner.assistant_id}
|
||||||
if input is not None:
|
if input is not None:
|
||||||
params["input"] = input
|
params["input"] = input
|
||||||
@@ -184,6 +194,8 @@ class RunModule:
|
|||||||
params["metadata"] = metadata
|
params["metadata"] = metadata
|
||||||
if langsmith_tracing is not None:
|
if langsmith_tracing is not None:
|
||||||
params["langsmith_tracer"] = langsmith_tracing
|
params["langsmith_tracer"] = langsmith_tracing
|
||||||
|
if context is not None:
|
||||||
|
params["context"] = context
|
||||||
loop = asyncio.get_running_loop()
|
loop = asyncio.get_running_loop()
|
||||||
gate: asyncio.Future[None] = loop.create_future()
|
gate: asyncio.Future[None] = loop.create_future()
|
||||||
self._owner._run_start_ready = gate
|
self._owner._run_start_ready = gate
|
||||||
@@ -216,6 +228,7 @@ class RunModule:
|
|||||||
response: Any,
|
response: Any,
|
||||||
*,
|
*,
|
||||||
interrupt_id: str | None = None,
|
interrupt_id: str | None = None,
|
||||||
|
context: dict[str, Any] | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""Reply to a server-side interrupt and resume the run.
|
"""Reply to a server-side interrupt and resume the run.
|
||||||
|
|
||||||
@@ -224,6 +237,8 @@ class RunModule:
|
|||||||
wire (protocol field name).
|
wire (protocol field name).
|
||||||
interrupt_id: optional explicit id. When omitted, requires exactly
|
interrupt_id: optional explicit id. When omitted, requires exactly
|
||||||
one outstanding interrupt and uses its id.
|
one outstanding interrupt and uses its id.
|
||||||
|
context: optional per-run static context for the resumed run;
|
||||||
|
forwarded with the `input.respond` command when non-None.
|
||||||
|
|
||||||
Raises:
|
Raises:
|
||||||
RuntimeError: no outstanding interrupts; `interrupt_id` is None but
|
RuntimeError: no outstanding interrupts; `interrupt_id` is None but
|
||||||
@@ -266,6 +281,8 @@ class RunModule:
|
|||||||
"namespace": match["namespace"],
|
"namespace": match["namespace"],
|
||||||
"response": response,
|
"response": response,
|
||||||
}
|
}
|
||||||
|
if context is not None:
|
||||||
|
params["context"] = context
|
||||||
return await self._owner._send_command("input.respond", params)
|
return await self._owner._send_command("input.respond", params)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -216,8 +216,18 @@ class SyncRunModule:
|
|||||||
config: dict[str, Any] | None = None,
|
config: dict[str, Any] | None = None,
|
||||||
metadata: dict[str, Any] | None = None,
|
metadata: dict[str, Any] | None = None,
|
||||||
langsmith_tracing: LangSmithTracing | None = None,
|
langsmith_tracing: LangSmithTracing | None = None,
|
||||||
|
context: dict[str, Any] | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""Send `run.start` to the server. Returns the result (`{"run_id": ...}`)."""
|
"""Send `run.start` to the server. Returns the result (`{"run_id": ...}`).
|
||||||
|
|
||||||
|
Args:
|
||||||
|
input: the run input; omitted from the wire payload when None.
|
||||||
|
config: the run config; omitted when None.
|
||||||
|
metadata: run metadata; omitted when None.
|
||||||
|
langsmith_tracing: tracing options; omitted when None.
|
||||||
|
context: per-run static context; omitted from the wire payload
|
||||||
|
when None (server applies its default context behavior).
|
||||||
|
"""
|
||||||
params: dict[str, Any] = {"assistant_id": self._owner.assistant_id}
|
params: dict[str, Any] = {"assistant_id": self._owner.assistant_id}
|
||||||
if input is not None:
|
if input is not None:
|
||||||
params["input"] = input
|
params["input"] = input
|
||||||
@@ -227,6 +237,8 @@ class SyncRunModule:
|
|||||||
params["metadata"] = metadata
|
params["metadata"] = metadata
|
||||||
if langsmith_tracing is not None:
|
if langsmith_tracing is not None:
|
||||||
params["langsmith_tracer"] = langsmith_tracing
|
params["langsmith_tracer"] = langsmith_tracing
|
||||||
|
if context is not None:
|
||||||
|
params["context"] = context
|
||||||
result = self._owner._send_command("run.start", params)
|
result = self._owner._send_command("run.start", params)
|
||||||
self._owner._run_seen = True
|
self._owner._run_seen = True
|
||||||
controller = self._owner._controller
|
controller = self._owner._controller
|
||||||
@@ -239,6 +251,7 @@ class SyncRunModule:
|
|||||||
response: Any,
|
response: Any,
|
||||||
*,
|
*,
|
||||||
interrupt_id: str | None = None,
|
interrupt_id: str | None = None,
|
||||||
|
context: dict[str, Any] | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""Reply to a server-side interrupt and resume the run.
|
"""Reply to a server-side interrupt and resume the run.
|
||||||
|
|
||||||
@@ -246,6 +259,8 @@ class SyncRunModule:
|
|||||||
response: the response value forwarded as `params.response` on the wire.
|
response: the response value forwarded as `params.response` on the wire.
|
||||||
interrupt_id: optional explicit id. When omitted, requires exactly one
|
interrupt_id: optional explicit id. When omitted, requires exactly one
|
||||||
outstanding interrupt.
|
outstanding interrupt.
|
||||||
|
context: optional per-run static context for the resumed run;
|
||||||
|
forwarded with the `input.respond` command when non-None.
|
||||||
|
|
||||||
Raises:
|
Raises:
|
||||||
RuntimeError: no outstanding interrupts; `interrupt_id` is None but
|
RuntimeError: no outstanding interrupts; `interrupt_id` is None but
|
||||||
@@ -282,6 +297,8 @@ class SyncRunModule:
|
|||||||
"namespace": match["namespace"],
|
"namespace": match["namespace"],
|
||||||
"response": response,
|
"response": response,
|
||||||
}
|
}
|
||||||
|
if context is not None:
|
||||||
|
params["context"] = context
|
||||||
return self._owner._send_command("input.respond", params)
|
return self._owner._send_command("input.respond", params)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -439,6 +439,64 @@ def test_sync_run_start_sends_command():
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_sync_run_start_forwards_context():
|
||||||
|
fake = SyncFakeServer()
|
||||||
|
fake.script([lifecycle_completed_event(seq=1)])
|
||||||
|
with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
|
||||||
|
threads = SyncThreadsClient(SyncHttpClient(raw))
|
||||||
|
with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
||||||
|
thread.run.start(input={"x": 1}, context={"user_id": "u-1"})
|
||||||
|
|
||||||
|
assert fake.received_commands[0]["params"]["context"] == {"user_id": "u-1"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_sync_run_start_omits_context_when_not_provided():
|
||||||
|
fake = SyncFakeServer()
|
||||||
|
fake.script([lifecycle_completed_event(seq=1)])
|
||||||
|
with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
|
||||||
|
threads = SyncThreadsClient(SyncHttpClient(raw))
|
||||||
|
with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
||||||
|
thread.run.start(input={"x": 1})
|
||||||
|
|
||||||
|
assert "context" not in fake.received_commands[0]["params"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_sync_run_respond_forwards_context():
|
||||||
|
fake = SyncFakeServer()
|
||||||
|
fake.script([lifecycle_completed_event(seq=1)])
|
||||||
|
with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
|
||||||
|
threads = SyncThreadsClient(SyncHttpClient(raw))
|
||||||
|
with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
||||||
|
thread.run.start(input={})
|
||||||
|
thread.interrupts.append(
|
||||||
|
{"interrupt_id": "i-1", "value": None, "namespace": []}
|
||||||
|
)
|
||||||
|
thread.interrupted = True
|
||||||
|
thread.run.respond("yes", context={"user_id": "u-1"})
|
||||||
|
|
||||||
|
command = fake.received_commands[-1]
|
||||||
|
assert command["method"] == "input.respond"
|
||||||
|
assert command["params"]["context"] == {"user_id": "u-1"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_sync_run_respond_omits_context_when_not_provided():
|
||||||
|
fake = SyncFakeServer()
|
||||||
|
fake.script([lifecycle_completed_event(seq=1)])
|
||||||
|
with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
|
||||||
|
threads = SyncThreadsClient(SyncHttpClient(raw))
|
||||||
|
with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
||||||
|
thread.run.start(input={})
|
||||||
|
thread.interrupts.append(
|
||||||
|
{"interrupt_id": "i-1", "value": None, "namespace": []}
|
||||||
|
)
|
||||||
|
thread.interrupted = True
|
||||||
|
thread.run.respond("yes")
|
||||||
|
|
||||||
|
command = fake.received_commands[-1]
|
||||||
|
assert command["method"] == "input.respond"
|
||||||
|
assert "context" not in command["params"]
|
||||||
|
|
||||||
|
|
||||||
def test_sync_events_iterates_raw_events():
|
def test_sync_events_iterates_raw_events():
|
||||||
|
|
||||||
fake = SyncFakeServer()
|
fake = SyncFakeServer()
|
||||||
|
|||||||
@@ -311,6 +311,28 @@ async def test_run_start_forwards_config_metadata_and_langsmith_tracing():
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
async def test_run_start_forwards_context():
|
||||||
|
fake = FakeServer()
|
||||||
|
transport = httpx.ASGITransport(app=fake.app)
|
||||||
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
|
||||||
|
threads = ThreadsClient(HttpClient(raw))
|
||||||
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
||||||
|
await thread.run.start(input={"x": 1}, context={"user_id": "u-1"})
|
||||||
|
params = fake.received_commands[0]["params"]
|
||||||
|
assert params["context"] == {"user_id": "u-1"}
|
||||||
|
|
||||||
|
|
||||||
|
async def test_run_start_omits_context_when_not_provided():
|
||||||
|
fake = FakeServer()
|
||||||
|
transport = httpx.ASGITransport(app=fake.app)
|
||||||
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
|
||||||
|
threads = ThreadsClient(HttpClient(raw))
|
||||||
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
||||||
|
await thread.run.start(input={"x": 1})
|
||||||
|
params = fake.received_commands[0]["params"]
|
||||||
|
assert "context" not in params
|
||||||
|
|
||||||
|
|
||||||
async def test_run_start_raises_outside_context_manager():
|
async def test_run_start_raises_outside_context_manager():
|
||||||
|
|
||||||
async with httpx.AsyncClient(base_url="http://test") as raw:
|
async with httpx.AsyncClient(base_url="http://test") as raw:
|
||||||
@@ -616,6 +638,38 @@ async def test_run_respond_dispatches_input_respond_command():
|
|||||||
assert command["params"]["namespace"] == []
|
assert command["params"]["namespace"] == []
|
||||||
|
|
||||||
|
|
||||||
|
async def test_run_respond_forwards_context():
|
||||||
|
fake = FakeServer()
|
||||||
|
asgi = httpx.ASGITransport(app=fake.app)
|
||||||
|
async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
|
||||||
|
threads = ThreadsClient(HttpClient(raw))
|
||||||
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
||||||
|
await thread.run.start(input={})
|
||||||
|
thread.interrupts.append(
|
||||||
|
{"interrupt_id": "i-1", "value": None, "namespace": []}
|
||||||
|
)
|
||||||
|
thread.interrupted = True
|
||||||
|
await thread.run.respond("yes", context={"user_id": "u-1"})
|
||||||
|
params = fake.received_commands[-1]["params"]
|
||||||
|
assert params["context"] == {"user_id": "u-1"}
|
||||||
|
|
||||||
|
|
||||||
|
async def test_run_respond_omits_context_when_not_provided():
|
||||||
|
fake = FakeServer()
|
||||||
|
asgi = httpx.ASGITransport(app=fake.app)
|
||||||
|
async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
|
||||||
|
threads = ThreadsClient(HttpClient(raw))
|
||||||
|
async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
|
||||||
|
await thread.run.start(input={})
|
||||||
|
thread.interrupts.append(
|
||||||
|
{"interrupt_id": "i-1", "value": None, "namespace": []}
|
||||||
|
)
|
||||||
|
thread.interrupted = True
|
||||||
|
await thread.run.respond("yes")
|
||||||
|
params = fake.received_commands[-1]["params"]
|
||||||
|
assert "context" not in params
|
||||||
|
|
||||||
|
|
||||||
async def test_run_respond_with_explicit_interrupt_id():
|
async def test_run_respond_with_explicit_interrupt_id():
|
||||||
fake = FakeServer()
|
fake = FakeServer()
|
||||||
asgi = httpx.ASGITransport(app=fake.app)
|
asgi = httpx.ASGITransport(app=fake.app)
|
||||||
|
|||||||
Reference in New Issue
Block a user