From 9ff778fe9baa9288214e5d8e9cfe6a0699e9ccaa Mon Sep 17 00:00:00 2001 From: George Pickett <297992784+georgeatparallel@users.noreply.github.com> Date: Tue, 20 Oct 2026 04:44:18 +0000 Subject: [PATCH] Add default anonymous web tools to ReCall inference --- README.md | 41 +++- requirements-inference.txt | 3 + setup.py | 4 +- src/re_call/inference/parallel_search.py | 95 ++++++++ src/re_call/inference/re_call.py | 40 +++- tests/test_parallel_search.py | 273 +++++++++++++++++++++++ 6 files changed, 450 insertions(+), 6 deletions(-) create mode 100644 requirements-inference.txt create mode 100644 src/re_call/inference/parallel_search.py create mode 100644 tests/test_parallel_search.py diff --git a/README.md b/README.md index 63e6047..4f7a45a 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,45 @@ bash train.sh \ If you want to **fully reproduce** *ReCall*, please refer to the multi-node training script in `scripts/train/train_multi_node.sh`. ### Inference +#### Web search without a sandbox + +With no executor or tool environment configured, the inference wrapper offers +`web_search` and `web_fetch` through [Parallel Search MCP](https://docs.parallel.ai/integrations/mcp/search-mcp). +Search uses Fast mode and needs no account or API key. A served ReCall model is +still required. For an inference-only installation in a Python 3.10 environment: + +```bash +pip install -r requirements-inference.txt +pip install --no-deps -e . +``` + +```python +from re_call import ReCall + +agent = ReCall(model_url="http://localhost:8000") +response = agent.run(question="What are the latest releases of Python? Read the release notes and cite source URLs.") +print(response) +``` + +The wrapper discovers the two tools over Streamable HTTP at +`https://search.parallel.ai/mcp`, includes their schemas in the model prompt, +and returns their results to the model in the existing tool loop. Each `run` +starts a new conversation; related search and fetch calls, including generation +retries, share one opaque identifier. The adapter sends no authentication and +does not read API keys from the environment. The free service has usage limits; +tool errors are returned to the model, and discovery errors stop the run. +`search_timeout=30` sets the deadline in seconds for each discovery or tool request. + +Set `ReCall(model_url, search_enabled=False)` to disable the built-in web tools, +or pass `func_schemas=[]` to expose no tools for one run. The API is synchronous; +from async code, use `await asyncio.to_thread(agent.run, question=question)`. +Supplying an executor URL, environment and schemas keeps the existing sandbox +path below. Explicit sandbox configurations never fall back to web search. +This integration applies to inference; the training and evaluation retrievers +continue to use their configured environments. + +#### Custom sandbox tools + This section demonstrates how to perform inference using the trained *ReCall* model. We provide a standard wrapper class in `src/re_call/inference/re_call.py` that simplifies the inference process. To get started, you only need to provide the model URL and sandbox URL, then use the `run` function to execute inference. The `ReCall` class handles all the orchestration between model generation and tool execution internally. For a practical example of using the `ReCall` class, please refer to our sample implementation at `scripts/inference/re_call_use_case.py`. For model serving, we recommend using [SGLang](https://docs.sglang.ai/). You can either download our open-source models or train your own models to conduct the inference. Here is an example of how to launch the model service: @@ -175,4 +214,4 @@ If you find this work useful, please cite it as follows: primaryClass={cs.AI}, url={https://arxiv.org/abs/2503.19470}, } -``` \ No newline at end of file +``` diff --git a/requirements-inference.txt b/requirements-inference.txt new file mode 100644 index 0000000..f400f00 --- /dev/null +++ b/requirements-inference.txt @@ -0,0 +1,3 @@ +# The inference wrapper does not require the training or GPU dependencies. +requests +mcp>=1.26,<2 diff --git a/setup.py b/setup.py index eeca547..caa90bf 100644 --- a/setup.py +++ b/setup.py @@ -54,6 +54,8 @@ 'jieba', # others + 'requests', + 'mcp>=1.26,<2', 'sglang', 'jsonlines', ] @@ -73,4 +75,4 @@ include_package_data=True, long_description=long_description, long_description_content_type='text/markdown' -) \ No newline at end of file +) diff --git a/src/re_call/inference/parallel_search.py b/src/re_call/inference/parallel_search.py new file mode 100644 index 0000000..915bb23 --- /dev/null +++ b/src/re_call/inference/parallel_search.py @@ -0,0 +1,95 @@ +"""Anonymous web tools for a single ReCall inference run.""" + +import asyncio +import copy +import json +import math +from datetime import timedelta + +import anyio +import httpx +from mcp import ClientSession +from mcp.client.streamable_http import streamable_http_client + + +ENDPOINT = "https://search.parallel.ai/mcp" +USER_AGENT = "ReCall (+https://github.com/Agent-RL/ReCall)" +TOOL_NAMES = frozenset({"web_search", "web_fetch"}) + + +class ParallelSearch: + """Stateless sync adapter; conversation identity is supplied by ReCall.run.""" + + def __init__(self, timeout=30): + if not math.isfinite(timeout) or timeout <= 0: + raise ValueError("search_timeout must be finite and positive") + self.timeout = timeout + + async def _request(self, name=None, arguments=None): + # Bound discovery and tool execution, including transport setup/cleanup. + with anyio.fail_after(self.timeout): + async with httpx.AsyncClient( + headers={"User-Agent": USER_AGENT}, timeout=self.timeout, + ) as client: + async with streamable_http_client(ENDPOINT, http_client=client) as streams: + async with ClientSession( + streams[0], streams[1], + read_timeout_seconds=timedelta(seconds=self.timeout), + ) as session: + await session.initialize() + if name is None: + return await session.list_tools() + return await session.call_tool(name, arguments) + + def schemas(self): + """Discover model-visible schemas without exposing host-owned metadata.""" + tools = asyncio.run(self._request()).tools + schemas = [] + for tool in tools: + if tool.name not in TOOL_NAMES: + continue + parameters = copy.deepcopy(tool.inputSchema) + for key in ("session_id", "model_name"): + parameters.get("properties", {}).pop(key, None) + if key in parameters.get("required", []): + parameters["required"].remove(key) + schemas.append({ + "name": tool.name, "description": tool.description or tool.name, + "parameters": parameters, + }) + if {tool["name"] for tool in schemas} != TOOL_NAMES: + raise RuntimeError("Parallel Search MCP did not provide web_search and web_fetch") + return json.dumps(schemas) + + def execute(self, tool_calls, session_id): + results = [] + for raw_call in tool_calls: + try: + call = json.loads(raw_call) + name = call["name"] + arguments = call.get("arguments", {}) + if name not in TOOL_NAMES or not isinstance(arguments, dict): + raise ValueError("only web_search and web_fetch with object arguments are allowed") + arguments = dict(arguments) + # The model cannot replace the run's identity or invent its model ID. + arguments.pop("model_name", None) + arguments["session_id"] = session_id + result = asyncio.run(self._request(name, arguments)) + text = "\n".join( + block.text for block in result.content if block.type == "text" + ) + if not text and result.structuredContent is not None: + text = json.dumps(result.structuredContent) + results.append(("error: " if result.isError else "result: \n") + text) + except Exception as error: + # Cancellation/KeyboardInterrupt propagate; ordinary failures become + # observations, never a silent switch to a different provider. + results.append("error: Parallel Search MCP: " + self._error_text(error)) + return results + + @staticmethod + def _error_text(error): + children = getattr(error, "exceptions", None) + if children: + return "; ".join(ParallelSearch._error_text(child) for child in children) + return str(error) or type(error).__name__ diff --git a/src/re_call/inference/re_call.py b/src/re_call/inference/re_call.py index 09c3169..b2675a1 100644 --- a/src/re_call/inference/re_call.py +++ b/src/re_call/inference/re_call.py @@ -2,6 +2,7 @@ import json import requests import time +import uuid from typing import List from functools import wraps @@ -40,9 +41,11 @@ class ReCall(): {{"name": , "arguments": }} """ - def __init__(self, model_url, executor_url): + def __init__(self, model_url, executor_url=None, *, search_enabled=True, search_timeout=30): self.model_url = model_url self.executor_url = executor_url + self.search_enabled = search_enabled + self.search_timeout = search_timeout def init_prompt(self, func_schemas, question): system_prompt = f"<|im_start|>system\n{self.system_prompt.format(func_schemas=func_schemas)}<|im_end|>" @@ -137,8 +140,37 @@ def extract_tool_calls(self, output_str): except Exception as e: return [] + def run(self, env=None, func_schemas=None, question=None): + """Run inference with explicit sandbox tools or unconfigured web tools. + + Each call starts an independent conversation. This is a synchronous API; + async applications should call it in a worker thread. + """ + if question is None: + raise ValueError("question is required") + if env is None and func_schemas is None and self.executor_url is None: + if self.search_enabled: + from .parallel_search import ParallelSearch + + search = ParallelSearch(timeout=self.search_timeout) + session_id = str(uuid.uuid4()) + func_schemas = search.schemas() + execute = lambda calls: search.execute(calls, session_id) + else: + func_schemas = "[]" + execute = lambda calls: ["error: web tools are disabled"] * len(calls) + elif env is None and func_schemas in ([], "[]") and self.executor_url is None: + execute = lambda calls: ["error: no tools are configured"] * len(calls) + else: + # A partially configured sandbox is never an invitation to use the web. + if env is None or func_schemas is None or self.executor_url is None: + raise ValueError("sandbox tools require env, func_schemas and executor_url") + execute = lambda calls: self.execute_tool_calls(env, calls) + # Keep the ID and selected executor across the existing generation retry. + return self._run(func_schemas, question, execute) + @retry(max=5, sleep=1) - def run(self, env, func_schemas, question): + def _run(self, func_schemas, question, execute): curr_prompt = self.init_prompt(func_schemas, question) for _ in range(5): response = requests.post( @@ -157,7 +189,7 @@ def run(self, env, func_schemas, question): if len(tool_calls) == 0: break - results: List[str] = self.execute_tool_calls(env, tool_calls) + results: List[str] = execute(tool_calls) curr_prompt = self.cat_tool_results(curr_prompt, tool_calls, results) - return curr_prompt \ No newline at end of file + return curr_prompt diff --git a/tests/test_parallel_search.py b/tests/test_parallel_search.py new file mode 100644 index 0000000..8bc8d35 --- /dev/null +++ b/tests/test_parallel_search.py @@ -0,0 +1,273 @@ +"""Exercise the inference loop with a scripted model and real MCP SDK transport.""" + +import asyncio +import json +import os +import unittest +from concurrent.futures import ThreadPoolExecutor +from unittest.mock import patch + +import httpx +import requests + +from re_call import ReCall +from re_call.inference import parallel_search + + +ASYNC_CLIENT = httpx.AsyncClient +SOURCE_URL = "https://www.python.org/downloads/" + + +def tool_call(name, **arguments): + return '' + json.dumps({"name": name, "arguments": arguments}) + '' + + +class ModelResponse: + def __init__(self, text): + self.text = text + + def json(self): + return {"text": self.text} + + +class MCPFixture: + """Only HTTP is replaced: SDK initialize/list/call and parsing all execute.""" + + def __init__(self): + self.requests = [] + self.failure = None + self.clients = [] + + def client(self, **kwargs): + client = ASYNC_CLIENT(transport=httpx.MockTransport(self.handle), **kwargs) + self.clients.append(client) + return client + + async def handle(self, request): + message = json.loads(request.content) + self.requests.append((request, message)) + method = message["method"] + if method == "notifications/initialized": + return httpx.Response(202) + if method == "initialize": + result = { + "protocolVersion": message["params"]["protocolVersion"], + "capabilities": {"tools": {}}, + "serverInfo": {"name": "fixture", "version": "1.0"}, + } + elif method == "tools/list": + result = {"tools": [ + {"name": name, "description": name, "inputSchema": { + "type": "object", "properties": { + "session_id": {"type": "string"}, + "model_name": {"type": "string"}, + field: {"type": "array", "items": {"type": "string"}}, + }, "required": [field], + }} + for name, field in [("web_search", "search_queries"), ("web_fetch", "urls")] + ]} + elif method == "tools/call": + if self.failure == "timeout": + raise httpx.ReadTimeout("fixture read deadline", request=request) + if self.failure == "hang": + await asyncio.sleep(10) + result = {"content": [{"type": "text", "text": json.dumps({ + "url": SOURCE_URL, "excerpts": ["Python release notes"], + })}], "isError": False} + if self.failure == "quota": + result = {"content": [{"type": "text", "text": "anonymous quota exceeded"}], "isError": True} + elif self.failure == "malformed": + return httpx.Response(200, text="not JSON", headers={"content-type": "application/json"}) + else: + raise AssertionError(method) + return httpx.Response(200, json={"jsonrpc": "2.0", "id": message["id"], "result": result}) + + def calls(self): + return [msg["params"] for _, msg in self.requests if msg["method"] == "tools/call"] + + +class InferenceSearchTests(unittest.TestCase): + def setUp(self): + self.fixture = MCPFixture() + self.addCleanup(patch.stopall) + patch.object(parallel_search.httpx, "AsyncClient", side_effect=self.fixture.client).start() + patch.dict(os.environ, {"HOME": "/nonexistent-recall-test-home"}, clear=True).start() + + def scripted_model(self, responses): + self.prompts = [] + iterator = iter(responses) + + def generate(url, json, **kwargs): + self.assertEqual(url, "http://model/generate") + self.prompts.append(json["text"]) + response = next(iterator) + if isinstance(response, BaseException): + raise response + return ModelResponse(response) + + return patch("re_call.inference.re_call.requests.post", side_effect=generate) + + def test_default_search_fetch_followup_are_consumed_with_one_run_id(self): + with self.scripted_model([ + tool_call("web_search", search_queries=["Python recent release notes"], session_id="model-chosen", model_name="invented"), + tool_call("web_fetch", urls=[SOURCE_URL]), + tool_call("web_search", search_queries=["Python release security changes"]), + "The source is " + SOURCE_URL, + ]): + result = ReCall("http://model").run(question="Find Python releases") + self.assertIn("Python release notes", self.prompts[1]) + self.assertIn(SOURCE_URL, self.prompts[2]) + self.assertIn("The source is " + SOURCE_URL, result) + schemas = self.prompts[0].split('```json\n', 1)[1].split('\n```', 1)[0] + for schema in json.loads(schemas): + self.assertNotIn("session_id", schema["parameters"]["properties"]) + self.assertNotIn("model_name", schema["parameters"]["properties"]) + calls = self.fixture.calls() + ids = {call["arguments"]["session_id"] for call in calls} + self.assertEqual(len(ids), 1) + self.assertNotIn("model-chosen", ids) + self.assertEqual(len(next(iter(ids))), 36) + self.assertEqual(calls[1]["arguments"]["urls"], [SOURCE_URL]) + for request, _ in self.fixture.requests: + self.assertEqual(str(request.url), parallel_search.ENDPOINT) + self.assertEqual(request.headers["user-agent"], parallel_search.USER_AGENT) + self.assertNotIn("authorization", request.headers) + self.assertTrue(all("model_name" not in call["arguments"] for call in calls)) + self.assertTrue(all(client.is_closed for client in self.fixture.clients)) + + def test_new_runs_on_same_agent_start_new_conversations(self): + agent = ReCall("http://model") + for _ in range(2): + with self.scripted_model([tool_call("web_search", search_queries=["Python release notes"]), "done"]): + agent.run(question="Find releases") + self.assertEqual(len({c["arguments"]["session_id"] for c in self.fixture.calls()}), 2) + + def test_interleaved_runs_do_not_share_conversation_identity(self): + agent = ReCall("http://model") + + def generate(url, json, **kwargs): + prompt = json["text"] + question = "task-A" if "task-A" in prompt else "task-B" + count = prompt.count("") + if count == 0: + return ModelResponse(tool_call("web_search", search_queries=[question])) + if count == 1: + return ModelResponse(tool_call("web_fetch", urls=[SOURCE_URL], objective=question)) + return ModelResponse("done") + + with patch("re_call.inference.re_call.requests.post", side_effect=generate): + with ThreadPoolExecutor(max_workers=2) as pool: + results = list(pool.map(lambda q: agent.run(question=q), ["task-A", "task-B"])) + ids = {} + for call in self.fixture.calls(): + args = call["arguments"] + task = args.get("objective") or args["search_queries"][0] + ids.setdefault(task, set()).add(args["session_id"]) + self.assertEqual([len(ids[k]) for k in sorted(ids)], [1, 1]) + self.assertNotEqual(ids["task-A"], ids["task-B"]) + self.assertTrue(all("done" in r for r in results)) + + def test_generation_retry_keeps_identity(self): + search = tool_call("web_search", search_queries=["Python release notes"]) + with self.scripted_model([search, requests.ConnectionError("model reset"), search, "done"]): + with patch("re_call.inference.re_call.time.sleep"): + ReCall("http://model").run(question="Find releases") + self.assertEqual(len(self.fixture.calls()), 2) + self.assertEqual(len({c["arguments"]["session_id"] for c in self.fixture.calls()}), 1) + + def test_disabled_and_explicit_empty_schemas_cannot_execute_web_tools(self): + for kwargs, run_kwargs in [({"search_enabled": False}, {}), ({}, {"func_schemas": []}), ({}, {"func_schemas": "[]"})]: + with self.subTest(kwargs=kwargs, run_kwargs=run_kwargs): + with self.scripted_model([tool_call("web_search", search_queries=["should not run"]), "done"]): + result = ReCall("http://model", **kwargs).run(question="Offline", **run_kwargs) + self.assertIn("error:", result) + self.assertEqual(self.fixture.requests, []) + + def test_explicit_sandbox_preserved_even_on_failure(self): + for status in [200, 503]: + seen = [] + + def post(url, json, **kwargs): + seen.append((url, json)) + if url == "http://sandbox/execute": + response = requests.Response() + response.status_code = status + response._content = b'{"result":"custom result","output":"","error":""}' + return response + return ModelResponse(tool_call("custom", query="Python") if len(seen) == 1 else "done") + + with patch("re_call.inference.re_call.requests.post", side_effect=post): + result = ReCall("http://model", "http://sandbox").run("custom env", "custom schemas", "Question") + self.assertEqual(seen[1], ("http://sandbox/execute", {"env": "custom env", "call": "custom(query='Python')"})) + self.assertIn("custom result" if status == 200 else "error: 503", result) + self.assertEqual(self.fixture.requests, []) + + def test_partial_sandbox_never_falls_back(self): + for agent, kwargs in [ + (ReCall("http://model", "http://sandbox"), {}), + (ReCall("http://model"), {"env": "explicit"}), + (ReCall("http://model"), {"func_schemas": "custom"}), + ]: + with self.assertRaises(ValueError): + agent.run(question="Question", **kwargs) + self.assertEqual(self.fixture.requests, []) + + def test_unknown_and_malformed_calls_denied_before_http(self): + with self.scripted_model([ + tool_call("delete_files", path="/"), + '{"name":"web_search","arguments":[]}', + 'bad json', "done", + ]): + result = ReCall("http://model").run(question="Question") + self.assertIn("error: Parallel Search MCP", result) + self.assertEqual(self.fixture.calls(), []) + + def test_errors_and_timeouts_are_model_observations_without_fallback(self): + for failure in ["quota", "timeout", "malformed", "hang"]: + with self.subTest(failure=failure): + self.fixture.failure = failure + with self.scripted_model([ + tool_call("web_search", search_queries=["Python release notes"]), + tool_call("web_fetch", urls=[SOURCE_URL]), "done", + ]): + result = ReCall("http://model", search_timeout=0.15).run(question="Question") + self.assertIn("error:", self.prompts[1]) + self.assertIn("done", result) + self.assertEqual(len({c["arguments"]["session_id"] for c in self.fixture.calls()[-2:]}), 1) + self.assertTrue(all(client.is_closed for client in self.fixture.clients)) + + def test_discovery_failure_stops_before_model_execution(self): + with patch.object(self.fixture, "handle", side_effect=httpx.ConnectError("unavailable")): + with patch("re_call.inference.re_call.requests.post") as model: + with self.assertRaises(Exception): + ReCall("http://model", search_timeout=0.15).run(question="Question") + model.assert_not_called() + + def test_cancellation_propagates_and_closes_sdk_transport(self): + self.fixture.failure = "hang" + run_async = asyncio.run + + async def cancel_request(request): + task = asyncio.create_task(request) + while not self.fixture.calls(): + await asyncio.sleep(0.001) + task.cancel() + return await task + + def run_with_cancellation(request): + # Discovery completes normally. Cancel the active tool request at + # the sync/async boundary used by the actual ReCall.run dispatcher. + if any(msg["method"] == "tools/list" for _, msg in self.fixture.requests): + return run_async(cancel_request(request)) + return run_async(request) + + with self.scripted_model([tool_call("web_search", search_queries=["Python release notes"])]): + with patch.object(parallel_search.asyncio, "run", side_effect=run_with_cancellation): + with self.assertRaises(asyncio.CancelledError): + ReCall("http://model", search_timeout=2).run(question="Question") + self.assertEqual(len(self.prompts), 1) + self.assertTrue(all(client.is_closed for client in self.fixture.clients)) + + +if __name__ == "__main__": + unittest.main()